PMG/Service/pmgtunnel.pm

   1 package PMG::Service::pmgtunnel;
   2
   3 use strict;
   4 use warnings;
   5 use Data::Dumper;
   6 use Time::HiRes qw (gettimeofday);
   7
   8 use PVE::SafeSyslog;
   9 use PVE::Tools qw(extract_param);
  10 use PVE::INotify;
  11 use PVE::Daemon;
  12
  13 use PMG::RESTEnvironment;
  14 use PMG::DBTools;
  15 use PMG::RuleDB;
  16 use PMG::Cluster;
  17 use PMG::ClusterConfig;
  18 use PMG::Statistic;
  19
  20 use base qw(PVE::Daemon);
  21
  22 my $cmdline = [$0, @ARGV];
  23
  24 my %daemon_options = (restart_on_error => 5, stop_wait_time => 5);
  25
  26 my $daemon = __PACKAGE__->new('pmgtunnel', $cmdline, %daemon_options);
  27
  28 my $restart_request = 0;
  29 my $next_update = 0;
  30
  31 my $cycle = 0;
  32 my $updatetime = 10;
  33
  34 my $workers = {};
  35 my $delayed_exec = {};
  36 my $startcount = {};
  37
  38 sub finish_children {
  39     while ((my $cpid = waitpid(-1, POSIX::WNOHANG())) > 0) {
  40         if (defined($workers->{$cpid})) {
  41             my $ip = $workers->{$cpid}->{ip};
  42             my $cid = $workers->{$cpid}->{cid};
  43             syslog('err', "tunnel finished $cpid $ip");
  44             $delayed_exec->{$cid} = time + ($startcount->{$cid} > 5 ? 60 : 10);
  45             delete $workers->{$cpid};
  46         }
  47     }
  48 }
  49
  50 sub start_tunnels {
  51     my ($self, $cinfo) = @_;
  52
  53     my $role = $cinfo->{local}->{type} // '-';
  54     return if $role eq '-';
  55
  56     foreach my $cid (keys %{$cinfo->{ids}}) {
  57         my $ni = $cinfo->{ids}->{$cid};
  58         next if $ni->{ip} eq $cinfo->{local}->{ip}; # just to be sure
  59
  60         my $dbport = $cinfo->{dbport}->{$cid};
  61         next if !$dbport; # just to be sure
  62
  63         my $running;
  64         foreach my $cpid (keys %$workers) {
  65             $running = 1 if $workers->{$cpid}->{ip} eq  $ni->{ip};
  66         }
  67         next if $running;
  68
  69         if ($delayed_exec->{$cid} && (time < $delayed_exec->{$cid})) {
  70             next;
  71         }
  72         $delayed_exec->{$cid} = 0;
  73         $startcount->{$cid}++;
  74
  75         my $pid = fork;
  76
  77         if (!defined ($pid)) {
  78
  79             syslog('err', "can't fork tunnel");
  80
  81         } elsif($pid) { # parent
  82
  83             $workers->{$pid}->{ip} = $ni->{ip};
  84             $workers->{$pid}->{cid} = $cid;
  85             $workers->{$pid}->{dbport} = $dbport;
  86
  87             if ($startcount->{$cid} > 1) {
  88                 syslog('info', "restarting crashed tunnel $pid $ni->{ip}");
  89             } else {
  90                 syslog('info', "starting tunnel $pid $ni->{ip}");
  91             }
  92
  93         } else { # child
  94
  95             $self->after_fork_cleanup();
  96
  97             # make sure we use ipv4 127.0.0.1 (instead of ipv6 :::1)
  98             exec('/usr/bin/ssh', '-N', '-o', 'BatchMode=yes',
  99                  '-o', "HostKeyAlias=$ni->{name}",
 100                  '-L', "$dbport:/var/run/postgresql/.s.PGSQL.5432",
 101                  $ni->{ip});
 102             exit (0);
 103         }
 104     }
 105 }
 106
 107 sub purge_tunnels {
 108     my ($self, $cinfo) = @_;
 109
 110     foreach my $cpid (keys %$workers) {
 111         my $ip = $workers->{$cpid}->{ip};
 112         my $dbport = $workers->{$cpid}->{dbport};
 113         my $cid = $workers->{$cpid}->{cid};
 114
 115         my $found;
 116         foreach my $ni (values %{$cinfo->{ids}}) {
 117             my $ni_dbport = $cinfo->{dbport}->{$ni->{cid}};
 118             $found = 1 if (($ni->{ip} eq $ip) && ($ni_dbport eq $dbport));
 119         }
 120
 121         my $role = $cinfo->{local}->{type} // '-';
 122         $found = 0 if $role eq '-';
 123
 124         if (!$found) {
 125             syslog ('info', "trying to finish tunnel $cpid $ip");
 126             kill(15, $cpid);
 127             $delayed_exec->{$cid} = time + ($startcount->{$cid} > 5 ? 60 : 10);
 128             delete $workers->{$cpid};
 129         }
 130     }
 131 }
 132
 133 sub init {
 134     # syslog('INIT');
 135 }
 136
 137 sub shutdown {
 138     my ($self) = @_;
 139
 140     syslog('info' , "server closing");
 141
 142     foreach my $cpid (keys %$workers) {
 143         if (kill (15, $cpid) || ! kill(0, $cpid)) {
 144             my $ip = $workers->{$cpid}->{ip};
 145             delete $workers->{$cpid};
 146             syslog ('info', "successfully deleted tunnel $cpid $ip");
 147         }
 148     }
 149
 150     # wait for children
 151     1 while (waitpid(-1, POSIX::WNOHANG()) > 0);
 152
 153  #   $self->exit_daemon(0);
 154 }
 155
 156 sub hup {
 157     my ($self) = @_;
 158
 159     $restart_request = 1;
 160 }
 161
 162
 163
 164 sub run {
 165     my ($self) = @_;
 166
 167     local $SIG{CHLD} = \&finish_children;
 168
 169     for (;;) { # forever
 170
 171         $next_update = time() + $updatetime;
 172
 173         eval {
 174             my $cinfo = PMG::ClusterConfig->new(); # reload
 175             $self->purge_tunnels($cinfo);
 176             $self->start_tunnels($cinfo);
 177         };
 178
 179         if (my $err = $@) {
 180
 181             syslog('err', "status update error: $err");
 182         }
 183
 184         my $wcount = 0;
 185         while ((time() < $next_update) &&
 186                ($wcount < $updatetime) && # protect against time wrap
 187                !$restart_request && !$self->{terminate}) {
 188
 189             finish_children();
 190
 191             $wcount++; sleep (1);
 192         };
 193
 194         last if $self->{terminate};
 195
 196         $self->restart_daemon() if $restart_request;
 197     }
 198 }
 199
 200 __PACKAGE__->register_method ({
 201     name => 'status',
 202     path => 'status',
 203     method => 'GET',
 204     description => "Print cluster tunnel status.",
 205     parameters => {
 206         additionalProperties => 0,
 207         properties => {},
 208     },
 209     returns => { type => 'null' },
 210     code => sub {
 211         my ($param) = @_;
 212
 213         my $status = $daemon->running() ? 'running' : 'stopped';
 214         print "$status\n";
 215
 216         return undef;
 217     }});
 218
 219
 220 $daemon->register_start_command("Start the Cluster Tunnel Daemon");
 221 $daemon->register_stop_command("Stop the Cluster Tunnel Daemon");
 222 $daemon->register_restart_command(1, "Restart the Cluster Tunnel Daemon");
 223
 224 our $cmddef = {
 225     start => [ __PACKAGE__, 'start', []],
 226     restart => [ __PACKAGE__, 'restart', []],
 227     stop => [ __PACKAGE__, 'stop', []],
 228     status => [ __PACKAGE__, 'status', []]
 229 };