sync_userprefs_db: fix merge function

[pmg-api.git] / PMG / Cluster.pm

diff --git a/PMG/Cluster.pm b/PMG/Cluster.pm
index e99b2c83412c65350ed1e18491be7fbaac437e0a..5eb41b5046ae001aa2177086680f4f857c8a58e7 100644 (file)
--- a/PMG/Cluster.pm
+++ b/PMG/Cluster.pm
@@ -10,6 +10,7 @@ use Time::HiRes qw (gettimeofday tv_interval);
 use PVE::SafeSyslog;
 use PVE::Tools;
 use PVE::INotify;
+use PVE::APIClient::LWP;
 
 use PMG::Utils;
 use PMG::Config;
@@ -17,7 +18,7 @@ use PMG::ClusterConfig;
 use PMG::RuleDB;
 use PMG::RuleCache;
 use PMG::MailQueue;
-use PVE::APIClient::LWP;
+use PMG::Fetchmail;
 
 sub remote_node_ip {
     my ($nodename, $noerr) = @_;
@@ -195,15 +196,22 @@ my $ssh_rsa_id = "/root/.ssh/id_rsa.pub";
 sub update_ssh_keys {
     my ($cinfo) = @_;
 
+    my $old = '';
     my $data = '';
+
     foreach my $node (values %{$cinfo->{ids}}) {
        $data .= "$node->{ip} ssh-rsa $node->{hostrsapubkey}\n";
        $data .= "$node->{name} ssh-rsa $node->{hostrsapubkey}\n";
     }
 
-    PVE::Tools::file_set_contents($sshglobalknownhosts, $data);
+    $old = PVE::Tools::file_get_contents($sshglobalknownhosts, 1024*1024)
+       if -f $sshglobalknownhosts;
+
+    PVE::Tools::file_set_contents($sshglobalknownhosts, $data)
+       if $old ne $data;
 
     $data = '';
+    $old = '';
 
     # always add ourself
     if (-f $ssh_rsa_id) {
@@ -232,7 +240,11 @@ sub update_ssh_keys {
        $newdata .= "$line\n";
     }
 
-    PVE::Tools::file_set_contents($rootsshauthkeys, $newdata, 0600);
+    $old = PVE::Tools::file_get_contents($rootsshauthkeys, 1024*1024)
+       if -f $rootsshauthkeys;
+
+    PVE::Tools::file_set_contents($rootsshauthkeys, $newdata, 0600)
+       if $old ne $newdata;
 }
 
 my $cfgdir = '/etc/pmg';
@@ -256,6 +268,9 @@ my $cond_commit_synced_file = sub {
        return 0 if $new eq $old;
     }
 
+    # set mtime (touch) to avoid time drift problems
+    utime(undef, undef, $srcfn);
+
     rename($srcfn, $dstfn) ||
        die "cond_rename_file '$filename' failed - $!\n";
 
@@ -276,10 +291,14 @@ my $rsync_command = sub {
 };
 
 sub sync_quarantine_files {
-    my ($host_ip, $host_name, $flistname) = @_;
+    my ($host_ip, $host_name, $flistname, $rcid) = @_;
 
     my $spooldir = $PMG::MailQueue::spooldir;
 
+    mkdir "$spooldir/cluster/";
+    my $syncdir = "$spooldir/cluster/$rcid";
+    mkdir $syncdir;
+
     my $cmd = $rsync_command->(
        $host_name, '--timeout', '10', "${host_ip}:$spooldir", $spooldir,
        '--files-from', $flistname);
@@ -364,6 +383,10 @@ sub sync_config_from_master {
 
     $cond_commit_synced_file->('cluster.conf');
 
+    update_ssh_keys($cinfo); # rewrite ssh keys
+
+    PMG::Fetchmail::update_fetchmail_default(0); # disable on slave
+
     my $files = [
        'pmg-authkey.key',
        'pmg-authkey.pub',
@@ -373,12 +396,15 @@ sub sync_config_from_master {
        'domains',
        'mynetworks',
        'transport',
+       'tls_policy',
+       'fetchmailrc',
        ];
 
     foreach my $filename (@$files) {
        $cond_commit_synced_file->($filename);
     }
 
+
     my $force_restart = {};
 
     if ($cond_commit_synced_file->($sa_custom_cf, "${sa_conf_dir}/${sa_custom_cf}")) {
@@ -520,7 +546,10 @@ sub sync_quarantine_db {
            my $callback = sub {
                my $ref = shift;
                $maxid = $ref->{rid};
-               print $flistfh "$ref->{file}\n";
+               my $filename = $ref->{file};
+                # skip files generated before cluster was created
+               return if $filename !~ m!^cluster/!;
+               print $flistfh "$filename\n";
            };
 
            my $attrs = [qw(cid rid time qtype bytes spamlevel info sender header file)];
@@ -529,7 +558,7 @@ sub sync_quarantine_db {
            close($flistfh);
 
            my $starttime = [ gettimeofday() ];
-           sync_quarantine_files($ni->{ip}, $ni->{name}, $flistname);
+           sync_quarantine_files($ni->{ip}, $ni->{name}, $flistname, $rcid);
            $$rsynctime_ref += tv_interval($starttime);
 
            if ($maxid) {
@@ -541,7 +570,7 @@ sub sync_quarantine_db {
                    "AND CMailStore_RID <= ?");
                $sth->execute($rcid, $lastid, $maxid);
 
-               $attrs = [qw(cmailstore_cid cmailstore_rid pmail receiver status mtime)];
+               $attrs = [qw(cmailstore_cid cmailstore_rid pmail receiver ticketid status mtime)];
                PMG::DBTools::copy_selected_data($ldb, $sth, 'CMSReceivers', $attrs);
 
                PMG::DBTools::write_maxint_clusterinfo($ldb, $rcid, 'lastid_CMailStore', $maxid);
@@ -576,10 +605,10 @@ sub sync_quarantine_db {
 
        my $update_sth = $ldb->prepare(
            "UPDATE CMSReceivers SET status = ? WHERE " .
-           "CMailstore_CID = ? AND CMailstore_RID = ? AND PMail = ?;");
+           "CMailstore_CID = ? AND CMailstore_RID = ? AND TicketID = ?");
        while (my $ref = $sth->fetchrow_hashref()) {
            $update_sth->execute($ref->{status}, $ref->{cmailstore_cid},
-                                $ref->{cmailstore_rid}, $ref->{pmail});
+                                $ref->{cmailstore_rid}, $ref->{ticketid});
        }
 
        PMG::DBTools::write_maxint_clusterinfo($ldb, $rcid, 'lastmt_CMSReceivers', $ctime);
@@ -647,10 +676,10 @@ sub sync_statistic_db {
 
                $attrs = [qw(cstatistic_cid cstatistic_rid blocked receiver)];
                PMG::DBTools::copy_selected_data($ldb, $sth, 'CReceivers', $attrs);
-
-               PMG::DBTools::write_maxint_clusterinfo ($ldb, $rcid, 'lastid_CStatistic', $maxid);
            }
 
+           PMG::DBTools::write_maxint_clusterinfo ($ldb, $rcid, 'lastid_CStatistic', $maxid);
+
            $ldb->commit;
        };
        if (my $err = $@) {
@@ -710,6 +739,31 @@ my $sync_generic_mtime_db = sub {
     return $updates;
 };
 
+sub sync_localstat_db {
+    my ($dbh, $rdb, $ni) = @_;
+
+    my $rcid = $ni->{cid};
+
+    my $selectfunc = sub {
+       my ($ctime, $lastmt) = @_;
+       return "SELECT * from LocalStat WHERE mtime >= $lastmt AND cid = $rcid";
+    };
+
+    my $merge_sth = $dbh->prepare(
+       'INSERT INTO LocalStat (Time, RBLCount, PregreetCount, CID, MTime) ' .
+       'VALUES (?, ?, ?, ?, ?) ' .
+       'ON CONFLICT (Time, CID) DO UPDATE SET ' .
+       'RBLCount = excluded.RBLCount, PregreetCount = excluded.PregreetCount, MTime = excluded.MTime');
+
+    my $mergefunc = sub {
+       my ($ref) = @_;
+
+       $merge_sth->execute($ref->{time}, $ref->{rblcount}, $ref->{pregreetcount}, $ref->{cid}, $ref->{mtime});
+    };
+
+    return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'LocalStat', $selectfunc, $mergefunc);
+}
+
 sub sync_greylist_db {
     my ($dbh, $rdb, $ni) = @_;
 
@@ -743,7 +797,7 @@ sub sync_userprefs_db {
 
     my $merge_sth = $dbh->prepare(
        "INSERT INTO UserPrefs (PMail, Name, Data, MTime) " .
-       'VALUES (?, ?, ?, 0) ' .
+       'VALUES (?, ?, ?, ?) ' .
        'ON CONFLICT (PMail, Name) DO UPDATE SET ' .
        # Note: MTime = 0 ==> this is just a copy from somewhere else, not modified
        'MTime = CASE WHEN excluded.MTime >= UserPrefs.MTime THEN 0 ELSE UserPrefs.MTime END, ' .
@@ -752,7 +806,7 @@ sub sync_userprefs_db {
     my $mergefunc = sub {
        my ($ref) = @_;
 
-       $merge_sth->execute($ref->{pmail}, $ref->{name}, $ref->{data});
+       $merge_sth->execute($ref->{pmail}, $ref->{name}, $ref->{data}, $ref->{mtime});
     };
 
     return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'UserPrefs', $selectfunc, $mergefunc);