use PVE::SafeSyslog;
use PVE::Tools;
use PVE::INotify;
+use PVE::APIClient::LWP;
use PMG::Utils;
use PMG::Config;
use PMG::RuleDB;
use PMG::RuleCache;
use PMG::MailQueue;
-use PVE::APIClient::LWP;
+use PMG::Fetchmail;
sub remote_node_ip {
my ($nodename, $noerr) = @_;
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) {
$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';
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";
};
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);
my $sa_custom_cf = "custom.cf";
my $cmd = $rsync_command->(
- $master_name, '-lpgoq',
+ $master_name, '-aq',
"${master_ip}:$cfgdir/* ${sa_conf_dir}/${sa_custom_cf}",
"$syncdir/",
+ '--exclude', 'master/',
'--exclude', '*~',
'--exclude', '*.db',
'--exclude', 'pmg-api.pem',
$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',
'pmg-csrf.key',
'ldap.conf',
'user.conf',
+ '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}")) {
$cond_commit_synced_file->('pmg.conf');
- my $cfg = PMG::Config->new();
-
- $cfg->rewrite_config(1, $force_restart);
+ # sync user templates files/symlinks (not recursive)
+ my $srcdir = "$syncdir/templates";
+ if (-d $srcdir) {
+ my $dstdir = "$cfgdir/templates";
+ mkdir $dstdir;
+ my $names_hash = {};
+ foreach my $fn (<$srcdir/*>) {
+ next if $fn !~ m|^($srcdir/(.*))$|;
+ $fn = $1; # untaint;
+ my $name = $2;
+ $names_hash->{$name} = 1;
+ my $target = "$dstdir/$name";
+ if (-f $fn) {
+ $cond_commit_synced_file->("templates/$name", $target);
+ } elsif (-l $fn) {
+ warn "update $target failed - $!\n" if !rename($fn, $target);
+ }
+ }
+ # remove vanished files
+ foreach my $fn (<$dstdir/*>) {
+ next if $fn !~ m|^($dstdir/(.*))$|;
+ $fn = $1; # untaint;
+ my $name = $2;
+ next if $names_hash->{$name};
+ warn "unlink $fn failed - $!\n" if !unlink($fn);
+ }
+ }
}
sub sync_ruledb_from_master {
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)];
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) {
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);
$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 = $@) {
last if $mscount >= $maxmails;
} while ($count >= $maxcount);
+
+ return $mscount;
}
-sub sync_generic_mtime_db {
+my $sync_generic_mtime_db = sub {
my ($ldb, $rdb, $ni, $table, $selectfunc, $mergefunc) = @_;
- my ($cnew, $cold) = (0, 0);
-
my $ctime = PMG::DBTools::get_remote_time($rdb);
PMG::DBTools::create_clusterinfo_default($ldb, $ni->{cid}, "lastmt_$table", 0, undef);
my $lastmt = PMG::DBTools::read_int_clusterinfo($ldb, $ni->{cid}, "lastmt_$table");
- my $cmd = $selectfunc->($ctime, $lastmt);
+ my $sql_cmd = $selectfunc->($ctime, $lastmt);
- my $sth = $rdb->prepare($cmd);
+ my $sth = $rdb->prepare($sql_cmd);
$sth->execute();
- my $max = 100; # UPDATE MAX ENTRIES AT ONCE
-
- my $merge_data = sub {
- my ($data) = @_;
- eval {
- $ldb->begin_work;
-
- # avoid locks
- # $ldb->do ("LOCK TABLE $table IN EXCLUSIVE MODE");
+ my $updates = 0;
- foreach my $ref1 (@$data) {
- $mergefunc->($ldb, $ref1, \$cnew, \$cold);
+ eval {
+ # use transaction to speedup things
+ my $max = 1000; # UPDATE MAX ENTRIES AT ONCE
+ my $count = 0;
+ while (my $ref = $sth->fetchrow_hashref()) {
+ $ldb->begin_work if !$count;
+ $mergefunc->($ref);
+ if (++$count >= $max) {
+ $count = 0;
+ $ldb->commit;
}
-
- $ldb->commit;
- };
- if (my $err = $@) {
- $ldb->rollback;
- die $err;
+ $updates++;
}
+
+ $ldb->commit if $count;
};
+ if (my $err = $@) {
+ $ldb->rollback;
+ die $err;
+ }
- my $data = [];
+ PMG::DBTools::write_maxint_clusterinfo($ldb, $ni->{cid}, "lastmt_$table", $ctime);
- while (my $ref = $sth->fetchrow_hashref()) {
- push @$data, $ref;
- if (scalar(@$data) >= $max) {
- $merge_data->($data);
- $data = [];
- }
- }
+ return $updates;
+};
+
+sub sync_localstat_db {
+ my ($dbh, $rdb, $ni) = @_;
- $merge_data->($data) if scalar(@$data);
+ my $rcid = $ni->{cid};
- PMG::DBTools::write_maxint_clusterinfo($ldb, $ni->{cid}, "lastmt_$table", $ctime);
+ 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 ($cnew, $cold);
+ return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'LocalStat', $selectfunc, $mergefunc);
}
sub sync_greylist_db {
"mtime >= $lastmt AND CID != 0";
};
+ my $merge_sth = $dbh->prepare($PMG::DBTools::cgreylist_merge_sql);
my $mergefunc = sub {
- my ($ldb, $ref, $cnewref, $coldref) = @_;
-
- my $sth = $ldb->prepare("SELECT merge_greylist(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) AS newcount");
-
- $sth->execute($ref->{ipnet}, $ref->{host}, $ref->{sender}, $ref->{receiver},
- $ref->{instance}, $ref->{rctime}, $ref->{extime}, $ref->{delay}, $ref->{blocked},
- $ref->{passed}, 0, $ref->{cid});
-
- my $res = $sth->fetchrow_hashref();
-
- $sth->finish();
+ my ($ref) = @_;
- if ($res->{newcount}) {
- $$cnewref++;
- } else {
- $$coldref++;
- }
+ $merge_sth->execute(
+ $ref->{ipnet}, $ref->{host}, $ref->{sender}, $ref->{receiver},
+ $ref->{instance}, $ref->{rctime}, $ref->{extime}, $ref->{delay},
+ $ref->{blocked}, $ref->{passed}, 0, $ref->{cid});
};
- return sync_generic_mtime_db($dbh, $rdb, $ni, 'CGreylist', $selectfunc, $mergefunc);
+ return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'CGreylist', $selectfunc, $mergefunc);
}
sub sync_userprefs_db {
return "SELECT * from UserPrefs WHERE mtime >= $lastmt";
};
- my $mergefunc = sub {
- my ($ldb, $ref, $cnewref, $coldref) = @_;
-
- my $sth = $ldb->prepare(
- "UPDATE UserPrefs " .
- "SET Data = CASE WHEN MTime >= $ref->{mtime} THEN Data ELSE ? END " .
- "WHERE PMail = ? AND Name = ?");
-
- $sth->execute($ref->{data}, $ref->{pmail}, $ref->{name});
-
- my $rows = $sth->rows;
+ my $merge_sth = $dbh->prepare(
+ "INSERT INTO UserPrefs (PMail, Name, Data, MTime) " .
+ 'VALUES (?, ?, ?, 0) ' .
+ '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, ' .
+ 'Data = CASE WHEN excluded.MTime >= UserPrefs.MTime THEN excluded.Data ELSE UserPrefs.Data END');
- $sth->finish();
+ my $mergefunc = sub {
+ my ($ref) = @_;
- if ($rows) {
- $$coldref++;
- } else {
- $ldb->do ("INSERT INTO UserPrefs (PMail, Name, Data, MTime) " .
- "VALUES (?, ?, ?, 0)", undef,
- $ref->{pmail}, $ref->{name}, $ref->{data});
- $$cnewref++;
- }
+ $merge_sth->execute($ref->{pmail}, $ref->{name}, $ref->{data});
};
- return sync_generic_mtime_db($dbh, $rdb, $ni, 'UserPrefs', $selectfunc, $mergefunc);
+ return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'UserPrefs', $selectfunc, $mergefunc);
}
sub sync_domainstat_db {
return "SELECT * from DomainStat WHERE mtime >= $lastmt";
};
+ my $merge_sth = $dbh->prepare(
+ 'INSERT INTO Domainstat ' .
+ '(Time,Domain,CountIn,CountOut,BytesIn,BytesOut,VirusIn,VirusOut,SpamIn,SpamOut,' .
+ 'BouncesIn,BouncesOut,PTimeSum,Mtime) ' .
+ 'VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?) ' .
+ 'ON CONFLICT (Time, Domain) DO UPDATE SET ' .
+ 'CountIn = excluded.CountIn, CountOut = excluded.CountOut, ' .
+ 'BytesIn = excluded.BytesIn, BytesOut = excluded.BytesOut, ' .
+ 'VirusIn = excluded.VirusIn, VirusOut = excluded.VirusOut, ' .
+ 'SpamIn = excluded.SpamIn, SpamOut = excluded.SpamOut, ' .
+ 'BouncesIn = excluded.BouncesIn, BouncesOut = excluded.BouncesOut, ' .
+ 'PTimeSum = excluded.PTimeSum, MTime = excluded.MTime');
+
my $mergefunc = sub {
- my ($ldb, $ref, $cnewref, $coldref) = @_;
-
- my $sth = $ldb->prepare (
- "UPDATE Domainstat " .
- "SET CountIn = $ref->{countin}, CountOut = $ref->{countout}, " .
- "BytesIn = $ref->{bytesin}, BytesOut = $ref->{bytesout}, " .
- "VirusIn = $ref->{virusin}, VirusOut = $ref->{virusout}, " .
- "SpamIn = $ref->{spamin}, SpamOut = $ref->{spamout}, " .
- "BouncesIn = $ref->{bouncesin}, BouncesOut = $ref->{bouncesout}, " .
- "PTimeSum = $ref->{ptimesum}, MTime = $ref->{mtime} " .
- "WHERE Time = ? AND Domain = ?");
-
- $sth->execute($ref->{time}, $ref->{domain});
-
- my $rows = $sth->rows;
-
- $sth->finish();
-
- if ($rows) {
- $$coldref++;
- } else {
- $ldb->do(
- "INSERT INTO Domainstat " .
- "(Time,Domain,CountIn,CountOut,BytesIn,BytesOut,VirusIn,VirusOut,SpamIn,SpamOut," .
- "BouncesIn,BouncesOut,PTimeSum,Mtime) " .
- "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)", undef,
- $ref->{time}, $ref->{domain}, $ref->{countin}, $ref->{countout},
- $ref->{bytesin}, $ref->{bytesout},
- $ref->{virusin}, $ref->{virusout}, $ref->{spamin}, $ref->{spamout},
- $ref->{bouncesin}, $ref->{bouncesout}, $ref->{ptimesum}, $ref->{mtime});
- $$cnewref++;
- }
+ my ($ref) = @_;
+
+ $merge_sth->execute(
+ $ref->{time}, $ref->{domain}, $ref->{countin}, $ref->{countout},
+ $ref->{bytesin}, $ref->{bytesout},
+ $ref->{virusin}, $ref->{virusout}, $ref->{spamin}, $ref->{spamout},
+ $ref->{bouncesin}, $ref->{bouncesout}, $ref->{ptimesum}, $ref->{mtime});
};
- return sync_generic_mtime_db($dbh, $rdb, $ni, 'DomainStat', $selectfunc, $mergefunc);
+ return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'DomainStat', $selectfunc, $mergefunc);
}
sub sync_dailystat_db {
my $selectfunc = sub {
my ($ctime, $lastmt) = @_;
return "SELECT * from DailyStat WHERE mtime >= $lastmt";
- };
+ };
+
+ my $merge_sth = $dbh->prepare(
+ 'INSERT INTO DailyStat ' .
+ '(Time,CountIn,CountOut,BytesIn,BytesOut,VirusIn,VirusOut,SpamIn,SpamOut,' .
+ 'BouncesIn,BouncesOut,GreylistCount,SPFCount,RBLCount,PTimeSum,Mtime) ' .
+ 'VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) ' .
+ 'ON CONFLICT (Time) DO UPDATE SET ' .
+ 'CountIn = excluded.CountIn, CountOut = excluded.CountOut, ' .
+ 'BytesIn = excluded.BytesIn, BytesOut = excluded.BytesOut, ' .
+ 'VirusIn = excluded.VirusIn, VirusOut = excluded.VirusOut, ' .
+ 'SpamIn = excluded.SpamIn, SpamOut = excluded.SpamOut, ' .
+ 'BouncesIn = excluded.BouncesIn, BouncesOut = excluded.BouncesOut, ' .
+ 'GreylistCount = excluded.GreylistCount, SPFCount = excluded.SpfCount, ' .
+ 'RBLCount = excluded.RBLCount, ' .
+ 'PTimeSum = excluded.PTimeSum, MTime = excluded.MTime');
my $mergefunc = sub {
- my ($ldb, $ref, $cnewref, $coldref) = @_;
-
- my $sth = $ldb->prepare(
- "UPDATE DailyStat " .
- "SET CountIn = $ref->{countin}, CountOut = $ref->{countout}, " .
- "BytesIn = $ref->{bytesin}, BytesOut = $ref->{bytesout}, " .
- "VirusIn = $ref->{virusin}, VirusOut = $ref->{virusout}, " .
- "SpamIn = $ref->{spamin}, SpamOut = $ref->{spamout}, " .
- "BouncesIn = $ref->{bouncesin}, BouncesOut = $ref->{bouncesout}, " .
- "GreylistCount = $ref->{greylistcount}, SPFCount = $ref->{spfcount}, " .
- "RBLCount = $ref->{rblcount}, " .
- "PTimeSum = $ref->{ptimesum}, MTime = $ref->{mtime} " .
- "WHERE Time = ?");
-
- $sth->execute($ref->{time});
-
- my $rows = $sth->rows;
-
- $sth->finish();
-
- if ($rows) {
- $$coldref++;
- } else {
- $ldb->do (
- "INSERT INTO DailyStat " .
- "(Time,CountIn,CountOut,BytesIn,BytesOut,VirusIn,VirusOut,SpamIn,SpamOut," .
- "BouncesIn,BouncesOut,GreylistCount,SPFCount,RBLCount,PTimeSum,Mtime) " .
- "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", undef,
- $ref->{time}, $ref->{countin}, $ref->{countout},
- $ref->{bytesin}, $ref->{bytesout},
- $ref->{virusin}, $ref->{virusout}, $ref->{spamin}, $ref->{spamout},
- $ref->{bouncesin}, $ref->{bouncesout}, $ref->{greylistcount},
- $ref->{spfcount}, $ref->{rblcount}, $ref->{ptimesum}, $ref->{mtime});
- $$cnewref++;
- }
+ my ($ref) = @_;
+
+ $merge_sth->execute(
+ $ref->{time}, $ref->{countin}, $ref->{countout},
+ $ref->{bytesin}, $ref->{bytesout},
+ $ref->{virusin}, $ref->{virusout}, $ref->{spamin}, $ref->{spamout},
+ $ref->{bouncesin}, $ref->{bouncesout}, $ref->{greylistcount},
+ $ref->{spfcount}, $ref->{rblcount}, $ref->{ptimesum}, $ref->{mtime});
};
- return sync_generic_mtime_db($dbh, $rdb, $ni, 'DailyStat', $selectfunc, $mergefunc);
+ return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'DailyStat', $selectfunc, $mergefunc);
}
sub sync_virusinfo_db {
return "SELECT * from VirusInfo WHERE mtime >= $lastmt";
};
- my $mergefunc = sub {
- my ($ldb, $ref, $cnewref, $coldref) = @_;
-
- my $sth = $ldb->prepare(
- "UPDATE VirusInfo SET Count = ? , MTime = ? " .
- "WHERE Time = ? AND Name = ?");
+ my $merge_sth = $dbh->prepare(
+ 'INSERT INTO VirusInfo (Time,Name,Count,MTime) ' .
+ 'VALUES (?,?,?,?) ' .
+ 'ON CONFLICT (Time,Name) DO UPDATE SET ' .
+ 'Count = excluded.Count , MTime = excluded.MTime');
- $sth->execute($ref->{count}, $ref->{mtime}, $ref->{time}, $ref->{name});
-
- my $rows = $sth->rows;
-
- $sth->finish ();
+ my $mergefunc = sub {
+ my ($ref) = @_;
- if ($rows) {
- $$coldref++;
- } else {
- $ldb->do ("INSERT INTO VirusInfo (Time,Name,Count,MTime) " .
- "VALUES (?,?,?,?)", undef,
- $ref->{time}, $ref->{name}, $ref->{count}, $ref->{mtime});
- $$cnewref++;
- }
+ $merge_sth->execute($ref->{time}, $ref->{name}, $ref->{count}, $ref->{mtime});
};
- return sync_generic_mtime_db($dbh, $rdb, $ni, 'VirusInfo', $selectfunc, $mergefunc);
+ return $sync_generic_mtime_db->($dbh, $rdb, $ni, 'VirusInfo', $selectfunc, $mergefunc);
}
sub sync_deleted_nodes_from_master {
my $rsynctime = 0;
my $cid_hash = {}; # fast lookup
- foreach my $ni (@{$cinfo->{nodes}}) {
+ foreach my $ni (values %{$cinfo->{ids}}) {
$cid_hash->{$ni->{cid}} = $ni;
}
my $spooldir = $PMG::MailQueue::spooldir;
- for (my $rcid = 1; $rcid <= $cinfo->{maxcid}; $rcid++) {
+ my $maxcid = $cinfo->{master}->{maxcid} // 0;
+
+ for (my $rcid = 1; $rcid <= $maxcid; $rcid++) {
next if $cid_hash->{$rcid};
my $done_marker = "$spooldir/cluster/$rcid/.synced-deleted-node";
projects / pmg-api.git / blobdiff