8 use Time
::HiRes qw
(gettimeofday tv_interval
);
13 use PVE
::APIClient
::LWP
;
18 use PMG
::ClusterConfig
;
26 my ($nodename, $noerr) = @_;
28 my $cinfo = PMG
::ClusterConfig-
>new();
30 foreach my $entry (values %{$cinfo->{ids
}}) {
31 if ($entry->{name
} eq $nodename) {
32 my $ip = $entry->{ip
};
33 return $ip if !wantarray;
34 my $family = PVE
::Tools
::get_host_address_family
($ip);
35 return ($ip, $family);
39 # fallback: try to get IP by other means
40 if ($nodename eq 'localhost' || $nodename eq PVE
::INotify
::nodename
()) {
41 return PVE
::Network
::get_local_ip
();
43 return PVE
::Network
::get_ip_from_hostname
($nodename, $noerr);
50 $cinfo = PMG
::ClusterConfig-
>new() if !$cinfo;
52 return $cinfo->{master
}->{name
} if defined($cinfo->{master
});
57 sub read_local_ssl_cert_fingerprint
{
58 my $cert_path = "/etc/pmg/pmg-api.pem";
62 my $bio = Net
::SSLeay
::BIO_new_file
($cert_path, 'r');
63 $cert = Net
::SSLeay
::PEM_read_bio_X509
($bio);
64 Net
::SSLeay
::BIO_free
($bio);
67 die "unable to read certificate '$cert_path' - $err\n";
70 if (!defined($cert)) {
71 die "unable to read certificate '$cert_path' - got empty value\n";
76 $fp = Net
::SSLeay
::X509_get_fingerprint
($cert, 'sha256');
79 die "unable to get fingerprint for '$cert_path' - $err\n";
82 if (!defined($fp) || $fp eq '') {
83 die "unable to get fingerprint for '$cert_path' - got empty value\n";
89 my $hostrsapubkey_fn = '/etc/ssh/ssh_host_rsa_key.pub';
90 my $rootrsakey_fn = '/root/.ssh/id_rsa';
91 my $rootrsapubkey_fn = '/root/.ssh/id_rsa.pub';
93 sub read_local_cluster_info
{
97 my $hostrsapubkey = PVE
::Tools
::file_read_firstline
($hostrsapubkey_fn);
98 $hostrsapubkey =~ s/^.*ssh-rsa\s+//i;
99 $hostrsapubkey =~ s/\s+root\@\S+\s*$//i;
101 my $sshpubkeypattern = PMG
::ClusterConfig
::Node
::valid_ssh_pubkey_regex
();
102 die "unable to parse ${hostrsapubkey_fn}\n"
103 if $hostrsapubkey !~ m/$sshpubkeypattern/;
105 my $nodename = PVE
::INotify
::nodename
();
107 $res->{name
} = $nodename;
109 $res->{ip
} = PVE
::Network
::get_local_ip
();
111 $res->{hostrsapubkey
} = $hostrsapubkey;
113 if (! -f
$rootrsapubkey_fn) {
114 unlink $rootrsakey_fn;
115 my $cmd = ['ssh-keygen', '-t', 'rsa', '-N', '', '-b', '2048',
116 '-f', $rootrsakey_fn];
117 PMG
::Utils
::run_silent_cmd
($cmd);
120 my $rootrsapubkey = PVE
::Tools
::file_read_firstline
($rootrsapubkey_fn);
121 $rootrsapubkey =~ s/^.*ssh-rsa\s+//i;
122 $rootrsapubkey =~ s/\s+root\@\S+\s*$//i;
124 die "unable to parse ${rootrsapubkey_fn}\n"
125 if $rootrsapubkey !~ m/$sshpubkeypattern/;
127 $res->{rootrsapubkey
} = $rootrsapubkey;
129 $res->{fingerprint
} = read_local_ssl_cert_fingerprint
();
134 # X509 Certificate cache helper
136 my $cert_cache_nodes = {};
137 my $cert_cache_timestamp = time();
138 my $cert_cache_fingerprints = {};
140 sub update_cert_cache
{
142 $cert_cache_timestamp = time();
144 $cert_cache_fingerprints = {};
145 $cert_cache_nodes = {};
147 my $cinfo = PMG
::ClusterConfig-
>new();
149 foreach my $entry (values %{$cinfo->{ids
}}) {
150 my $node = $entry->{name
};
151 my $fp = $entry->{fingerprint
};
153 $cert_cache_fingerprints->{$fp} = 1;
154 $cert_cache_nodes->{$node} = $fp;
159 # load and cache cert fingerprint once
160 sub initialize_cert_cache
{
164 if defined($node) && !defined($cert_cache_nodes->{$node});
167 sub check_cert_fingerprint
{
170 # clear cache every 30 minutes at least
171 update_cert_cache
() if time() - $cert_cache_timestamp >= 60*30;
173 # get fingerprint of server certificate
176 $fp = Net
::SSLeay
::X509_get_fingerprint
($cert, 'sha256');
178 return 0 if $@ || !defined($fp) || $fp eq ''; # error
181 for my $expected (keys %$cert_cache_fingerprints) {
182 return 1 if $fp eq $expected;
187 return 1 if $check->();
189 # clear cache and retry at most once every minute
190 if (time() - $cert_cache_timestamp >= 60) {
191 syslog
('info', "Could not verify remote node certificate '$fp' with list of pinned certificates, refreshing cache");
199 my $sshglobalknownhosts = "/etc/ssh/ssh_known_hosts2";
200 my $rootsshauthkeys = "/root/.ssh/authorized_keys";
201 my $ssh_rsa_id = "/root/.ssh/id_rsa.pub";
203 sub update_ssh_keys
{
209 foreach my $node (values %{$cinfo->{ids
}}) {
210 $data .= "$node->{ip} ssh-rsa $node->{hostrsapubkey}\n";
211 $data .= "$node->{name} ssh-rsa $node->{hostrsapubkey}\n";
214 $old = PVE
::Tools
::file_get_contents
($sshglobalknownhosts, 1024*1024)
215 if -f
$sshglobalknownhosts;
217 PVE
::Tools
::file_set_contents
($sshglobalknownhosts, $data)
224 if (-f
$ssh_rsa_id) {
225 my $pub = PVE
::Tools
::file_get_contents
($ssh_rsa_id);
230 foreach my $node (values %{$cinfo->{ids
}}) {
231 $data .= "ssh-rsa $node->{rootrsapubkey} root\@$node->{name}\n";
234 if (-f
$rootsshauthkeys) {
235 my $mykey = PVE
::Tools
::file_get_contents
($rootsshauthkeys, 128*1024);
242 my @lines = split(/\n/, $data);
243 foreach my $line (@lines) {
244 if ($line !~ /^#/ && $line =~ m/(^|\s)ssh-(rsa|dsa)\s+(\S+)\s+\S+$/) {
245 next if $vhash->{$3}++;
247 $newdata .= "$line\n";
250 $old = PVE
::Tools
::file_get_contents
($rootsshauthkeys, 1024*1024)
251 if -f
$rootsshauthkeys;
253 PVE
::Tools
::file_set_contents
($rootsshauthkeys, $newdata, 0600)
257 my $cfgdir = '/etc/pmg';
258 my $syncdir = "$cfgdir/master";
260 my $cond_commit_synced_file = sub {
261 my ($filename, $dstfn) = @_;
263 $dstfn = "$cfgdir/$filename" if !defined($dstfn);
264 my $srcfn = "$syncdir/$filename";
271 my $new = PVE
::Tools
::file_get_contents
($srcfn, 1024*1024);
274 my $old = PVE
::Tools
::file_get_contents
($dstfn, 1024*1024);
275 return 0 if $new eq $old;
278 # set mtime (touch) to avoid time drift problems
279 utime(undef, undef, $srcfn);
281 rename($srcfn, $dstfn) ||
282 die "cond_rename_file '$filename' failed - $!\n";
284 print STDERR
"updated $dstfn\n";
289 my $ssh_command = sub {
290 my ($host_key_alias, @args) = @_;
292 my $cmd = ['ssh', '-l', 'root', '-o', 'BatchMode=yes'];
293 push @$cmd, '-o', "HostKeyAlias=${host_key_alias}" if $host_key_alias;
294 push @$cmd, @args if @args;
298 sub get_remote_cert_fingerprint
{
301 my $ssh_cmd = $ssh_command->(
304 'openssl x509 -noout -fingerprint -sha256 -in /etc/pmg/pmg-api.pem'
308 PVE
::Tools
::run_command
($ssh_cmd, outfunc
=> sub {
310 if ($line =~ m/SHA256 Fingerprint=((?:[A-Fa-f0-9]{2}:){31}[A-Fa-f0-9]{2})/) {
314 die "parsing failed\n" if !$fp;
316 die "unable to get remote node fingerprint from '$ni->{name}': $@\n" if $@;
321 sub trigger_update_fingerprints
{
324 my $master = $cinfo->{master
} || die "unable to lookup master node\n";
325 my $cached_fp = { $master->{fingerprint
} => 1 };
327 # if running on master the current fingerprint for the API-connection is needed
328 # in addition (to prevent races with restarting pmgproxy
329 if ($cinfo->{local}->{type
} eq 'master') {
330 my $new_fp = PMG
::Cluster
::read_local_ssl_cert_fingerprint
();
331 $cached_fp->{$new_fp} = 1;
334 my $ticket = PMG
::Ticket
::assemble_ticket
('root@pam');
335 my $csrftoken = PMG
::Ticket
::assemble_csrf_prevention_token
('root@pam');
336 my $conn = PVE
::APIClient
::LWP-
>new(
338 csrftoken
=> $csrftoken,
339 cookie_name
=> 'PMGAuthCookie',
340 host
=> $master->{ip
},
341 cached_fingerprints
=> $cached_fp,
344 $conn->post("/config/cluster/update-fingerprints", {});
348 my $rsync_command = sub {
349 my ($host_key_alias, @args) = @_;
351 my $ssh_cmd = join(' ', @{$ssh_command->($host_key_alias)});
353 my $cmd = ['rsync', "--rsh=$ssh_cmd", '-q', @args];
358 sub sync_quarantine_files
{
359 my ($host_ip, $host_name, $flistname, $rcid) = @_;
361 my $spooldir = $PMG::MailQueue
::spooldir
;
363 mkdir "$spooldir/cluster/";
364 my $syncdir = "$spooldir/cluster/$rcid";
367 my $cmd = $rsync_command->(
368 $host_name, '--timeout', '10', "[${host_ip}]:$spooldir", $spooldir,
369 '--files-from', $flistname);
371 PVE
::Tools
::run_command
($cmd);
375 my ($host_ip, $host_name, $rcid) = @_;
377 my $spooldir = $PMG::MailQueue
::spooldir
;
379 mkdir "$spooldir/cluster/";
380 my $syncdir = "$spooldir/cluster/$rcid";
383 my $cmd = $rsync_command->(
384 $host_name, '-aq', '--timeout', '10', "[${host_ip}]:$syncdir/", $syncdir);
386 foreach my $incl (('spam/', 'spam/*', 'spam/*/*', 'virus/', 'virus/*', 'virus/*/*')) {
387 push @$cmd, '--include', $incl;
390 push @$cmd, '--exclude', '*';
392 PVE
::Tools
::run_command
($cmd);
395 sub sync_master_quar
{
396 my ($host_ip, $host_name) = @_;
398 my $spooldir = $PMG::MailQueue
::spooldir
;
400 my $syncdir = "$spooldir/cluster/";
403 my $cmd = $rsync_command->(
404 $host_name, '-aq', '--timeout', '10', "[${host_ip}]:$syncdir", $syncdir);
406 PVE
::Tools
::run_command
($cmd);
409 sub sync_config_from_master
{
410 my ($master_name, $master_ip, $noreload) = @_;
413 File
::Path
::remove_tree
($syncdir, {keep_root
=> 1});
415 my $sa_conf_dir = "/etc/mail/spamassassin";
416 my $sa_custom_cf = "custom.cf";
417 my $sa_rules_cf = "pmg-scores.cf";
419 my $cmd = $rsync_command->(
421 "[${master_ip}]:$cfgdir/* ${sa_conf_dir}/${sa_custom_cf} ${sa_conf_dir}/${sa_rules_cf}",
423 '--exclude', 'master/',
426 '--exclude', 'pmg-api.pem',
427 '--exclude', 'pmg-tls.pem',
430 my $errmsg = "syncing master configuration from '${master_ip}' failed";
431 PVE
::Tools
::run_command
($cmd, errmsg
=> $errmsg);
433 # verify that the remote host is cluster master
434 open (my $fh, '<', "$syncdir/cluster.conf") ||
435 die "unable to open synced cluster.conf - $!\n";
437 my $cinfo = PMG
::ClusterConfig
::read_cluster_conf
('cluster.conf', $fh);
439 if (!$cinfo->{master
} || ($cinfo->{master
}->{ip
} ne $master_ip)) {
440 die "host '$master_ip' is not cluster master\n";
443 my $role = $cinfo->{'local'}->{type
} // '-';
444 die "local node '$cinfo->{local}->{name}' not part of cluster\n"
447 die "local node '$cinfo->{local}->{name}' is new cluster master\n"
448 if $role eq 'master';
450 $cond_commit_synced_file->('cluster.conf');
452 update_ssh_keys
($cinfo); # rewrite ssh keys
454 PMG
::Fetchmail
::update_fetchmail_default
(0); # disable on slave
469 foreach my $filename (@$files) {
470 $cond_commit_synced_file->($filename);
480 foreach my $dir (@$dirs) {
481 my $srcdir = "$syncdir/$dir";
484 my $cmd = ['rsync', '-aq', '--delete-after', "$srcdir/", "$cfgdir/$dir"];
485 PVE
::Tools
::run_command
($cmd);
490 my $force_restart = {};
492 for my $file (($sa_custom_cf, $sa_rules_cf)) {
493 if ($cond_commit_synced_file->($file, "${sa_conf_dir}/${file}")) {
494 $force_restart->{'pmg-smtp-filter'} = 1;
498 $cond_commit_synced_file->('pmg.conf');
500 return $force_restart;
503 sub sync_ruledb_from_master
{
504 my ($ldb, $rdb, $ni, $ticket) = @_;
506 my $ruledb = PMG
::RuleDB-
>new($ldb);
507 my $rulecache = PMG
::RuleCache-
>new($ruledb);
509 my $conn = PVE
::APIClient
::LWP-
>new(
511 cookie_name
=> 'PMGAuthCookie',
513 cached_fingerprints
=> {
514 $ni->{fingerprint
} => 1,
517 my $digest = $conn->get("/config/ruledb/digest", {});
519 return if $digest eq $rulecache->{digest
}; # no changes
521 syslog
('info', "detected rule database changes - starting sync from '$ni->{ip}'");
526 $ldb->do("DELETE FROM Rule");
527 $ldb->do("DELETE FROM RuleGroup");
528 $ldb->do("DELETE FROM ObjectGroup");
529 $ldb->do("DELETE FROM Object");
530 $ldb->do("DELETE FROM Attribut");
535 # read a consistent snapshot
536 $rdb->do("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE");
538 PMG
::DBTools
::copy_table
($ldb, $rdb, "Rule");
539 PMG
::DBTools
::copy_table
($ldb, $rdb, "RuleGroup");
540 PMG
::DBTools
::copy_table
($ldb, $rdb, "ObjectGroup");
541 PMG
::DBTools
::copy_table
($ldb, $rdb, "Object", 'value');
542 PMG
::DBTools
::copy_table
($ldb, $rdb, "Attribut", 'value');
545 $rdb->rollback; # end transaction
551 $ldb->do("SELECT setval('rule_id_seq', max(id)+1) FROM Rule");
552 $ldb->do("SELECT setval('object_id_seq', max(id)+1) FROM Object");
553 $ldb->do("SELECT setval('objectgroup_id_seq', max(id)+1) FROM ObjectGroup");
562 PMG
::DBTools
::reload_ruledb
();
564 syslog
('info', "finished rule database sync from host '$ni->{ip}'");
567 sub sync_quarantine_db
{
568 my ($ldb, $rdb, $ni, $rsynctime_ref) = @_;
570 my $rcid = $ni->{cid
};
572 my $maxmails = 100000;
576 my $ctime = PMG
::DBTools
::get_remote_time
($rdb);
582 PMG
::DBTools
::create_clusterinfo_default
($ldb, $rcid, 'lastid_CMailStore', -1, undef);
584 do { # get new values
588 my $flistname = "/tmp/quarantinefilelist.$$";
593 open(my $flistfh, '>', $flistname) ||
594 die "unable to open file '$flistname' - $!\n";
596 my $lastid = PMG
::DBTools
::read_int_clusterinfo
($ldb, $rcid, 'lastid_CMailStore');
600 my $sth = $rdb->prepare(
601 "SELECT * from CMailstore WHERE cid = ? AND rid > ? " .
602 "ORDER BY cid,rid LIMIT ?");
603 $sth->execute($rcid, $lastid, $maxcount);
608 $maxid = $ref->{rid
};
609 my $filename = $ref->{file
};
610 # skip files generated before cluster was created
611 return if $filename !~ m!^cluster/!;
612 print $flistfh "$filename\n";
615 my $attrs = [qw(cid rid time qtype bytes spamlevel info sender header file)];
616 $count += PMG
::DBTools
::copy_selected_data
($ldb, $sth, 'CMailStore', $attrs, $callback);
620 my $starttime = [ gettimeofday
() ];
621 sync_quarantine_files
($ni->{ip
}, $ni->{name
}, $flistname, $rcid);
622 $$rsynctime_ref += tv_interval
($starttime);
627 $sth = $rdb->prepare(
628 "SELECT * from CMSReceivers WHERE " .
629 "CMailStore_CID = ? AND CMailStore_RID > ? " .
630 "AND CMailStore_RID <= ?");
631 $sth->execute($rcid, $lastid, $maxid);
633 $attrs = [qw(cmailstore_cid cmailstore_rid pmail receiver ticketid status mtime)];
634 PMG
::DBTools
::copy_selected_data
($ldb, $sth, 'CMSReceivers', $attrs);
636 PMG
::DBTools
::write_maxint_clusterinfo
($ldb, $rcid, 'lastid_CMailStore', $maxid);
652 } while (($count >= $maxcount) && ($mscount < $maxmails));
654 PMG
::DBTools
::create_clusterinfo_default
($ldb, $rcid, 'lastmt_CMSReceivers', 0, undef);
656 eval { # synchronize status updates
659 my $lastmt = PMG
::DBTools
::read_int_clusterinfo
($ldb, $rcid, 'lastmt_CMSReceivers');
661 my $sth = $rdb->prepare ("SELECT * from CMSReceivers WHERE mtime >= ? AND status != 'N'");
662 $sth->execute($lastmt);
664 my $update_sth = $ldb->prepare(
665 "UPDATE CMSReceivers SET status = ? WHERE " .
666 "CMailstore_CID = ? AND CMailstore_RID = ? AND TicketID = ?");
667 while (my $ref = $sth->fetchrow_hashref()) {
668 $update_sth->execute($ref->{status
}, $ref->{cmailstore_cid
},
669 $ref->{cmailstore_rid
}, $ref->{ticketid
});
672 PMG
::DBTools
::write_maxint_clusterinfo
($ldb, $rcid, 'lastmt_CMSReceivers', $ctime);
684 sub sync_statistic_db
{
685 my ($ldb, $rdb, $ni) = @_;
687 my $rcid = $ni->{cid
};
689 my $maxmails = 100000;
697 PMG
::DBTools
::create_clusterinfo_default
(
698 $ldb, $rcid, 'lastid_CStatistic', -1, undef);
700 do { # get new values
707 my $lastid = PMG
::DBTools
::read_int_clusterinfo
(
708 $ldb, $rcid, 'lastid_CStatistic');
712 my $sth = $rdb->prepare(
713 "SELECT * from CStatistic " .
714 "WHERE cid = ? AND rid > ? " .
715 "ORDER BY cid, rid LIMIT ?");
716 $sth->execute($rcid, $lastid, $maxcount);
721 $maxid = $ref->{rid
};
724 my $attrs = [qw(cid rid time bytes direction spamlevel ptime virusinfo sender)];
725 $count += PMG
::DBTools
::copy_selected_data
($ldb, $sth, 'CStatistic', $attrs, $callback);
730 $sth = $rdb->prepare(
731 "SELECT * from CReceivers WHERE " .
732 "CStatistic_CID = ? AND CStatistic_RID > ? AND CStatistic_RID <= ?");
733 $sth->execute($rcid, $lastid, $maxid);
735 $attrs = [qw(cstatistic_cid cstatistic_rid blocked receiver)];
736 PMG
::DBTools
::copy_selected_data
($ldb, $sth, 'CReceivers', $attrs);
739 PMG
::DBTools
::write_maxint_clusterinfo
($ldb, $rcid, 'lastid_CStatistic', $maxid);
750 } while (($count >= $maxcount) && ($mscount < $maxmails));
755 my $sync_generic_mtime_db = sub {
756 my ($ldb, $rdb, $ni, $table, $selectfunc, $mergefunc) = @_;
758 my $ctime = PMG
::DBTools
::get_remote_time
($rdb);
760 PMG
::DBTools
::create_clusterinfo_default
($ldb, $ni->{cid
}, "lastmt_$table", 0, undef);
762 my $lastmt = PMG
::DBTools
::read_int_clusterinfo
($ldb, $ni->{cid
}, "lastmt_$table");
764 my $sql_cmd = $selectfunc->($ctime, $lastmt);
766 my $sth = $rdb->prepare($sql_cmd);
773 # use transaction to speedup things
774 my $max = 1000; # UPDATE MAX ENTRIES AT ONCE
776 while (my $ref = $sth->fetchrow_hashref()) {
777 $ldb->begin_work if !$count;
779 if (++$count >= $max) {
786 $ldb->commit if $count;
793 PMG
::DBTools
::write_maxint_clusterinfo
($ldb, $ni->{cid
}, "lastmt_$table", $ctime);
798 sub sync_localstat_db
{
799 my ($dbh, $rdb, $ni) = @_;
801 my $rcid = $ni->{cid
};
803 my $selectfunc = sub {
804 my ($ctime, $lastmt) = @_;
805 return "SELECT * from LocalStat WHERE mtime >= $lastmt AND cid = $rcid";
808 my $merge_sth = $dbh->prepare(
809 'INSERT INTO LocalStat (Time, RBLCount, PregreetCount, CID, MTime) ' .
810 'VALUES (?, ?, ?, ?, ?) ' .
811 'ON CONFLICT (Time, CID) DO UPDATE SET ' .
812 'RBLCount = excluded.RBLCount, PregreetCount = excluded.PregreetCount, MTime = excluded.MTime');
814 my $mergefunc = sub {
817 $merge_sth->execute($ref->{time}, $ref->{rblcount
}, $ref->{pregreetcount
}, $ref->{cid
}, $ref->{mtime
});
820 return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'LocalStat', $selectfunc, $mergefunc);
823 sub sync_greylist_db
{
824 my ($dbh, $rdb, $ni) = @_;
826 my $selectfunc = sub {
827 my ($ctime, $lastmt) = @_;
828 return "SELECT * from CGreylist WHERE extime >= $ctime AND " .
829 "mtime >= $lastmt AND CID != 0";
832 my $merge_sth = $dbh->prepare(PMG
::DBTools
::cgreylist_merge_sql
());
833 my $mergefunc = sub {
836 my $ipnet = $ref->{ipnet
};
837 $ipnet .= '.0/24' if $ipnet !~ /\
/\d+$/;
839 $ipnet, $ref->{sender
}, $ref->{receiver
},
840 $ref->{instance
}, $ref->{rctime
}, $ref->{extime
}, $ref->{delay
},
841 $ref->{blocked
}, $ref->{passed
}, 0, $ref->{cid
});
844 return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'CGreylist', $selectfunc, $mergefunc);
847 sub sync_userprefs_db
{
848 my ($dbh, $rdb, $ni) = @_;
850 my $selectfunc = sub {
851 my ($ctime, $lastmt) = @_;
853 return "SELECT * from UserPrefs WHERE mtime >= $lastmt";
856 my $merge_sth = $dbh->prepare(
857 "INSERT INTO UserPrefs (PMail, Name, Data, MTime) " .
858 'VALUES (?, ?, ?, ?) ' .
859 'ON CONFLICT (PMail, Name) DO UPDATE SET ' .
860 # Note: MTime = 0 ==> this is just a copy from somewhere else, not modified
861 'MTime = CASE WHEN excluded.MTime >= UserPrefs.MTime THEN 0 ELSE UserPrefs.MTime END, ' .
862 'Data = CASE WHEN excluded.MTime >= UserPrefs.MTime THEN excluded.Data ELSE UserPrefs.Data END');
864 my $mergefunc = sub {
867 $merge_sth->execute($ref->{pmail
}, $ref->{name
}, $ref->{data
}, $ref->{mtime
});
870 return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'UserPrefs', $selectfunc, $mergefunc);
873 sub sync_domainstat_db
{
874 my ($dbh, $rdb, $ni) = @_;
876 my $selectfunc = sub {
877 my ($ctime, $lastmt) = @_;
878 return "SELECT * from DomainStat WHERE mtime >= $lastmt";
881 my $merge_sth = $dbh->prepare(
882 'INSERT INTO Domainstat ' .
883 '(Time,Domain,CountIn,CountOut,BytesIn,BytesOut,VirusIn,VirusOut,SpamIn,SpamOut,' .
884 'BouncesIn,BouncesOut,PTimeSum,Mtime) ' .
885 'VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?) ' .
886 'ON CONFLICT (Time, Domain) DO UPDATE SET ' .
887 'CountIn = excluded.CountIn, CountOut = excluded.CountOut, ' .
888 'BytesIn = excluded.BytesIn, BytesOut = excluded.BytesOut, ' .
889 'VirusIn = excluded.VirusIn, VirusOut = excluded.VirusOut, ' .
890 'SpamIn = excluded.SpamIn, SpamOut = excluded.SpamOut, ' .
891 'BouncesIn = excluded.BouncesIn, BouncesOut = excluded.BouncesOut, ' .
892 'PTimeSum = excluded.PTimeSum, MTime = excluded.MTime');
894 my $mergefunc = sub {
898 $ref->{time}, $ref->{domain
}, $ref->{countin
}, $ref->{countout
},
899 $ref->{bytesin
}, $ref->{bytesout
},
900 $ref->{virusin
}, $ref->{virusout
}, $ref->{spamin
}, $ref->{spamout
},
901 $ref->{bouncesin
}, $ref->{bouncesout
}, $ref->{ptimesum
}, $ref->{mtime
});
904 return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'DomainStat', $selectfunc, $mergefunc);
907 sub sync_dailystat_db
{
908 my ($dbh, $rdb, $ni) = @_;
910 my $selectfunc = sub {
911 my ($ctime, $lastmt) = @_;
912 return "SELECT * from DailyStat WHERE mtime >= $lastmt";
915 my $merge_sth = $dbh->prepare(
916 'INSERT INTO DailyStat ' .
917 '(Time,CountIn,CountOut,BytesIn,BytesOut,VirusIn,VirusOut,SpamIn,SpamOut,' .
918 'BouncesIn,BouncesOut,GreylistCount,SPFCount,RBLCount,PTimeSum,Mtime) ' .
919 'VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ' .
920 'ON CONFLICT (Time) DO UPDATE SET ' .
921 'CountIn = excluded.CountIn, CountOut = excluded.CountOut, ' .
922 'BytesIn = excluded.BytesIn, BytesOut = excluded.BytesOut, ' .
923 'VirusIn = excluded.VirusIn, VirusOut = excluded.VirusOut, ' .
924 'SpamIn = excluded.SpamIn, SpamOut = excluded.SpamOut, ' .
925 'BouncesIn = excluded.BouncesIn, BouncesOut = excluded.BouncesOut, ' .
926 'GreylistCount = excluded.GreylistCount, SPFCount = excluded.SpfCount, ' .
927 'RBLCount = excluded.RBLCount, ' .
928 'PTimeSum = excluded.PTimeSum, MTime = excluded.MTime');
930 my $mergefunc = sub {
934 $ref->{time}, $ref->{countin
}, $ref->{countout
},
935 $ref->{bytesin
}, $ref->{bytesout
},
936 $ref->{virusin
}, $ref->{virusout
}, $ref->{spamin
}, $ref->{spamout
},
937 $ref->{bouncesin
}, $ref->{bouncesout
}, $ref->{greylistcount
},
938 $ref->{spfcount
}, $ref->{rblcount
}, $ref->{ptimesum
}, $ref->{mtime
});
941 return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'DailyStat', $selectfunc, $mergefunc);
944 sub sync_virusinfo_db
{
945 my ($dbh, $rdb, $ni) = @_;
947 my $selectfunc = sub {
948 my ($ctime, $lastmt) = @_;
949 return "SELECT * from VirusInfo WHERE mtime >= $lastmt";
952 my $merge_sth = $dbh->prepare(
953 'INSERT INTO VirusInfo (Time,Name,Count,MTime) ' .
954 'VALUES (?,?,?,?) ' .
955 'ON CONFLICT (Time,Name) DO UPDATE SET ' .
956 'Count = excluded.Count , MTime = excluded.MTime');
958 my $mergefunc = sub {
961 $merge_sth->execute($ref->{time}, $ref->{name
}, $ref->{count
}, $ref->{mtime
});
964 return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'VirusInfo', $selectfunc, $mergefunc);
967 sub sync_deleted_nodes_from_master
{
968 my ($ldb, $masterdb, $cinfo, $masterni, $rsynctime_ref) = @_;
972 my $cid_hash = {}; # fast lookup
973 foreach my $ni (values %{$cinfo->{ids
}}) {
974 $cid_hash->{$ni->{cid
}} = $ni;
977 my $spooldir = $PMG::MailQueue
::spooldir
;
979 my $maxcid = $cinfo->{master
}->{maxcid
} // 0;
981 for (my $rcid = 1; $rcid <= $maxcid; $rcid++) {
982 next if $cid_hash->{$rcid};
984 my $done_marker = "$spooldir/cluster/$rcid/.synced-deleted-node";
986 next if -f
$done_marker; # already synced
988 syslog
('info', "syncing deleted node $rcid from master '$masterni->{ip}'");
990 my $starttime = [ gettimeofday
() ];
991 sync_spooldir
($masterni->{ip
}, $masterni->{name
}, $rcid);
992 $$rsynctime_ref += tv_interval
($starttime);
995 ip
=> $masterni->{ip
},
996 name
=> $masterni->{name
},
1000 sync_quarantine_db
($ldb, $masterdb, $fake_ni);
1002 sync_statistic_db
($ldb, $masterdb, $fake_ni);
1004 open(my $fh, ">>", $done_marker);