use strict;
use warnings;
-use POSIX qw(:sys_wait_h EINTR);
-use IO::Handle;
+
+use Exporter qw(import);
+use Fcntl qw(:flock);
use IO::File;
+use IO::Handle;
use IO::Select;
-use Fcntl qw(:flock);
+use POSIX qw(:sys_wait_h EINTR);
+use AnyEvent;
+
use PVE::Exception qw(raise raise_perm_exc);
-use PVE::SafeSyslog;
-use PVE::Tools;
use PVE::INotify;
use PVE::ProcFSTools;
+use PVE::SafeSyslog;
+use PVE::Tools;
+our @EXPORT_OK = qw(log_warn);
my $rest_env;
# and register forked processes with &$register_worker(pid)
# Note: using $SIG{CHLD} = 'IGNORE' or $SIG{CHLD} = sub { wait (); } or ...
# has serious side effects, because perls built in system() and open()
-# functions can't get the correct exit status of a child. So we cant use
+# functions can't get the correct exit status of a child. So we can't use
# that (also see perlipc)
my $WORKER_PIDS;
die "unknown environment type"
if !$type || $type !~ m/^(cli|pub|priv|ha)$/;
- $SIG{CHLD} = $worker_reaper;
+ $SIG{CHLD} = sub {
+ # when we're using AnyEvent, we have to postpone the call to worker_reaper, otherwise it
+ # might interfere with running api calls
+ if (defined($AnyEvent::MODEL)) {
+ AnyEvent::postpone { $worker_reaper->() };
+ } else {
+ $worker_reaper->();
+ }
+ };
# environment types
# cli ... command started fron command line
- # pub ... access from public server (apache)
+ # pub ... access from public server (pveproxy)
# priv ... access from private server (pvedaemon)
- # ha ... access from HA resource manager agent (rgmanager)
+ # ha ... access from HA resource manager agent (pve-ha-manager)
- my $self = { type => $type };
+ my $self = {
+ type => $type,
+ warning_count => 0,
+ };
bless $self, $class;
die "user name not set\n";
}
+sub set_u2f_challenge {
+ my ($self, $challenge) = @_;
+
+ $self->{u2f_challenge} = $challenge;
+}
+
+sub get_u2f_challenge {
+ my ($self, $noerr) = @_;
+
+ return $self->{u2f_challenge} if defined($self->{u2f_challenge}) || $noerr;
+
+ die "no active u2f challenge\n";
+}
+
+sub set_request_host {
+ my ($self, $host) = @_;
+
+ $self->{request_host} = $host;
+}
+
+sub get_request_host {
+ my ($self, $noerr) = @_;
+
+ return $self->{request_host} if defined($self->{request_host}) || $noerr;
+
+ die "no hostname available in current environment\n";
+}
+
sub is_worker {
my ($class) = @_;
return $WORKER_FLAG;
}
-# read/update list of active workers
-# we move all finished tasks to the archive index,
-# but keep aktive and most recent task in the active file.
-# $nocheck ... consider $new_upid still running (avoid that
-# we try to read the reult to early.
-sub active_workers {
+# read/update list of active workers.
+#
+# we move all finished tasks to the archive index, but keep active, and most recent tasks in the
+# active file.
+# $nocheck ... consider $new_upid still running (avoid that we try to read the result to early).
+sub active_workers {
my ($self, $new_upid, $nocheck) = @_;
- my $lkfn = "/var/log/pve/tasks/.active.lock";
-
my $timeout = 10;
- my $code = sub {
-
+ my $res = PVE::Tools::lock_file("/var/log/pve/tasks/.active.lock", $timeout, sub {
my $tasklist = PVE::INotify::read_file('active');
my @ta;
&$check_task($task);
}
- if ($new_upid && !(my $task = $thash->{$new_upid})) {
- $task = PVE::Tools::upid_decode($new_upid);
+ if ($new_upid && !$thash->{$new_upid}) {
+ my $task = PVE::Tools::upid_decode($new_upid);
$task->{upid} = $new_upid;
$thash->{$new_upid} = $task;
&$check_task($task, $nocheck);
}
}
- # we try to reduce the amount of data
- # list all running tasks and task and a few others
- # try to limit to 25 tasks
- my $max = 25 - scalar(@$tlist);
+ # we try to reduce the amount of data list all running tasks and task and a few others
+ my $MAX_FINISHED = 25;
+ my $max = $MAX_FINISHED - scalar(@$tlist);
foreach my $task (@ta) {
last if $max <= 0;
push @$tlist, $task;
PVE::INotify::write_file('active', $tlist) if $save;
return $tlist;
- };
-
- my $res = PVE::Tools::lock_file($lkfn, $timeout, $code);
+ });
die $@ if $@;
return $res;
};
local $SIG{PIPE} = sub { die "broken pipe\n"; };
- my $select = new IO::Select;
+ my $select = IO::Select->new();
my $fh = IO::Handle->new_from_fd($childfd, 'r');
$select->add($fh);
}
}
- # get status (error or OK)
POSIX::read($ctrlfd, $readbuf, 4096);
if ($readbuf =~ m/^TASK OK\n?$/) {
# skip printing to stdout
} elsif ($readbuf =~ m/^TASK ERROR: (.*)\n?$/) {
print STDERR "$1\n";
print $taskfh "\n$readbuf"; # ensure start on new line for webUI
+ } elsif ($readbuf =~ m/^TASK WARNINGS: (\d+)\n?$/) {
+ print STDERR "Task finished with $1 warning(s)!\n";
+ print $taskfh "\n$readbuf"; # ensure start on new line for webUI
} else {
die "got unexpected control message: $readbuf\n";
}
$dtype = 'unknown' if !defined ($dtype);
$id = '' if !defined ($id);
- $user = 'root@pve' if !defined ($user);
+ # note: below is only used for the task log entry
+ $user = $self->get_user(1) // 'root@pam' if !defined($user);
my $sync = ($self->{type} eq 'cli' && !$background) ? 1 : 0;
my @psync = POSIX::pipe();
my @csync = POSIX::pipe();
- my @ctrlfd = POSIX::pipe() if $sync;
+ my @ctrlfd = $sync ? POSIX::pipe() : ();
my $node = $self->{nodename};
$SIG{CHLD} = $SIG{PIPE} = 'DEFAULT';
$SIG{TTOU} = 'IGNORE';
+ my $ppgid;
# set session/process group allows to kill the process group
if ($sync && -t STDIN) {
# some sync'ed workers operate on the tty but setsid sessions lose
# the tty, so just create a new pgroup and give it the tty
- POSIX::setpgid(0, 0) or die "failed to setpgid: $!\n";;
+ $ppgid = POSIX::getpgrp() or die "failed to get old pgid: $!\n";
+ POSIX::setpgid(0, 0) or die "failed to setpgid: $!\n";
POSIX::tcsetpgrp(fileno(STDIN), $$) or die "failed to tcsetpgrp: $!\n";
} else {
POSIX::setsid();
&$atfork();
}
- # same algorythm as used inside SA
+ # same algorithm as used inside SA
# STDIN = /dev/null
my $fd = fileno (STDIN);
close STDIN;
POSIX::close(0) if $fd != 0;
- die "unable to redirect STDIN - $!"
- if !open(STDIN, "</dev/null");
+ open(STDIN, '<', '/dev/null') or die "unable to redirect STDIN - $!";
$outfh = PVE::Tools::upid_open($upid);
$resfh = fileno($outfh);
close STDOUT;
POSIX::close (1) if $fd != 1;
- die "unable to redirect STDOUT - $!"
- if !open(STDOUT, ">&", $outfh);
+ open(STDOUT, ">&", $outfh) or die "unable to redirect STDOUT - $!";
STDOUT->autoflush (1);
close STDERR;
POSIX::close(2) if $fd != 2;
- die "unable to redirect STDERR - $!"
- if !open(STDERR, ">&1");
+ open(STDERR, '>&', '1') or die "unable to redirect STDERR - $!";
STDERR->autoflush(1);
};
}
&$function($upid);
};
+ my ($msg, $exitcode);
my $err = $@;
if ($err) {
chomp $err;
$err =~ s/\n/ /mg;
syslog('err', $err);
- my $msg = "TASK ERROR: $err\n";
- POSIX::write($resfh, $msg, length($msg));
- POSIX::close($resfh) if $sync;
- POSIX::_exit(-1);
+ $msg = "TASK ERROR: $err\n";
+ $exitcode = -1;
+ } elsif (my $warnings = $self->{warning_count}) {
+ $msg = "TASK WARNINGS: $warnings\n";
+ $exitcode = 0;
} else {
- my $msg = "TASK OK\n";
- POSIX::write($resfh, $msg, length($msg));
- POSIX::close($resfh) if $sync;
- POSIX::_exit(0);
+ $msg = "TASK OK\n";
+ $exitcode = 0;
}
- kill(-9, $$);
+ POSIX::write($resfh, $msg, length($msg));
+
+ if ($sync) {
+ POSIX::close($resfh);
+ if ( -t STDIN) {
+ POSIX::tcsetpgrp(fileno(STDIN), $ppgid) or
+ die "failed to tcsetpgrp to parent: $!\n";
+ }
+ }
+ POSIX::_exit($exitcode);
+ kill(-9, $$); # not really needed, just to be sure
}
# parent
return wantarray ? ($upid, $res) : $upid;
}
+sub log_warn {
+ my ($message) = @_;
+
+ if ($rest_env) {
+ $rest_env->warn($message);
+ } else {
+ chomp($message);
+ print STDERR "WARN: $message\n";
+ }
+}
+
+sub warn {
+ my ($self, $message) = @_;
+
+ chomp($message);
+
+ print STDERR "WARN: $message\n";
+
+ $self->{warning_count}++;
+}
+
# Abstract function
sub log_cluster_msg {
projects / pve-common.git / blobdiff