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 $max = 25 - scalar(@$tlist);
 322         foreach my $task (@ta) {
 323             last if $max <= 0;
 324             push @$tlist, $task;
 325             $max--;
 326         }
 327
 328         PVE::INotify::write_file('active', $tlist) if $save;
 329
 330         return $tlist;
 331     };
 332
 333     my $res = PVE::Tools::lock_file($lkfn, $timeout, $code);
 334     die $@ if $@;
 335
 336     return $res;
 337 }
 338
 339 my $kill_process_group = sub {
 340     my ($pid, $pstart) = @_;
 341
 342     # send kill to process group (negative pid)
 343     my $kpid = -$pid;
 344
 345     # always send signal to all pgrp members
 346     kill(15, $kpid); # send TERM signal
 347
 348     # give max 5 seconds to shut down
 349     for (my $i = 0; $i < 5; $i++) {
 350         return if !PVE::ProcFSTools::check_process_running($pid, $pstart);
 351         sleep (1);
 352     }
 353
 354     # to be sure
 355     kill(9, $kpid);
 356 };
 357
 358 sub check_worker {
 359     my ($self, $upid, $killit) = @_;
 360
 361     my $task = PVE::Tools::upid_decode($upid);
 362
 363     my $running = PVE::ProcFSTools::check_process_running($task->{pid}, $task->{pstart});
 364
 365     return 0 if !$running;
 366
 367     if ($killit) {
 368         &$kill_process_group($task->{pid});
 369         return 0;
 370     }
 371
 372     return 1;
 373 }
 374
 375 # start long running workers
 376 # STDIN is redirected to /dev/null
 377 # STDOUT,STDERR are redirected to the filename returned by upid_decode
 378 # NOTE: we simulate running in foreground if ($self->{type} eq 'cli')
 379 sub fork_worker {
 380     my ($self, $dtype, $id, $user, $function, $background) = @_;
 381
 382     $dtype = 'unknown' if !defined ($dtype);
 383     $id = '' if !defined ($id);
 384
 385     $user = 'root@pve' if !defined ($user);
 386
 387     my $sync = ($self->{type} eq 'cli' && !$background) ? 1 : 0;
 388
 389     local $SIG{INT} =
 390         local $SIG{QUIT} =
 391         local $SIG{PIPE} =
 392         local $SIG{TERM} = 'IGNORE';
 393
 394     my $starttime = time ();
 395
 396     my @psync = POSIX::pipe();
 397     my @csync = POSIX::pipe();
 398
 399     my $node = $self->{nodename};
 400
 401     my $cpid = fork();
 402     die "unable to fork worker - $!" if !defined($cpid);
 403
 404     my $workerpuid = $cpid ? $cpid : $$;
 405
 406     my $pstart = PVE::ProcFSTools::read_proc_starttime($workerpuid) ||
 407         die "unable to read process start time";
 408
 409     my $upid = PVE::Tools::upid_encode ({
 410         node => $node, pid => $workerpuid, pstart => $pstart,
 411         starttime => $starttime, type => $dtype, id => $id, user => $user });
 412
 413     my $outfh;
 414
 415     if (!$cpid) { # child
 416
 417         $0 = "task $upid";
 418         $WORKER_FLAG = 1;
 419
 420         $SIG{INT} = $SIG{QUIT} = $SIG{TERM} = sub { die "received interrupt\n"; };
 421
 422         $SIG{CHLD} = $SIG{PIPE} = 'DEFAULT';
 423
 424         # set sess/process group - we want to be able to kill the
 425         # whole process group
 426         POSIX::setsid();
 427
 428         POSIX::close ($psync[0]);
 429         POSIX::close ($csync[1]);
 430
 431         $outfh = $sync ? $psync[1] : undef;
 432
 433         eval {
 434             PVE::INotify::inotify_close();
 435
 436             if (my $atfork = $self->{atfork}) {
 437                 &$atfork();
 438             }
 439
 440             # same algorythm as used inside SA
 441             # STDIN = /dev/null
 442             my $fd = fileno (STDIN);
 443
 444             if (!$sync) {
 445                 close STDIN;
 446                 POSIX::close(0) if $fd != 0;
 447
 448                 die "unable to redirect STDIN - $!"
 449                     if !open(STDIN, "</dev/null");
 450
 451                 $outfh = PVE::Tools::upid_open($upid);
 452             }
 453
 454
 455             # redirect STDOUT
 456             $fd = fileno(STDOUT);
 457             close STDOUT;
 458             POSIX::close (1) if $fd != 1;
 459
 460             die "unable to redirect STDOUT - $!"
 461                 if !open(STDOUT, ">&", $outfh);
 462
 463             STDOUT->autoflush (1);
 464
 465             #  redirect STDERR to STDOUT
 466             $fd = fileno (STDERR);
 467             close STDERR;
 468             POSIX::close(2) if $fd != 2;
 469
 470             die "unable to redirect STDERR - $!"
 471                 if !open(STDERR, ">&1");
 472
 473             STDERR->autoflush(1);
 474         };
 475         if (my $err = $@) {
 476             my $msg =  "ERROR: $err";
 477             POSIX::write($psync[1], $msg, length ($msg));
 478             POSIX::close($psync[1]);
 479             POSIX::_exit(1);
 480             kill(-9, $$);
 481         }
 482
 483         # sync with parent (signal that we are ready)
 484         if ($sync) {
 485             print "$upid\n";
 486         } else {
 487             POSIX::write($psync[1], $upid, length ($upid));
 488             POSIX::close($psync[1]);
 489         }
 490
 491         my $readbuf = '';
 492         # sync with parent (wait until parent is ready)
 493         POSIX::read($csync[0], $readbuf, 4096);
 494         die "parent setup error\n" if $readbuf ne 'OK';
 495
 496         if ($self->{type} eq 'ha') {
 497             print "task started by HA resource agent\n";
 498         }
 499         eval { &$function($upid); };
 500         my $err = $@;
 501         if ($err) {
 502             chomp $err;
 503             $err =~ s/\n/ /mg;
 504             syslog('err', $err);
 505             print STDERR "TASK ERROR: $err\n";
 506             POSIX::_exit(-1);
 507         } else {
 508             print STDERR "TASK OK\n";
 509             POSIX::_exit(0);
 510         }
 511         kill(-9, $$);
 512     }
 513
 514     # parent
 515
 516     POSIX::close ($psync[1]);
 517     POSIX::close ($csync[0]);
 518
 519     my $readbuf = '';
 520     # sync with child (wait until child starts)
 521     POSIX::read($psync[0], $readbuf, 4096);
 522
 523     if (!$sync) {
 524         POSIX::close($psync[0]);
 525         &$register_worker($cpid, $user, $upid);
 526     } else {
 527         chomp $readbuf;
 528     }
 529
 530     eval {
 531         die "got no worker upid - start worker failed\n" if !$readbuf;
 532
 533         if ($readbuf =~ m/^ERROR:\s*(.+)$/m) {
 534             die "starting worker failed: $1\n";
 535         }
 536
 537         if ($readbuf ne $upid) {
 538             die "got strange worker upid ('$readbuf' != '$upid') - start worker failed\n";
 539         }
 540
 541         if ($sync) {
 542             $outfh = PVE::Tools::upid_open($upid);
 543         }
 544     };
 545     my $err = $@;
 546
 547     if (!$err) {
 548         my $msg = 'OK';
 549         POSIX::write($csync[1], $msg, length ($msg));
 550         POSIX::close($csync[1]);
 551
 552     } else {
 553         POSIX::close($csync[1]);
 554         kill(-9, $cpid); # make sure it gets killed
 555         die $err;
 556     }
 557
 558     $self->log_cluster_msg('info', $user, "starting task $upid");
 559
 560     my $tlist = $self->active_workers($upid, $sync);
 561     $self->broadcast_tasklist($tlist);
 562
 563     my $res = 0;
 564
 565     if ($sync) {
 566         my $count;
 567         my $outbuf = '';
 568         my $int_count = 0;
 569         eval {
 570             local $SIG{INT} = local $SIG{QUIT} = local $SIG{TERM} = sub {
 571                 # always send signal to all pgrp members
 572                 my $kpid = -$cpid;
 573                 if ($int_count < 3) {
 574                     kill(15, $kpid); # send TERM signal
 575                 } else {
 576                     kill(9, $kpid); # send KILL signal
 577                 }
 578                 $int_count++;
 579             };
 580             local $SIG{PIPE} = sub { die "broken pipe\n"; };
 581
 582             my $select = new IO::Select;
 583             my $fh = IO::Handle->new_from_fd($psync[0], 'r');
 584             $select->add($fh);
 585
 586             while ($select->count) {
 587                 my @handles = $select->can_read(1);
 588                 if (scalar(@handles)) {
 589                     my $count = sysread ($handles[0], $readbuf, 4096);
 590                     if (!defined ($count)) {
 591                         my $err = $!;
 592                         die "sync pipe read error: $err\n";
 593                     }
 594                     last if $count == 0; # eof
 595
 596                     $outbuf .= $readbuf;
 597                     while ($outbuf =~ s/^(([^\010\r\n]*)(\r|\n|(\010)+|\r\n))//s) {
 598                         my $line = $1;
 599                         my $data = $2;
 600                         if ($data =~ m/^TASK OK$/) {
 601                             # skip
 602                         } elsif ($data =~ m/^TASK ERROR: (.+)$/) {
 603                             print STDERR "$1\n";
 604                         } else {
 605                             print $line;
 606                         }
 607                         if ($outfh) {
 608                             print $outfh $line;
 609                             $outfh->flush();
 610                         }
 611                     }
 612                 } else {
 613                     # some commands daemonize without closing stdout
 614                     last if !PVE::ProcFSTools::check_process_running($cpid);
 615                 }
 616             }
 617         };
 618         my $err = $@;
 619
 620         POSIX::close($psync[0]);
 621
 622         if ($outbuf) { # just to be sure
 623             print $outbuf;
 624             if ($outfh) {
 625                 print $outfh $outbuf;
 626             }
 627         }
 628
 629         if ($err) {
 630             $err =~ s/\n/ /mg;
 631             print STDERR "$err\n";
 632             if ($outfh) {
 633                 print $outfh "TASK ERROR: $err\n";
 634             }
 635         }
 636
 637         &$kill_process_group($cpid, $pstart); # make sure it gets killed
 638
 639         close($outfh);
 640
 641         waitpid($cpid, 0);
 642         $res = $?;
 643         &$log_task_result($upid, $user, $res);
 644     }
 645
 646     return wantarray ? ($upid, $res) : $upid;
 647 }
 648
 649 # Abstract function
 650
 651 sub log_cluster_msg {
 652     my ($self, $pri, $user, $msg) = @_;
 653
 654     syslog($pri, "%s", $msg);
 655
 656     # PVE::Cluster::log_msg($pri, $user, $msg);
 657 }
 658
 659 sub broadcast_tasklist {
 660     my ($self, $tlist) = @_;
 661
 662     # PVE::Cluster::broadcast_tasklist($tlist);
 663 }
 664
 665 sub check_api2_permissions {
 666     my ($self, $perm, $username, $param) = @_;
 667
 668     return 1 if !$username && $perm->{user} eq 'world';
 669
 670     raise_perm_exc("user != null") if !$username;
 671
 672     return 1 if $username eq 'root@pam';
 673
 674     raise_perm_exc('user != root@pam') if !$perm;
 675
 676     return 1 if $perm->{user} && $perm->{user} eq 'all';
 677
 678     ##return $self->exec_api2_perm_check($perm->{check}, $username, $param)
 679     ##if $perm->{check};
 680
 681     raise_perm_exc();
 682 }
 683
 684 # init_request - should be called before each REST/CLI request
 685 sub init_request {
 686     my ($self, %params) = @_;
 687
 688     $self->{result_attributes} = {}
 689
 690     # if you nedd more, implement in subclass
 691 }
 692
 693 1;