close #1887: register tls_policy file with reader/writer

[pmg-api.git] / PMG / Service / pmgmirror.pm

package PMG::Service::pmgmirror;

use strict;
use warnings;
use Data::Dumper;
use Time::HiRes qw (gettimeofday tv_interval);

use PVE::SafeSyslog;
use PVE::Tools qw(extract_param);
use PVE::INotify;
use PVE::Daemon;
use PVE::ProcFSTools;

use PMG::Ticket;
use PMG::RESTEnvironment;
use PMG::DBTools;
use PMG::RuleDB;
use PMG::Cluster;
use PMG::ClusterConfig;
use PMG::Statistic;

use base qw(PVE::Daemon);

my $cmdline = [$0, @ARGV];

my %daemon_options = (restart_on_error => 5, stop_wait_time => 5);

my $daemon = __PACKAGE__->new('pmgmirror', $cmdline, %daemon_options);

my $restart_request = 0;
my $next_update = 0;

my $cycle = 0;
my $updatetime = 60*2;
my $maxtimediff = 5;

my $initial_memory_usage;

sub init {
    # syslog('INIT');
}

sub hup {
    my ($self) = @_;

    $restart_request = 1;
}

sub sync_data_from_node {
    my ($dbh, $rdb, $cinfo, $ni, $ticket, $rsynctime_ref) = @_;

    my $ctime = PMG::DBTools::get_remote_time($rdb);
    my $ltime = time();

    my $timediff = abs($ltime - $ctime);
    if ($timediff > $maxtimediff) {
	die "large time difference (> $timediff seconds) - not syncing\n";
    }

    if ($ni->{type} eq 'master') {
	PMG::Cluster::sync_ruledb_from_master($dbh, $rdb, $ni, $ticket);
	PMG::Cluster::sync_deleted_nodes_from_master($dbh, $rdb, $cinfo, $ni, $rsynctime_ref);
    }

    PMG::Cluster::sync_quarantine_db($dbh, $rdb, $ni, $rsynctime_ref);

    PMG::Cluster::sync_greylist_db($dbh, $rdb, $ni);

    PMG::Cluster::sync_userprefs_db($dbh, $rdb, $ni);

    PMG::Cluster::sync_statistic_db($dbh, $rdb, $ni);

    if ($ni->{type} eq 'master') {
	PMG::Cluster::sync_domainstat_db($dbh, $rdb, $ni);

	PMG::Cluster::sync_dailystat_db($dbh, $rdb, $ni);

	PMG::Cluster::sync_virusinfo_db($dbh, $rdb, $ni);
    }

    PMG::Cluster::sync_localstat_db($dbh, $rdb, $ni);
}

sub cluster_sync {

    my $cinfo = PMG::ClusterConfig->new(); # reload
    my $role = $cinfo->{local}->{type} // '-';

    return if $role eq '-';
    return if !$cinfo->{master}; # just to be sure

    my $start_time = [ gettimeofday() ];

    syslog ('info', "starting cluster syncronization");

    my $master_ip = $cinfo->{master}->{ip};
    my $master_name = $cinfo->{master}->{name};

    if ($role ne 'master') {
	PMG::Cluster::sync_config_from_master($master_name, $master_ip);
    }

    my $csynctime = tv_interval($start_time);

    $cinfo = PMG::ClusterConfig->new(); # reload
    $role = $cinfo->{local}->{type} // '-';

    return if $role eq '-';
    return if !$cinfo->{master}; # just to be sure

    my $ticket = PMG::Ticket::assemble_ticket('root@pam');

    my $dbh = PMG::DBTools::open_ruledb();

    my $errcount = 0;

    my $rsynctime = 0;

    my $sync_node = sub {
	my ($ni) = @_;

	my $rdb;
	eval {
	    $rdb = PMG::DBTools::open_ruledb(undef, '/var/run/pmgtunnel', $ni->{cid});
	    sync_data_from_node($dbh, $rdb, $cinfo, $ni, $ticket, \$rsynctime);
	};
	my $err = $@;

	$rdb->disconnect() if $rdb;

	if ($err) {
	    $errcount++;
	    syslog ('err', "database sync '$ni->{name}' failed - $err");
	} else {
	    PMG::DBTools::create_clusterinfo_default($dbh, $ni->{cid}, 'lastsync', 0, undef);
	    PMG::DBTools::write_maxint_clusterinfo($dbh, $ni->{cid}, 'lastsync', time());
	}
    };

    # sync data from master first
    if ($cinfo->{master}->{cid} ne $cinfo->{local}->{cid}) {
	$sync_node->($cinfo->{master});

	# rewrite config after sync from master
	my $cfg = PMG::Config->new();
	my $ruledb = PMG::RuleDB->new($dbh);
	my $rulecache = PMG::RuleCache->new($ruledb);
	$cfg->rewrite_config($rulecache, 1);
    }

    foreach my $ni (values %{$cinfo->{ids}}) {
	next if $ni->{cid} eq $cinfo->{local}->{cid};
	next if $ni->{cid} eq $cinfo->{master}->{cid};
	$sync_node->($ni);
    }

    $dbh->disconnect();

    my $cptime = tv_interval($start_time);

    my $dbtime = $cptime - $rsynctime - $csynctime;

    syslog('info', sprintf("cluster syncronization finished  (%d errors, %.2f seconds " .
			   "(files %.2f, database %.2f, config %.2f))",
			   $errcount, $cptime, $rsynctime, $dbtime, $csynctime));

}

sub run {
    my ($self) = @_;

    for (;;) { # forever

	$next_update = time() + $updatetime;

	eval {
	    # Note: do nothing in first cycle (give pmgtunnel some time to startup)
	    cluster_sync() if $cycle > 0;
	};
	if (my $err = $@) {
	    syslog('err', "sync error: $err");
	}

	$cycle++;

	last if $self->{terminate};

	my $mem = PVE::ProcFSTools::read_memory_usage();

	if (!defined($initial_memory_usage) || ($cycle < 10)) {
	    $initial_memory_usage = $mem->{resident};
	} else {
	    my $diff = $mem->{resident} - $initial_memory_usage;
	    if ($diff > 5*1024*1024) {
		syslog ('info', "restarting server after $cycle cycles to " .
			"reduce memory usage (free $mem->{resident} ($diff) bytes)");
		$self->restart_daemon();
	    }
	}

	my $wcount = 0;
	while ((time() < $next_update) &&
	       ($wcount < $updatetime) && # protect against time wrap
	       !$restart_request && !$self->{terminate}) {

	    $wcount++; sleep (1);
	};

	last if $self->{terminate};

	$self->restart_daemon() if $restart_request;
    }
}

$daemon->register_start_command("Start the Database Mirror Daemon");
$daemon->register_stop_command("Stop the Database Mirror Daemon");
$daemon->register_restart_command(1, "Restart the Database Mirror Daemon");

our $cmddef = {
    start => [ __PACKAGE__, 'start', []],
    restart => [ __PACKAGE__, 'restart', []],
    stop => [ __PACKAGE__, 'stop', []],
};

1;