PMG/Cluster.pm

   1 package PMG::Cluster;
   2
   3 use strict;
   4 use warnings;
   5 use Data::Dumper;
   6 use Socket;
   7 use File::Path;
   8 use Time::HiRes qw (gettimeofday tv_interval);
   9
  10 use PVE::SafeSyslog;
  11 use PVE::Tools;
  12 use PVE::INotify;
  13
  14 use PMG::Utils;
  15 use PMG::Config;
  16 use PMG::ClusterConfig;
  17 use PMG::RuleDB;
  18 use PMG::RuleCache;
  19 use PMG::MailQueue;
  20 use PVE::APIClient::LWP;
  21
  22 sub remote_node_ip {
  23     my ($nodename, $noerr) = @_;
  24
  25     my $cinfo = PMG::ClusterConfig->new();
  26
  27     foreach my $entry (values %{$cinfo->{ids}}) {
  28         if ($entry->{name} eq $nodename) {
  29             my $ip = $entry->{ip};
  30             return $ip if !wantarray;
  31             my $family = PVE::Tools::get_host_address_family($ip);
  32             return ($ip, $family);
  33         }
  34     }
  35
  36     # fallback: try to get IP by other means
  37     return PMG::Utils::lookup_node_ip($nodename, $noerr);
  38 }
  39
  40 sub get_master_node {
  41     my ($cinfo) = @_;
  42
  43     $cinfo = PMG::ClusterConfig->new() if !$cinfo;
  44
  45     return $cinfo->{master}->{name} if defined($cinfo->{master});
  46
  47     return 'localhost';
  48 }
  49
  50 sub read_local_ssl_cert_fingerprint {
  51     my $cert_path = "/etc/pmg/pmg-api.pem";
  52
  53     my $cert;
  54     eval {
  55         my $bio = Net::SSLeay::BIO_new_file($cert_path, 'r');
  56         $cert = Net::SSLeay::PEM_read_bio_X509($bio);
  57         Net::SSLeay::BIO_free($bio);
  58     };
  59     if (my $err = $@) {
  60         die "unable to read certificate '$cert_path' - $err\n";
  61     }
  62
  63     if (!defined($cert)) {
  64         die "unable to read certificate '$cert_path' - got empty value\n";
  65     }
  66
  67     my $fp;
  68     eval {
  69         $fp = Net::SSLeay::X509_get_fingerprint($cert, 'sha256');
  70     };
  71     if (my $err = $@) {
  72         die "unable to get fingerprint for '$cert_path' - $err\n";
  73     }
  74
  75     if (!defined($fp) || $fp eq '') {
  76         die "unable to get fingerprint for '$cert_path' - got empty value\n";
  77     }
  78
  79     return $fp;
  80 }
  81
  82 my $hostrsapubkey_fn = '/etc/ssh/ssh_host_rsa_key.pub';
  83 my $rootrsakey_fn = '/root/.ssh/id_rsa';
  84 my $rootrsapubkey_fn = '/root/.ssh/id_rsa.pub';
  85
  86 sub read_local_cluster_info {
  87
  88     my $res = {};
  89
  90     my $hostrsapubkey = PVE::Tools::file_read_firstline($hostrsapubkey_fn);
  91     $hostrsapubkey =~ s/^.*ssh-rsa\s+//i;
  92     $hostrsapubkey =~ s/\s+root\@\S+\s*$//i;
  93
  94     die "unable to parse ${hostrsapubkey_fn}\n"
  95         if $hostrsapubkey !~ m/^[A-Za-z0-9\.\/\+]{200,}$/;
  96
  97     my $nodename = PVE::INotify::nodename();
  98
  99     $res->{name} = $nodename;
 100
 101     $res->{ip} = PMG::Utils::lookup_node_ip($nodename);
 102
 103     $res->{hostrsapubkey} = $hostrsapubkey;
 104
 105     if (! -f $rootrsapubkey_fn) {
 106         unlink $rootrsakey_fn;
 107         my $cmd = ['ssh-keygen', '-t', 'rsa', '-N', '', '-b', '2048',
 108                    '-f', $rootrsakey_fn];
 109         PMG::Utils::run_silent_cmd($cmd);
 110     }
 111
 112     my $rootrsapubkey = PVE::Tools::file_read_firstline($rootrsapubkey_fn);
 113     $rootrsapubkey =~ s/^.*ssh-rsa\s+//i;
 114     $rootrsapubkey =~ s/\s+root\@\S+\s*$//i;
 115
 116     die "unable to parse ${rootrsapubkey_fn}\n"
 117         if $rootrsapubkey !~ m/^[A-Za-z0-9\.\/\+]{200,}$/;
 118
 119     $res->{rootrsapubkey} = $rootrsapubkey;
 120
 121     $res->{fingerprint} = read_local_ssl_cert_fingerprint();
 122
 123     return $res;
 124 }
 125
 126 # X509 Certificate cache helper
 127
 128 my $cert_cache_nodes = {};
 129 my $cert_cache_timestamp = time();
 130 my $cert_cache_fingerprints = {};
 131
 132 sub update_cert_cache {
 133
 134     $cert_cache_timestamp = time();
 135
 136     $cert_cache_fingerprints = {};
 137     $cert_cache_nodes = {};
 138
 139     my $cinfo = PMG::ClusterConfig->new();
 140
 141     foreach my $entry (values %{$cinfo->{ids}}) {
 142         my $node = $entry->{name};
 143         my $fp = $entry->{fingerprint};
 144         if ($node && $fp) {
 145             $cert_cache_fingerprints->{$fp} = 1;
 146             $cert_cache_nodes->{$node} = $fp;
 147         }
 148     }
 149 }
 150
 151 # load and cache cert fingerprint once
 152 sub initialize_cert_cache {
 153     my ($node) = @_;
 154
 155     update_cert_cache()
 156         if defined($node) && !defined($cert_cache_nodes->{$node});
 157 }
 158
 159 sub check_cert_fingerprint {
 160     my ($cert) = @_;
 161
 162     # clear cache every 30 minutes at least
 163     update_cert_cache() if time() - $cert_cache_timestamp >= 60*30;
 164
 165     # get fingerprint of server certificate
 166     my $fp;
 167     eval {
 168         $fp = Net::SSLeay::X509_get_fingerprint($cert, 'sha256');
 169     };
 170     return 0 if $@ || !defined($fp) || $fp eq ''; # error
 171
 172     my $check = sub {
 173         for my $expected (keys %$cert_cache_fingerprints) {
 174             return 1 if $fp eq $expected;
 175         }
 176         return 0;
 177     };
 178
 179     return 1 if $check->();
 180
 181     # clear cache and retry at most once every minute
 182     if (time() - $cert_cache_timestamp >= 60) {
 183         syslog ('info', "Could not verify remote node certificate '$fp' with list of pinned certificates, refreshing cache");
 184         update_cert_cache();
 185         return $check->();
 186     }
 187
 188     return 0;
 189 }
 190
 191 my $sshglobalknownhosts = "/etc/ssh/ssh_known_hosts2";
 192 my $rootsshauthkeys = "/root/.ssh/authorized_keys";
 193 my $ssh_rsa_id = "/root/.ssh/id_rsa.pub";
 194
 195 sub update_ssh_keys {
 196     my ($cinfo) = @_;
 197
 198     my $data = '';
 199     foreach my $node (values %{$cinfo->{ids}}) {
 200         $data .= "$node->{ip} ssh-rsa $node->{hostrsapubkey}\n";
 201         $data .= "$node->{name} ssh-rsa $node->{hostrsapubkey}\n";
 202     }
 203
 204     PVE::Tools::file_set_contents($sshglobalknownhosts, $data);
 205
 206     $data = '';
 207
 208     # always add ourself
 209     if (-f $ssh_rsa_id) {
 210         my $pub = PVE::Tools::file_get_contents($ssh_rsa_id);
 211         chomp($pub);
 212         $data .= "$pub\n";
 213     }
 214
 215     foreach my $node (values %{$cinfo->{ids}}) {
 216         $data .= "ssh-rsa $node->{rootrsapubkey} root\@$node->{name}\n";
 217     }
 218
 219     if (-f $rootsshauthkeys) {
 220         my $mykey = PVE::Tools::file_get_contents($rootsshauthkeys, 128*1024);
 221         chomp($mykey);
 222         $data .= "$mykey\n";
 223     }
 224
 225     my $newdata = "";
 226     my $vhash = {};
 227     my @lines = split(/\n/, $data);
 228     foreach my $line (@lines) {
 229         if ($line !~ /^#/ && $line =~ m/(^|\s)ssh-(rsa|dsa)\s+(\S+)\s+\S+$/) {
 230             next if $vhash->{$3}++;
 231         }
 232         $newdata .= "$line\n";
 233     }
 234
 235     PVE::Tools::file_set_contents($rootsshauthkeys, $newdata, 0600);
 236 }
 237
 238 my $cfgdir = '/etc/pmg';
 239 my $syncdir = "$cfgdir/master";
 240
 241 my $cond_commit_synced_file = sub {
 242     my ($filename, $dstfn) = @_;
 243
 244     $dstfn = "$cfgdir/$filename" if !defined($dstfn);
 245     my $srcfn = "$syncdir/$filename";
 246
 247     if (! -f $srcfn) {
 248         unlink $dstfn;
 249         return;
 250     }
 251
 252     my $new = PVE::Tools::file_get_contents($srcfn, 1024*1024);
 253
 254     if (-f $dstfn) {
 255         my $old = PVE::Tools::file_get_contents($dstfn, 1024*1024);
 256         return 0 if $new eq $old;
 257     }
 258
 259     rename($srcfn, $dstfn) ||
 260         die "cond_rename_file '$filename' failed - $!\n";
 261
 262     print STDERR "updated $dstfn\n";
 263
 264     return 1;
 265 };
 266
 267 my $rsync_command = sub {
 268     my ($host_key_alias, @args) = @_;
 269
 270     my $ssh_cmd = '--rsh=ssh -l root -o BatchMode=yes';
 271     $ssh_cmd .=  " -o HostKeyAlias=${host_key_alias}" if $host_key_alias;
 272
 273     my $cmd = ['rsync', $ssh_cmd,  '-q', @args];
 274
 275     return $cmd;
 276 };
 277
 278 sub sync_quarantine_files {
 279     my ($host_ip, $host_name, $flistname) = @_;
 280
 281     my $spooldir = $PMG::MailQueue::spooldir;
 282
 283     my $cmd = $rsync_command->(
 284         $host_name, '--timeout', '10', "${host_ip}:$spooldir", $spooldir,
 285         '--files-from', $flistname);
 286
 287     PVE::Tools::run_command($cmd);
 288 }
 289
 290 sub sync_spooldir {
 291     my ($host_ip, $host_name, $rcid) = @_;
 292
 293     my $spooldir = $PMG::MailQueue::spooldir;
 294
 295     mkdir "$spooldir/cluster/";
 296     my $syncdir = "$spooldir/cluster/$rcid";
 297     mkdir $syncdir;
 298
 299     my $cmd = $rsync_command->(
 300         $host_name, '-aq', '--timeout', '10', "${host_ip}:$syncdir/", $syncdir);
 301
 302     foreach my $incl (('spam/', 'spam/*', 'spam/*/*', 'virus/', 'virus/*', 'virus/*/*')) {
 303         push @$cmd, '--include', $incl;
 304     }
 305
 306     push @$cmd, '--exclude', '*';
 307
 308     PVE::Tools::run_command($cmd);
 309 }
 310
 311 sub sync_master_quar {
 312     my ($host_ip, $host_name) = @_;
 313
 314     my $spooldir = $PMG::MailQueue::spooldir;
 315
 316     my $syncdir = "$spooldir/cluster/";
 317     mkdir $syncdir;
 318
 319     my $cmd = $rsync_command->(
 320         $host_name, '-aq', '--timeout', '10', "${host_ip}:$syncdir", $syncdir);
 321
 322     PVE::Tools::run_command($cmd);
 323 }
 324
 325 sub sync_config_from_master {
 326     my ($master_name, $master_ip, $noreload) = @_;
 327
 328     mkdir $syncdir;
 329     File::Path::remove_tree($syncdir, {keep_root => 1});
 330
 331     my $sa_conf_dir = "/etc/mail/spamassassin";
 332     my $sa_custom_cf = "custom.cf";
 333
 334     my $cmd = $rsync_command->(
 335         $master_name, '-lpgoq',
 336         "${master_ip}:$cfgdir/* ${sa_conf_dir}/${sa_custom_cf}",
 337         "$syncdir/",
 338         '--exclude', '*~',
 339         '--exclude', '*.db',
 340         '--exclude', 'pmg-api.pem',
 341         '--exclude', 'pmg-tls.pem',
 342         );
 343
 344     my $errmsg = "syncing master configuration from '${master_ip}' failed";
 345     PVE::Tools::run_command($cmd, errmsg => $errmsg);
 346
 347     # verify that the remote host is cluster master
 348     open (my $fh, '<', "$syncdir/cluster.conf") ||
 349         die "unable to open synced cluster.conf - $!\n";
 350
 351     my $cinfo = PMG::ClusterConfig::read_cluster_conf('cluster.conf', $fh);
 352
 353     if (!$cinfo->{master} || ($cinfo->{master}->{ip} ne $master_ip)) {
 354         die "host '$master_ip' is not cluster master\n";
 355     }
 356
 357     my $role = $cinfo->{'local'}->{type} // '-';
 358     die "local node '$cinfo->{local}->{name}' not part of cluster\n"
 359         if $role eq '-';
 360
 361     die "local node '$cinfo->{local}->{name}' is new cluster master\n"
 362         if $role eq 'master';
 363
 364     $cond_commit_synced_file->('cluster.conf');
 365
 366     my $files = [
 367         'pmg-authkey.key',
 368         'pmg-authkey.pub',
 369         'pmg-csrf.key',
 370         'ldap.conf',
 371         'user.conf',
 372         ];
 373
 374     foreach my $filename (@$files) {
 375         $cond_commit_synced_file->($filename);
 376     }
 377
 378     my $force_restart = {};
 379
 380     if ($cond_commit_synced_file->($sa_custom_cf, "${sa_conf_dir}/${sa_custom_cf}")) {
 381         $force_restart->{spam} = 1;
 382     }
 383
 384     $cond_commit_synced_file->('pmg.conf');
 385 }
 386
 387 sub sync_ruledb_from_master {
 388     my ($ldb, $rdb, $ni, $ticket) = @_;
 389
 390     my $ruledb = PMG::RuleDB->new($ldb);
 391     my $rulecache = PMG::RuleCache->new($ruledb);
 392
 393     my $conn = PVE::APIClient::LWP->new(
 394         ticket => $ticket,
 395         cookie_name => 'PMGAuthCookie',
 396         host => $ni->{ip},
 397         cached_fingerprints => {
 398             $ni->{fingerprint} => 1,
 399         });
 400
 401     my $digest = $conn->get("/config/ruledb/digest", {});
 402
 403     return if $digest eq $rulecache->{digest}; # no changes
 404
 405     syslog('info', "detected rule database changes - starting sync from '$ni->{ip}'");
 406
 407     eval {
 408         $ldb->begin_work;
 409
 410         $ldb->do("DELETE FROM Rule");
 411         $ldb->do("DELETE FROM RuleGroup");
 412         $ldb->do("DELETE FROM ObjectGroup");
 413         $ldb->do("DELETE FROM Object");
 414         $ldb->do("DELETE FROM Attribut");
 415
 416         eval {
 417             $rdb->begin_work;
 418
 419             # read a consistent snapshot
 420             $rdb->do("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE");
 421
 422             PMG::DBTools::copy_table($ldb, $rdb, "Rule");
 423             PMG::DBTools::copy_table($ldb, $rdb, "RuleGroup");
 424             PMG::DBTools::copy_table($ldb, $rdb, "ObjectGroup");
 425             PMG::DBTools::copy_table($ldb, $rdb, "Object", 'value');
 426             PMG::DBTools::copy_table($ldb, $rdb, "Attribut", 'value');
 427         };
 428
 429         $rdb->rollback; # end transaction
 430
 431         die $@ if $@;
 432
 433         # update sequences
 434
 435         $ldb->do("SELECT setval('rule_id_seq', max(id)+1) FROM Rule");
 436         $ldb->do("SELECT setval('object_id_seq', max(id)+1) FROM Object");
 437         $ldb->do("SELECT setval('objectgroup_id_seq', max(id)+1) FROM ObjectGroup");
 438
 439         $ldb->commit;
 440     };
 441     if (my $err = $@) {
 442         $ldb->rollback;
 443         die $err;
 444     }
 445
 446     syslog('info', "finished rule database sync from host '$ni->{ip}'");
 447 }
 448
 449 sub sync_quarantine_db {
 450     my ($ldb, $rdb, $ni, $rsynctime_ref) = @_;
 451
 452     my $rcid = $ni->{cid};
 453
 454     my $maxmails = 100000;
 455
 456     my $mscount = 0;
 457
 458     my $ctime = PMG::DBTools::get_remote_time($rdb);
 459
 460     my $maxcount = 1000;
 461
 462     my $count;
 463
 464     PMG::DBTools::create_clusterinfo_default($ldb, $rcid, 'lastid_CMailStore', -1, undef);
 465
 466     do { # get new values
 467
 468         $count = 0;
 469
 470         my $flistname = "/tmp/quarantinefilelist.$$";
 471
 472         eval {
 473             $ldb->begin_work;
 474
 475             open(my $flistfh, '>', $flistname) ||
 476                 die "unable to open file '$flistname' - $!\n";
 477
 478             my $lastid = PMG::DBTools::read_int_clusterinfo($ldb, $rcid, 'lastid_CMailStore');
 479
 480             # sync CMailStore
 481
 482             my $sth = $rdb->prepare(
 483                 "SELECT * from CMailstore WHERE cid = ? AND rid > ? " .
 484                 "ORDER BY cid,rid LIMIT ?");
 485             $sth->execute($rcid, $lastid, $maxcount);
 486
 487             my $maxid;
 488             my $callback = sub {
 489                 my $ref = shift;
 490                 $maxid = $ref->{rid};
 491                 print $flistfh "$ref->{file}\n";
 492             };
 493
 494             my $attrs = [qw(cid rid time qtype bytes spamlevel info sender header file)];
 495             $count += PMG::DBTools::copy_selected_data($ldb, $sth, 'CMailStore', $attrs, $callback);
 496
 497             close($flistfh);
 498
 499             my $starttime = [ gettimeofday() ];
 500             sync_quarantine_files($ni->{ip}, $ni->{name}, $flistname);
 501             $$rsynctime_ref += tv_interval($starttime);
 502
 503             if ($maxid) {
 504                 # sync CMSReceivers
 505
 506                 $sth = $rdb->prepare(
 507                     "SELECT * from CMSReceivers WHERE " .
 508                     "CMailStore_CID = ? AND CMailStore_RID > ?  " .
 509                     "AND CMailStore_RID <= ?");
 510                 $sth->execute($rcid, $lastid, $maxid);
 511
 512                 $attrs = [qw(cmailstore_cid cmailstore_rid pmail receiver ticketid status mtime)];
 513                 PMG::DBTools::copy_selected_data($ldb, $sth, 'CMSReceivers', $attrs);
 514
 515                 PMG::DBTools::write_maxint_clusterinfo($ldb, $rcid, 'lastid_CMailStore', $maxid);
 516             }
 517
 518             $ldb->commit;
 519         };
 520         my $err = $@;
 521
 522         unlink $flistname;
 523
 524         if ($err) {
 525             $ldb->rollback;
 526             die $err;
 527         }
 528
 529         $mscount += $count;
 530
 531         last if $mscount >= $maxmails;
 532
 533     } while ($count >= $maxcount);
 534
 535     PMG::DBTools::create_clusterinfo_default($ldb, $rcid, 'lastmt_CMSReceivers', 0, undef);
 536
 537     eval { # synchronize status updates
 538         $ldb->begin_work;
 539
 540         my $lastmt = PMG::DBTools::read_int_clusterinfo($ldb, $rcid, 'lastmt_CMSReceivers');
 541
 542         my $sth = $rdb->prepare ("SELECT * from CMSReceivers WHERE mtime >= ? AND status != 'N'");
 543         $sth->execute($lastmt);
 544
 545         my $update_sth = $ldb->prepare(
 546             "UPDATE CMSReceivers SET status = ? WHERE " .
 547             "CMailstore_CID = ? AND CMailstore_RID = ? AND PMail = ?;");
 548         while (my $ref = $sth->fetchrow_hashref()) {
 549             $update_sth->execute($ref->{status}, $ref->{cmailstore_cid},
 550                                  $ref->{cmailstore_rid}, $ref->{pmail});
 551         }
 552
 553         PMG::DBTools::write_maxint_clusterinfo($ldb, $rcid, 'lastmt_CMSReceivers', $ctime);
 554
 555         $ldb->commit;
 556     };
 557     if (my $err = $@) {
 558         $ldb->rollback;
 559         die $err;
 560     }
 561
 562     return $mscount;
 563 }
 564
 565 sub sync_statistic_db {
 566     my ($ldb, $rdb, $ni) = @_;
 567
 568     my $rcid = $ni->{cid};
 569
 570     my $maxmails = 100000;
 571
 572     my $mscount = 0;
 573
 574     my $maxcount = 1000;
 575
 576     my $count;
 577
 578     PMG::DBTools::create_clusterinfo_default(
 579         $ldb, $rcid, 'lastid_CStatistic', -1, undef);
 580
 581     do { # get new values
 582
 583         $count = 0;
 584
 585         eval {
 586             $ldb->begin_work;
 587
 588             my $lastid = PMG::DBTools::read_int_clusterinfo(
 589                 $ldb, $rcid, 'lastid_CStatistic');
 590
 591             # sync CStatistic
 592
 593             my $sth = $rdb->prepare(
 594                 "SELECT * from CStatistic " .
 595                 "WHERE cid = ? AND rid > ? " .
 596                 "ORDER BY cid, rid LIMIT ?");
 597             $sth->execute($rcid, $lastid, $maxcount);
 598
 599             my $maxid;
 600             my $callback = sub {
 601                 my $ref = shift;
 602                 $maxid = $ref->{rid};
 603             };
 604
 605             my $attrs = [qw(cid rid time bytes direction spamlevel ptime virusinfo sender)];
 606             $count += PMG::DBTools::copy_selected_data($ldb, $sth, 'CStatistic', $attrs, $callback);
 607
 608             if ($maxid) {
 609                 # sync CReceivers
 610
 611                 $sth = $rdb->prepare(
 612                     "SELECT * from CReceivers WHERE " .
 613                     "CStatistic_CID = ? AND CStatistic_RID > ? AND CStatistic_RID <= ?");
 614                 $sth->execute($rcid, $lastid, $maxid);
 615
 616                 $attrs = [qw(cstatistic_cid cstatistic_rid blocked receiver)];
 617                 PMG::DBTools::copy_selected_data($ldb, $sth, 'CReceivers', $attrs);
 618
 619                 PMG::DBTools::write_maxint_clusterinfo ($ldb, $rcid, 'lastid_CStatistic', $maxid);
 620             }
 621
 622             $ldb->commit;
 623         };
 624         if (my $err = $@) {
 625             $ldb->rollback;
 626             die $err;
 627         }
 628
 629         $mscount += $count;
 630
 631         last if $mscount >= $maxmails;
 632
 633     } while ($count >= $maxcount);
 634
 635     return $mscount;
 636 }
 637
 638 my $sync_generic_mtime_db = sub {
 639     my ($ldb, $rdb, $ni, $table, $selectfunc, $mergefunc) = @_;
 640
 641     my $ctime = PMG::DBTools::get_remote_time($rdb);
 642
 643     PMG::DBTools::create_clusterinfo_default($ldb, $ni->{cid}, "lastmt_$table", 0, undef);
 644
 645     my $lastmt = PMG::DBTools::read_int_clusterinfo($ldb, $ni->{cid}, "lastmt_$table");
 646
 647     my $sql_cmd = $selectfunc->($ctime, $lastmt);
 648
 649     my $sth = $rdb->prepare($sql_cmd);
 650
 651     $sth->execute();
 652
 653     my $updates = 0;
 654
 655     eval {
 656         # use transaction to speedup things
 657         my $max = 1000; # UPDATE MAX ENTRIES AT ONCE
 658         my $count = 0;
 659         while (my $ref = $sth->fetchrow_hashref()) {
 660             $ldb->begin_work if !$count;
 661             $mergefunc->($ref);
 662             if (++$count >= $max) {
 663                 $count = 0;
 664                 $ldb->commit;
 665             }
 666             $updates++;
 667         }
 668
 669         $ldb->commit if $count;
 670     };
 671     if (my $err = $@) {
 672         $ldb->rollback;
 673         die $err;
 674     }
 675
 676     PMG::DBTools::write_maxint_clusterinfo($ldb, $ni->{cid}, "lastmt_$table", $ctime);
 677
 678     return $updates;
 679 };
 680
 681 sub sync_greylist_db {
 682     my ($dbh, $rdb, $ni) = @_;
 683
 684     my $selectfunc = sub {
 685         my ($ctime, $lastmt) = @_;
 686         return "SELECT * from CGreylist WHERE extime >= $ctime AND " .
 687             "mtime >= $lastmt AND CID != 0";
 688     };
 689
 690     my $merge_sth = $dbh->prepare($PMG::DBTools::cgreylist_merge_sql);
 691     my $mergefunc = sub {
 692         my ($ref) = @_;
 693
 694         $merge_sth->execute(
 695             $ref->{ipnet}, $ref->{host}, $ref->{sender}, $ref->{receiver},
 696             $ref->{instance}, $ref->{rctime}, $ref->{extime}, $ref->{delay},
 697             $ref->{blocked}, $ref->{passed}, 0, $ref->{cid});
 698     };
 699
 700     return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'CGreylist', $selectfunc, $mergefunc);
 701 }
 702
 703 sub sync_userprefs_db {
 704     my ($dbh, $rdb, $ni) = @_;
 705
 706     my $selectfunc = sub {
 707         my ($ctime, $lastmt) = @_;
 708
 709         return "SELECT * from UserPrefs WHERE mtime >= $lastmt";
 710     };
 711
 712     my $merge_sth = $dbh->prepare(
 713         "INSERT INTO UserPrefs (PMail, Name, Data, MTime) " .
 714         'VALUES (?, ?, ?, 0) ' .
 715         'ON CONFLICT (PMail, Name) DO UPDATE SET ' .
 716         # Note: MTime = 0 ==> this is just a copy from somewhere else, not modified
 717         'MTime = CASE WHEN excluded.MTime >= UserPrefs.MTime THEN 0 ELSE UserPrefs.MTime END' .
 718         'Data = CASE WHEN excluded.MTime >= UserPrefs.MTime THEN excluded.Data ELSE UserPrefs.Data END');
 719
 720     my $mergefunc = sub {
 721         my ($ref) = @_;
 722
 723         $merge_sth->execute($ref->{pmail}, $ref->{name}, $ref->{data});
 724     };
 725
 726     return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'UserPrefs', $selectfunc, $mergefunc);
 727 }
 728
 729 sub sync_domainstat_db {
 730     my ($dbh, $rdb, $ni) = @_;
 731
 732     my $selectfunc = sub {
 733         my ($ctime, $lastmt) = @_;
 734         return "SELECT * from DomainStat WHERE mtime >= $lastmt";
 735     };
 736
 737     my $merge_sth = $dbh->prepare(
 738         'INSERT INTO Domainstat ' .
 739         '(Time,Domain,CountIn,CountOut,BytesIn,BytesOut,VirusIn,VirusOut,SpamIn,SpamOut,' .
 740         'BouncesIn,BouncesOut,PTimeSum,Mtime) ' .
 741         'VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?) ' .
 742         'ON CONFLICT (Time, Domain) DO UPDATE SET ' .
 743         'CountIn = excluded.CountIn, CountOut = excluded.CountOut, ' .
 744         'BytesIn = excluded.BytesIn, BytesOut = excluded.BytesOut, ' .
 745         'VirusIn = excluded.VirusIn, VirusOut = excluded.VirusOut, ' .
 746         'SpamIn = excluded.SpamIn, SpamOut = excluded.SpamOut, ' .
 747         'BouncesIn = excluded.BouncesIn, BouncesOut = excluded.BouncesOut, ' .
 748         'PTimeSum = excluded.PTimeSum, MTime = excluded.MTime');
 749
 750     my $mergefunc = sub {
 751         my ($ref) = @_;
 752
 753         $merge_sth->execute(
 754             $ref->{time}, $ref->{domain}, $ref->{countin}, $ref->{countout},
 755             $ref->{bytesin}, $ref->{bytesout},
 756             $ref->{virusin}, $ref->{virusout}, $ref->{spamin}, $ref->{spamout},
 757             $ref->{bouncesin}, $ref->{bouncesout}, $ref->{ptimesum}, $ref->{mtime});
 758     };
 759
 760     return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'DomainStat', $selectfunc, $mergefunc);
 761 }
 762
 763 sub sync_dailystat_db {
 764     my ($dbh, $rdb, $ni) = @_;
 765
 766     my $selectfunc = sub {
 767         my ($ctime, $lastmt) = @_;
 768         return "SELECT * from DailyStat WHERE mtime >= $lastmt";
 769     };
 770
 771     my $merge_sth = $dbh->prepare(
 772         'INSERT INTO DailyStat ' .
 773         '(Time,CountIn,CountOut,BytesIn,BytesOut,VirusIn,VirusOut,SpamIn,SpamOut,' .
 774         'BouncesIn,BouncesOut,GreylistCount,SPFCount,RBLCount,PTimeSum,Mtime) ' .
 775         'VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ' .
 776         'ON CONFLICT (Time) DO UPDATE SET ' .
 777         'CountIn = excluded.CountIn, CountOut = excluded.CountOut, ' .
 778         'BytesIn = excluded.BytesIn, BytesOut = excluded.BytesOut, ' .
 779         'VirusIn = excluded.VirusIn, VirusOut = excluded.VirusOut, ' .
 780         'SpamIn = excluded.SpamIn, SpamOut = excluded.SpamOut, ' .
 781         'BouncesIn = excluded.BouncesIn, BouncesOut = excluded.BouncesOut, ' .
 782         'GreylistCount = excluded.GreylistCount, SPFCount = excluded.SpfCount, ' .
 783         'RBLCount = excluded.RBLCount, ' .
 784         'PTimeSum = excluded.PTimeSum, MTime = excluded.MTime');
 785
 786     my $mergefunc = sub {
 787         my ($ref) = @_;
 788
 789         $merge_sth->execute(
 790             $ref->{time}, $ref->{countin}, $ref->{countout},
 791             $ref->{bytesin}, $ref->{bytesout},
 792             $ref->{virusin}, $ref->{virusout}, $ref->{spamin}, $ref->{spamout},
 793             $ref->{bouncesin}, $ref->{bouncesout}, $ref->{greylistcount},
 794             $ref->{spfcount}, $ref->{rblcount}, $ref->{ptimesum}, $ref->{mtime});
 795     };
 796
 797     return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'DailyStat', $selectfunc, $mergefunc);
 798 }
 799
 800 sub sync_virusinfo_db {
 801     my ($dbh, $rdb, $ni) = @_;
 802
 803     my $selectfunc = sub {
 804         my ($ctime, $lastmt) = @_;
 805         return "SELECT * from VirusInfo WHERE mtime >= $lastmt";
 806     };
 807
 808     my $merge_sth = $dbh->prepare(
 809         'INSERT INTO VirusInfo (Time,Name,Count,MTime) ' .
 810         'VALUES (?,?,?,?) ' .
 811         'ON CONFLICT (Time,Name) DO UPDATE SET ' .
 812         'Count = excluded.Count , MTime = excluded.MTime');
 813
 814     my $mergefunc = sub {
 815         my ($ref) = @_;
 816
 817         $merge_sth->execute($ref->{time}, $ref->{name}, $ref->{count}, $ref->{mtime});
 818     };
 819
 820     return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'VirusInfo', $selectfunc, $mergefunc);
 821 }
 822
 823 sub sync_deleted_nodes_from_master {
 824     my ($ldb, $masterdb, $cinfo, $masterni, $rsynctime_ref) = @_;
 825
 826     my $rsynctime = 0;
 827
 828     my $cid_hash = {}; # fast lookup
 829     foreach my $ni (values %{$cinfo->{ids}}) {
 830         $cid_hash->{$ni->{cid}} = $ni;
 831     }
 832
 833     my $spooldir = $PMG::MailQueue::spooldir;
 834
 835     my $maxcid = $cinfo->{master}->{maxcid} // 0;
 836
 837     for (my $rcid = 1; $rcid <= $maxcid; $rcid++) {
 838         next if $cid_hash->{$rcid};
 839
 840         my $done_marker = "$spooldir/cluster/$rcid/.synced-deleted-node";
 841
 842         next if -f $done_marker; # already synced
 843
 844         syslog('info', "syncing deleted node $rcid from master '$masterni->{ip}'");
 845
 846         my $starttime = [ gettimeofday() ];
 847         sync_spooldir($masterni->{ip}, $masterni->{name}, $rcid);
 848         $$rsynctime_ref += tv_interval($starttime);
 849
 850         my $fake_ni = {
 851             ip => $masterni->{ip},
 852             name => $masterni->{name},
 853             cid => $rcid,
 854         };
 855
 856         sync_quarantine_db($ldb, $masterdb, $fake_ni);
 857
 858         sync_statistic_db ($ldb, $masterdb, $fake_ni);
 859
 860         open(my $fh, ">>",  $done_marker);
 861    }
 862 }
 863
 864
 865 1;