src/PVE/RESTEnvironment.pm

   1 package PVE::RESTEnvironment;
   2
   3 # NOTE: you can/should provide your own specialice class, and
   4 # use this a bas class (as example see PVE::RPCEnvironment).
   5
   6 # we use this singleton class to pass RPC related environment values
   7
   8 use strict;
   9 use warnings;
  10 use POSIX qw(:sys_wait_h EINTR);
  11 use IO::Handle;
  12 use IO::File;
  13 use IO::Select;
  14 use Fcntl qw(:flock);
  15 use PVE::Exception qw(raise raise_perm_exc);
  16 use PVE::SafeSyslog;
  17 use PVE::Tools;
  18 use PVE::INotify;
  19 use PVE::ProcFSTools;
  20
  21
  22 my $rest_env;
  23
  24 # save $SIG{CHLD} handler implementation.
  25 # simply set $SIG{CHLD} = $worker_reaper;
  26 # and register forked processes with &$register_worker(pid)
  27 # Note: using $SIG{CHLD} = 'IGNORE' or $SIG{CHLD} = sub { wait (); } or ...
  28 # has serious side effects, because perls built in system() and open()
  29 # functions can't get the correct exit status of a child. So we cant use
  30 # that (also see perlipc)
  31
  32 my $WORKER_PIDS;
  33 my $WORKER_FLAG = 0;
  34
  35 my $log_task_result = sub {
  36     my ($upid, $user, $status) = @_;
  37
  38     return if !$rest_env;
  39
  40     my $msg = 'successful';
  41     my $pri = 'info';
  42     if ($status != 0) {
  43         my $ec = $status >> 8;
  44         my $ic = $status & 255;
  45         $msg = $ec ? "failed ($ec)" : "interrupted ($ic)";
  46         $pri = 'err';
  47     }
  48
  49     my $tlist = $rest_env->active_workers($upid);
  50     $rest_env->broadcast_tasklist($tlist);
  51
  52     my $task;
  53     foreach my $t (@$tlist) {
  54         if ($t->{upid} eq $upid) {
  55             $task = $t;
  56             last;
  57         }
  58     }
  59     if ($task && $task->{status}) {
  60         $msg = $task->{status};
  61     }
  62
  63     $rest_env->log_cluster_msg($pri, $user, "end task $upid $msg");
  64 };
  65
  66 my $worker_reaper = sub {
  67     local $!; local $?;
  68     foreach my $pid (keys %$WORKER_PIDS) {
  69         my $waitpid = waitpid ($pid, WNOHANG);
  70         if (defined($waitpid) && ($waitpid == $pid)) {
  71             my $info = $WORKER_PIDS->{$pid};
  72             if ($info && $info->{upid} && $info->{user}) {
  73                 &$log_task_result($info->{upid}, $info->{user}, $?);
  74             }
  75             delete ($WORKER_PIDS->{$pid});
  76         }
  77     }
  78 };
  79
  80 my $register_worker = sub {
  81     my ($pid, $user, $upid) = @_;
  82
  83     return if !$pid;
  84
  85     # do not register if already finished
  86     my $waitpid = waitpid ($pid, WNOHANG);
  87     if (defined($waitpid) && ($waitpid == $pid)) {
  88         delete ($WORKER_PIDS->{$pid});
  89         return;
  90     }
  91
  92     $WORKER_PIDS->{$pid} = {
  93         user => $user,
  94         upid => $upid,
  95     };
  96 };
  97
  98 # initialize environment - must be called once at program startup
  99 sub init {
 100     my ($class, $type, %params) = @_;
 101
 102     $class = ref($class) || $class;
 103
 104     die "already initialized" if $rest_env;
 105
 106     die "unknown environment type"
 107         if !$type || $type !~ m/^(cli|pub|priv|ha)$/;
 108
 109     $SIG{CHLD} = $worker_reaper;
 110
 111     # environment types
 112     # cli  ... command started fron command line
 113     # pub  ... access from public server (apache)
 114     # priv ... access from private server (pvedaemon)
 115     # ha   ... access from HA resource manager agent (rgmanager)
 116
 117     my $self = { type => $type };
 118
 119     bless $self, $class;
 120
 121     foreach my $p (keys %params) {
 122         if ($p eq 'atfork') {
 123             $self->{$p} = $params{$p};
 124         } else {
 125             die "unknown option '$p'";
 126         }
 127     }
 128
 129     $rest_env = $self;
 130
 131     my ($sysname, $nodename) = POSIX::uname();
 132
 133     $nodename =~ s/\..*$//; # strip domain part, if any
 134
 135     $self->{nodename} = $nodename;
 136
 137     return $self;
 138 };
 139
 140 # convenience function for command line tools
 141 sub setup_default_cli_env {
 142     my ($class, $username) = @_;
 143
 144     $class = ref($class) || $class;
 145
 146     $username //= 'root@pam';
 147
 148     PVE::INotify::inotify_init();
 149
 150     my $rpcenv = $class->init('cli');
 151     $rpcenv->init_request();
 152     $rpcenv->set_language($ENV{LANG});
 153     $rpcenv->set_user($username);
 154
 155     die "please run as root\n"
 156         if ($username eq 'root@pam') && ($> != 0);
 157 }
 158
 159 # get the singleton
 160 sub get {
 161
 162     die "REST environment not initialized" if !$rest_env;
 163
 164     return $rest_env;
 165 }
 166
 167 sub set_client_ip {
 168     my ($self, $ip) = @_;
 169
 170     $self->{client_ip} = $ip;
 171 }
 172
 173 sub get_client_ip {
 174     my ($self) = @_;
 175
 176     return $self->{client_ip};
 177 }
 178
 179 sub set_result_attrib {
 180     my ($self, $key, $value) = @_;
 181
 182     $self->{result_attributes}->{$key} = $value;
 183 }
 184
 185 sub get_result_attrib {
 186     my ($self, $key) = @_;
 187
 188     return $self->{result_attributes}->{$key};
 189 }
 190
 191 sub set_language {
 192     my ($self, $lang) = @_;
 193
 194     # fixme: initialize I18N
 195
 196     $self->{language} = $lang;
 197 }
 198
 199 sub get_language {
 200     my ($self) = @_;
 201
 202     return $self->{language};
 203 }
 204
 205 sub set_user {
 206     my ($self, $user) = @_;
 207
 208     $self->{user} = $user;
 209 }
 210
 211 sub get_user {
 212     my ($self, $noerr) = @_;
 213
 214     return $self->{user} if defined($self->{user}) || $noerr;
 215
 216     die "user name not set\n";
 217 }
 218
 219 sub is_worker {
 220     my ($class) = @_;
 221
 222     return $WORKER_FLAG;
 223 }
 224
 225 # read/update list of active workers
 226 # we move all finished tasks to the archive index,
 227 # but keep aktive and most recent task in the active file.
 228 # $nocheck ... consider $new_upid still running (avoid that
 229 # we try to read the reult to early.
 230 sub active_workers  {
 231     my ($self, $new_upid, $nocheck) = @_;
 232
 233     my $lkfn = "/var/log/pve/tasks/.active.lock";
 234
 235     my $timeout = 10;
 236
 237     my $code = sub {
 238
 239         my $tasklist = PVE::INotify::read_file('active');
 240
 241         my @ta;
 242         my $tlist = [];
 243         my $thash = {}; # only list task once
 244
 245         my $check_task = sub {
 246             my ($task, $running) = @_;
 247
 248             if ($running || PVE::ProcFSTools::check_process_running($task->{pid}, $task->{pstart})) {
 249                 push @$tlist, $task;
 250             } else {
 251                 delete $task->{pid};
 252                 push @ta, $task;
 253             }
 254             delete $task->{pstart};
 255         };
 256
 257         foreach my $task (@$tasklist) {
 258             my $upid = $task->{upid};
 259             next if $thash->{$upid};
 260             $thash->{$upid} = $task;
 261             &$check_task($task);
 262         }
 263
 264         if ($new_upid && !(my $task = $thash->{$new_upid})) {
 265             $task = PVE::Tools::upid_decode($new_upid);
 266             $task->{upid} = $new_upid;
 267             $thash->{$new_upid} = $task;
 268             &$check_task($task, $nocheck);
 269         }
 270
 271
 272         @ta = sort { $b->{starttime} cmp $a->{starttime} } @ta;
 273
 274         my $save = defined($new_upid);
 275
 276         foreach my $task (@ta) {
 277             next if $task->{endtime};
 278             $task->{endtime} = time();
 279             $task->{status} = PVE::Tools::upid_read_status($task->{upid});
 280             $save = 1;
 281         }
 282
 283         my $archive = '';
 284         my @arlist = ();
 285         foreach my $task (@ta) {
 286             if (!$task->{saved}) {
 287                 $archive .= sprintf("%s %08X %s\n", $task->{upid}, $task->{endtime}, $task->{status});
 288                 $save = 1;
 289                 push @arlist, $task;
 290                 $task->{saved} = 1;
 291             }
 292         }
 293
 294         if ($archive) {
 295             my $size = 0;
 296             my $filename = "/var/log/pve/tasks/index";
 297             eval {
 298                 my $fh = IO::File->new($filename, '>>', 0644) ||
 299                     die "unable to open file '$filename' - $!\n";
 300                 PVE::Tools::safe_print($filename, $fh, $archive);
 301                 $size = -s $fh;
 302                 close($fh) ||
 303                     die "unable to close file '$filename' - $!\n";
 304             };
 305             my $err = $@;
 306             if ($err) {
 307                 syslog('err', $err);
 308                 foreach my $task (@arlist) { # mark as not saved
 309                     $task->{saved} = 0;
 310                 }
 311             }
 312             my $maxsize = 50000; # about 1000 entries
 313             if ($size > $maxsize) {
 314                 rename($filename, "$filename.1");
 315             }
 316         }
 317
 318         # we try to reduce the amount of data
 319         # list all running tasks and task and a few others
 320         # try to limit to 25 tasks
 321         my $ctime = time();
 322         my $max = 25 - scalar(@$tlist);
 323         foreach my $task (@ta) {
 324             last if $max <= 0;
 325             push @$tlist, $task;
 326             $max--;
 327         }
 328
 329         PVE::INotify::write_file('active', $tlist) if $save;
 330
 331         return $tlist;
 332     };
 333
 334     my $res = PVE::Tools::lock_file($lkfn, $timeout, $code);
 335     die $@ if $@;
 336
 337     return $res;
 338 }
 339
 340 my $kill_process_group = sub {
 341     my ($pid, $pstart) = @_;
 342
 343     # send kill to process group (negative pid)
 344     my $kpid = -$pid;
 345
 346     # always send signal to all pgrp members
 347     kill(15, $kpid); # send TERM signal
 348
 349     # give max 5 seconds to shut down
 350     for (my $i = 0; $i < 5; $i++) {
 351         return if !PVE::ProcFSTools::check_process_running($pid, $pstart);
 352         sleep (1);
 353     }
 354
 355     # to be sure
 356     kill(9, $kpid);
 357 };
 358
 359 sub check_worker {
 360     my ($self, $upid, $killit) = @_;
 361
 362     my $task = PVE::Tools::upid_decode($upid);
 363
 364     my $running = PVE::ProcFSTools::check_process_running($task->{pid}, $task->{pstart});
 365
 366     return 0 if !$running;
 367
 368     if ($killit) {
 369         &$kill_process_group($task->{pid});
 370         return 0;
 371     }
 372
 373     return 1;
 374 }
 375
 376 # start long running workers
 377 # STDIN is redirected to /dev/null
 378 # STDOUT,STDERR are redirected to the filename returned by upid_decode
 379 # NOTE: we simulate running in foreground if ($self->{type} eq 'cli')
 380 sub fork_worker {
 381     my ($self, $dtype, $id, $user, $function, $background) = @_;
 382
 383     $dtype = 'unknown' if !defined ($dtype);
 384     $id = '' if !defined ($id);
 385
 386     $user = 'root@pve' if !defined ($user);
 387
 388     my $sync = ($self->{type} eq 'cli' && !$background) ? 1 : 0;
 389
 390     local $SIG{INT} =
 391         local $SIG{QUIT} =
 392         local $SIG{PIPE} =
 393         local $SIG{TERM} = 'IGNORE';
 394
 395     my $starttime = time ();
 396
 397     my @psync = POSIX::pipe();
 398     my @csync = POSIX::pipe();
 399
 400     my $node = $self->{nodename};
 401
 402     my $cpid = fork();
 403     die "unable to fork worker - $!" if !defined($cpid);
 404
 405     my $workerpuid = $cpid ? $cpid : $$;
 406
 407     my $pstart = PVE::ProcFSTools::read_proc_starttime($workerpuid) ||
 408         die "unable to read process start time";
 409
 410     my $upid = PVE::Tools::upid_encode ({
 411         node => $node, pid => $workerpuid, pstart => $pstart,
 412         starttime => $starttime, type => $dtype, id => $id, user => $user });
 413
 414     my $outfh;
 415
 416     if (!$cpid) { # child
 417
 418         $0 = "task $upid";
 419         $WORKER_FLAG = 1;
 420
 421         $SIG{INT} = $SIG{QUIT} = $SIG{TERM} = sub { die "received interrupt\n"; };
 422
 423         $SIG{CHLD} = $SIG{PIPE} = 'DEFAULT';
 424
 425         # set sess/process group - we want to be able to kill the
 426         # whole process group
 427         POSIX::setsid();
 428
 429         POSIX::close ($psync[0]);
 430         POSIX::close ($csync[1]);
 431
 432         $outfh = $sync ? $psync[1] : undef;
 433
 434         eval {
 435             PVE::INotify::inotify_close();
 436
 437             if (my $atfork = $self->{atfork}) {
 438                 &$atfork();
 439             }
 440
 441             # same algorythm as used inside SA
 442             # STDIN = /dev/null
 443             my $fd = fileno (STDIN);
 444
 445             if (!$sync) {
 446                 close STDIN;
 447                 POSIX::close(0) if $fd != 0;
 448
 449                 die "unable to redirect STDIN - $!"
 450                     if !open(STDIN, "</dev/null");
 451
 452                 $outfh = PVE::Tools::upid_open($upid);
 453             }
 454
 455
 456             # redirect STDOUT
 457             $fd = fileno(STDOUT);
 458             close STDOUT;
 459             POSIX::close (1) if $fd != 1;
 460
 461             die "unable to redirect STDOUT - $!"
 462                 if !open(STDOUT, ">&", $outfh);
 463
 464             STDOUT->autoflush (1);
 465
 466             #  redirect STDERR to STDOUT
 467             $fd = fileno (STDERR);
 468             close STDERR;
 469             POSIX::close(2) if $fd != 2;
 470
 471             die "unable to redirect STDERR - $!"
 472                 if !open(STDERR, ">&1");
 473
 474             STDERR->autoflush(1);
 475         };
 476         if (my $err = $@) {
 477             my $msg =  "ERROR: $err";
 478             POSIX::write($psync[1], $msg, length ($msg));
 479             POSIX::close($psync[1]);
 480             POSIX::_exit(1);
 481             kill(-9, $$);
 482         }
 483
 484         # sync with parent (signal that we are ready)
 485         if ($sync) {
 486             print "$upid\n";
 487         } else {
 488             POSIX::write($psync[1], $upid, length ($upid));
 489             POSIX::close($psync[1]);
 490         }
 491
 492         my $readbuf = '';
 493         # sync with parent (wait until parent is ready)
 494         POSIX::read($csync[0], $readbuf, 4096);
 495         die "parent setup error\n" if $readbuf ne 'OK';
 496
 497         if ($self->{type} eq 'ha') {
 498             print "task started by HA resource agent\n";
 499         }
 500         eval { &$function($upid); };
 501         my $err = $@;
 502         if ($err) {
 503             chomp $err;
 504             $err =~ s/\n/ /mg;
 505             syslog('err', $err);
 506             print STDERR "TASK ERROR: $err\n";
 507             POSIX::_exit(-1);
 508         } else {
 509             print STDERR "TASK OK\n";
 510             POSIX::_exit(0);
 511         }
 512         kill(-9, $$);
 513     }
 514
 515     # parent
 516
 517     POSIX::close ($psync[1]);
 518     POSIX::close ($csync[0]);
 519
 520     my $readbuf = '';
 521     # sync with child (wait until child starts)
 522     POSIX::read($psync[0], $readbuf, 4096);
 523
 524     if (!$sync) {
 525         POSIX::close($psync[0]);
 526         &$register_worker($cpid, $user, $upid);
 527     } else {
 528         chomp $readbuf;
 529     }
 530
 531     eval {
 532         die "got no worker upid - start worker failed\n" if !$readbuf;
 533
 534         if ($readbuf =~ m/^ERROR:\s*(.+)$/m) {
 535             die "starting worker failed: $1\n";
 536         }
 537
 538         if ($readbuf ne $upid) {
 539             die "got strange worker upid ('$readbuf' != '$upid') - start worker failed\n";
 540         }
 541
 542         if ($sync) {
 543             $outfh = PVE::Tools::upid_open($upid);
 544         }
 545     };
 546     my $err = $@;
 547
 548     if (!$err) {
 549         my $msg = 'OK';
 550         POSIX::write($csync[1], $msg, length ($msg));
 551         POSIX::close($csync[1]);
 552
 553     } else {
 554         POSIX::close($csync[1]);
 555         kill(-9, $cpid); # make sure it gets killed
 556         die $err;
 557     }
 558
 559     $self->log_cluster_msg('info', $user, "starting task $upid");
 560
 561     my $tlist = $self->active_workers($upid, $sync);
 562     $self->broadcast_tasklist($tlist);
 563
 564     my $res = 0;
 565
 566     if ($sync) {
 567         my $count;
 568         my $outbuf = '';
 569         my $int_count = 0;
 570         eval {
 571             local $SIG{INT} = local $SIG{QUIT} = local $SIG{TERM} = sub {
 572                 # always send signal to all pgrp members
 573                 my $kpid = -$cpid;
 574                 if ($int_count < 3) {
 575                     kill(15, $kpid); # send TERM signal
 576                 } else {
 577                     kill(9, $kpid); # send KILL signal
 578                 }
 579                 $int_count++;
 580             };
 581             local $SIG{PIPE} = sub { die "broken pipe\n"; };
 582
 583             my $select = new IO::Select;
 584             my $fh = IO::Handle->new_from_fd($psync[0], 'r');
 585             $select->add($fh);
 586
 587             while ($select->count) {
 588                 my @handles = $select->can_read(1);
 589                 if (scalar(@handles)) {
 590                     my $count = sysread ($handles[0], $readbuf, 4096);
 591                     if (!defined ($count)) {
 592                         my $err = $!;
 593                         die "sync pipe read error: $err\n";
 594                     }
 595                     last if $count == 0; # eof
 596
 597                     $outbuf .= $readbuf;
 598                     while ($outbuf =~ s/^(([^\010\r\n]*)(\r|\n|(\010)+|\r\n))//s) {
 599                         my $line = $1;
 600                         my $data = $2;
 601                         if ($data =~ m/^TASK OK$/) {
 602                             # skip
 603                         } elsif ($data =~ m/^TASK ERROR: (.+)$/) {
 604                             print STDERR "$1\n";
 605                         } else {
 606                             print $line;
 607                         }
 608                         if ($outfh) {
 609                             print $outfh $line;
 610                             $outfh->flush();
 611                         }
 612                     }
 613                 } else {
 614                     # some commands daemonize without closing stdout
 615                     last if !PVE::ProcFSTools::check_process_running($cpid);
 616                 }
 617             }
 618         };
 619         my $err = $@;
 620
 621         POSIX::close($psync[0]);
 622
 623         if ($outbuf) { # just to be sure
 624             print $outbuf;
 625             if ($outfh) {
 626                 print $outfh $outbuf;
 627             }
 628         }
 629
 630         if ($err) {
 631             $err =~ s/\n/ /mg;
 632             print STDERR "$err\n";
 633             if ($outfh) {
 634                 print $outfh "TASK ERROR: $err\n";
 635             }
 636         }
 637
 638         &$kill_process_group($cpid, $pstart); # make sure it gets killed
 639
 640         close($outfh);
 641
 642         waitpid($cpid, 0);
 643         $res = $?;
 644         &$log_task_result($upid, $user, $res);
 645     }
 646
 647     return wantarray ? ($upid, $res) : $upid;
 648 }
 649
 650 # Abstract function
 651
 652 sub log_cluster_msg {
 653     my ($self, $pri, $user, $msg) = @_;
 654
 655     syslog($pri, "%s", $msg);
 656
 657     # PVE::Cluster::log_msg($pri, $user, $msg);
 658 }
 659
 660 sub broadcast_tasklist {
 661     my ($self, $tlist) = @_;
 662
 663     # PVE::Cluster::broadcast_tasklist($tlist);
 664 }
 665
 666 sub check_api2_permissions {
 667     my ($self, $perm, $username, $param) = @_;
 668
 669     return 1 if !$username && $perm->{user} eq 'world';
 670
 671     raise_perm_exc("user != null") if !$username;
 672
 673     return 1 if $username eq 'root@pam';
 674
 675     raise_perm_exc('user != root@pam') if !$perm;
 676
 677     return 1 if $perm->{user} && $perm->{user} eq 'all';
 678
 679     ##return $self->exec_api2_perm_check($perm->{check}, $username, $param)
 680     ##if $perm->{check};
 681
 682     raise_perm_exc();
 683 }
 684
 685 # init_request - should be called before each REST/CLI request
 686 sub init_request {
 687     my ($self, %params) = @_;
 688
 689     $self->{result_attributes} = {}
 690
 691     # if you nedd more, implement in subclass
 692 }
 693
 694 1;