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