[PATCH manager 11/19] vzdump: back up guests of a job in parallel

[Thomas Lamprecht <t.lamprecht@proxmox.com>]

Each node backed up the guests of a job strictly one after the other.
How long a job took thus depended on how its guests were spread over
the nodes: guests all on one node were backed up in series, even if
that node and the backup target could handle more.

Allow backing up several guests at the same time, up to the backup
concurrency of the node. It comes from the new node configuration
option or the datacenter-wide default, and stays at one guest unless
configured. A job can lower it with its 'concurrency' option.

Each guest backup then runs in a forked process. The job process
collects the results, prefixes the guest output in the task log with
the guest ID, keeps the warning count, and sends a single notification
as before. A guest backup process that ends without a result is
reported as failed.

Stopping the task signals all its processes, so the guest backups stop
on their own while the job process waits for them. It does not signal
them again, which could interrupt their cleanup. A stop request that
arrives while a guest backup process is forked does not reach that
process, so it only starts its backup once the job process confirmed
that there was none; otherwise the guest counts as not started. A
signal that only reached the job process keeps it from starting
further guests.

A guest listed more than once is backed up once, as its backups would
otherwise run at the same time.

Backups to external providers stay sequential, as providers are not
known to cope with backups from multiple processes. With a concurrency
of one, guests are backed up in the job process as before.

Signed-off-by: Thomas Lamprecht <t.lamprecht@proxmox.com>
---
 PVE/NodeConfig.pm               |   7 +
 PVE/VZDump.pm                   | 244 +++++++++++++++++++++++++++++++-
 test/Makefile                   |   6 +-
 test/vzdump_concurrency_test.pl |  89 ++++++++++++
 4 files changed, 340 insertions(+), 6 deletions(-)
 create mode 100755 test/vzdump_concurrency_test.pl

diff --git a/PVE/NodeConfig.pm b/PVE/NodeConfig.pm
index cc1a2a296..fbd7ecf87 100644
--- a/PVE/NodeConfig.pm
+++ b/PVE/NodeConfig.pm
@@ -108,6 +108,13 @@ my $confdesc = {
         default => 80,
         optional => 1,
     },
+    'backup-concurrency' => {
+        description => "Maximal number of guests backed up at the same time on this node."
+            . " Overrides the datacenter-wide default.",
+        type => 'integer',
+        minimum => 1,
+        optional => 1,
+    },
 };
 
 my $wakeonlan_desc = {
diff --git a/PVE/VZDump.pm b/PVE/VZDump.pm
index b12281790..33fe8793e 100644
--- a/PVE/VZDump.pm
+++ b/PVE/VZDump.pm
@@ -11,6 +11,7 @@ use IO::File;
 use IO::Select;
 use IPC::Open3;
 use POSIX qw(strftime);
+use Storable;
 use Time::Local;
 
 use PVE::Cluster qw(cfs_read_file);
@@ -19,6 +20,7 @@ use PVE::Exception qw(raise_param_exc);
 use PVE::HA::Config;
 use PVE::HA::Env::PVE2;
 use PVE::JSONSchema qw(get_standard_option);
+use PVE::NodeConfig;
 use PVE::Notify;
 use PVE::RPCEnvironment;
 use PVE::Storage;
@@ -288,10 +290,11 @@ sub check_vmids {
     my (@vmids) = @_;
 
     my $res = [];
+    my $seen = {};
     for my $vmid (sort { $a <=> $b } @vmids) {
         die "ERROR: strange VM ID '${vmid}'\n" if $vmid !~ m/^\d+$/;
         $vmid = int($vmid); # remove leading zeros
-        next if !$vmid;
+        next if !$vmid || $seen->{$vmid}++;
         push @$res, $vmid;
     }
 
@@ -1454,6 +1457,232 @@ sub skip_all_guests {
     return $tasklist;
 }
 
+my sub fail_task {
+    my ($task, $err) = @_;
+
+    $task->{state} = 'err';
+    $task->{msg} = $err;
+    debugmsg('err', "Backup of VM $task->{vmid} failed - $err");
+}
+
+sub node_backup_concurrency {
+    my $node_conf = PVE::NodeConfig::load_config(PVE::INotify::nodename());
+
+    return $node_conf->{'backup-concurrency'}
+        // cfs_read_file('datacenter.cfg')->{'backup-concurrency'}->{node} // 1;
+}
+
+sub guest_backup_concurrency {
+    my ($self) = @_;
+
+    # external providers are not known to cope with backups from multiple processes
+    return 1 if $self->{'backup-provider'};
+
+    my $node_limit = node_backup_concurrency();
+    my $limit = $self->{opts}->{concurrency} // $node_limit;
+
+    return $limit < $node_limit ? $limit : $node_limit;
+}
+
+my $guest_result_keys = [qw(backuptime hostname log msg size state target tmplog vmtype)];
+
+my sub fork_guest_backup {
+    my ($self, $task, $running, $interrupted) = @_;
+
+    pipe(my $output_reader, my $output_writer) or die "unable to create pipe - $!\n";
+    pipe(my $result_reader, my $result_writer) or die "unable to create pipe - $!\n";
+    pipe(my $start_reader, my $start_writer) or die "unable to create pipe - $!\n";
+
+    # Until the child set up its own handlers, a signal would only reach the inherited one of the
+    # job process and get lost.
+    my $signals = POSIX::SigSet->new(
+        POSIX::SIGINT(),
+        POSIX::SIGTERM(),
+        POSIX::SIGQUIT(),
+        POSIX::SIGHUP(),
+        POSIX::SIGPIPE(),
+    );
+    my $old_mask = POSIX::SigSet->new();
+    POSIX::sigprocmask(POSIX::SIG_BLOCK(), $signals, $old_mask)
+        or die "unable to block signals - $!\n";
+
+    my $pid = fork();
+    if (!defined($pid)) {
+        my $err = $!;
+        POSIX::sigprocmask(POSIX::SIG_SETMASK(), $old_mask);
+        die "unable to fork guest backup - $err\n";
+    }
+
+    if (!$pid) {
+        my $interrupted = 0;
+        $SIG{INT} = $SIG{TERM} = $SIG{QUIT} = $SIG{HUP} = $SIG{PIPE} = sub {
+            # a second signal must not abort the cleanup the first one started
+            die "interrupted by signal\n" if !$interrupted++;
+        };
+
+        my $exitcode = 1;
+        eval {
+            POSIX::sigprocmask(POSIX::SIG_SETMASK(), $old_mask);
+            close($output_reader);
+            close($result_reader);
+            close($start_writer);
+            # only the job process reads the output of the other guest backups
+            for my $sibling (values $running->%*) {
+                close($_) for grep { defined } $sibling->@{qw(output result)};
+            }
+            open(STDOUT, '>&', $output_writer) or die "unable to redirect STDOUT - $!\n";
+            open(STDERR, '>&', $output_writer) or die "unable to redirect STDERR - $!\n";
+            close($output_writer);
+
+            sysread($start_reader, my $start, 1) or die "interrupted by signal\n";
+            close($start_reader);
+
+            PVE::INotify::inotify_close();
+
+            eval { $self->exec_backup_task($task) };
+            my $err = $@;
+
+            my $result = { map { $_ => $task->{$_} } $guest_result_keys->@* };
+            print $result_writer Storable::nfreeze($result);
+            close($result_writer);
+
+            die $err if $err;
+            $exitcode = 0;
+        };
+        warn $@ if $@ && $@ !~ m/^interrupted by signal$/;
+        STDOUT->flush();
+        POSIX::_exit($exitcode);
+    }
+
+    POSIX::sigprocmask(POSIX::SIG_SETMASK(), $old_mask);
+
+    close($output_writer);
+    close($result_writer);
+    close($start_reader);
+
+    # A stop request that came in while forking reached only the job process, so the backup
+    # process must not start then.
+    my $cancelled = $$interrupted;
+    if (!$cancelled) {
+        local $SIG{PIPE} = 'IGNORE';
+        syswrite($start_writer, '1');
+    }
+    close($start_writer);
+
+    return {
+        pid => $pid,
+        cancelled => $cancelled,
+        task => $task,
+        output => $output_reader,
+        result => $result_reader,
+        'output-buffer' => '',
+        'result-data' => '',
+    };
+}
+
+my sub forward_guest_output {
+    my ($child, $data, $final) = @_;
+
+    my $vmid = $child->{task}->{vmid};
+
+    $child->{'output-buffer'} .= $data;
+    $child->{'output-buffer'} .= "\n" if $final && length($child->{'output-buffer'});
+
+    while ($child->{'output-buffer'} =~ s/^(.*)\n//) {
+        my $line = $1;
+        # warnings of a guest backup process count for the task, but its counter ends with it
+        PVE::RPCEnvironment::get()->{warning_count}++ if $line =~ m/^WARN: /;
+        print STDERR "$vmid: $line\n";
+    }
+}
+
+my sub read_guest_backup_pipe {
+    my ($child, $kind) = @_;
+
+    my $count = sysread($child->{$kind}, my $data, 64 * 1024);
+    return if !defined($count) && ($!{EINTR} || $!{EAGAIN});
+
+    if (!$count) {
+        close(delete $child->{$kind});
+        forward_guest_output($child, '', 1) if $kind eq 'output';
+    } elsif ($kind eq 'output') {
+        forward_guest_output($child, $data);
+    } else {
+        $child->{'result-data'} .= $data;
+    }
+}
+
+my sub finish_guest_backup {
+    my ($child, $status) = @_;
+
+    # the backup process is gone, but helpers it started might still hold the output pipe
+    for my $kind (qw(result output)) {
+        while ($child->{$kind} && IO::Select->new($child->{$kind})->can_read(0)) {
+            read_guest_backup_pipe($child, $kind);
+        }
+        close(delete $child->{$kind}) if $child->{$kind};
+    }
+    forward_guest_output($child, '', 1);
+    return if $child->{cancelled};
+
+    my $task = $child->{task};
+    my $result = eval { Storable::thaw($child->{'result-data'}) };
+    if (ref($result) eq 'HASH') {
+        $task->{$_} = $result->{$_} for keys $result->%*;
+    } else {
+        my $reason = $status & 127 ? "signal " . ($status & 127) : "exit code " . ($status >> 8);
+        fail_task($task, "backup process ended unexpectedly ($reason)");
+    }
+}
+
+my sub exec_parallel_backup_tasks {
+    my ($self, $tasklist, $concurrency) = @_;
+
+    my @queue = $tasklist->@*;
+    my $running = {};
+    my $interrupted = 0;
+    my $job_pid = $$;
+
+    # Stopping a task signals all its processes, so the guest backups stop on their own. Signalling
+    # them again could interrupt their cleanup.
+    local $SIG{INT} = local $SIG{TERM} = local $SIG{QUIT} = local $SIG{HUP} = local $SIG{PIPE} =
+        sub { $interrupted = 1; };
+
+    while (scalar(@queue) || scalar(keys $running->%*)) {
+        while (!$interrupted && scalar(@queue) && scalar(keys $running->%*) < $concurrency) {
+            my $task = shift @queue;
+            my $child = eval { fork_guest_backup($self, $task, $running, \$interrupted) };
+            # a guest backup process signalled while it exits dies into this eval
+            POSIX::_exit(1) if $$ != $job_pid;
+            if (my $err = $@) {
+                fail_task($task, $err);
+                next;
+            }
+            $running->{ $child->{pid} } = $child;
+            unshift @queue, $task if $child->{cancelled};
+        }
+
+        last if $interrupted && !scalar(keys $running->%*);
+
+        my $handles = {};
+        for my $child (values $running->%*) {
+            $handles->{ fileno($child->{$_}) } = [$child, $_]
+                for grep { $child->{$_} } qw(output result);
+        }
+        my $select = IO::Select->new(map { $_->[0]->{ $_->[1] } } values $handles->%*);
+        for my $fh ($select->can_read(1)) {
+            read_guest_backup_pipe($handles->{ fileno($fh) }->@*);
+        }
+
+        for my $pid (keys $running->%*) {
+            next if waitpid($pid, POSIX::WNOHANG()) != $pid;
+            finish_guest_backup(delete $running->{$pid}, $?);
+        }
+    }
+
+    die "interrupted by signal\n" if $interrupted;
+}
+
 sub exec_backup {
     my ($self, $rpcenv, $authuser) = @_;
 
@@ -1474,22 +1703,27 @@ sub exec_backup {
     open my $job_start_fd, '>', \$job_start_log;
     open my $job_end_fd, '>', \$job_end_log;
 
+    my $concurrency = $self->guest_backup_concurrency();
+    $concurrency = scalar($tasklist->@*) if scalar($tasklist->@*) < $concurrency;
+
     my $starttime = time();
-    my $errcount = 0;
     eval {
 
         $self->{'backup-provider'}->job_init($starttime) if $self->{'backup-provider'};
         $self->run_hook_script('job-start', undef, $job_start_fd);
 
-        foreach my $task (@$tasklist) {
-            $self->exec_backup_task($task);
-            $errcount += 1 if $task->{state} ne 'ok';
+        if ($concurrency > 1) {
+            debugmsg('info', "running up to $concurrency guest backups at the same time");
+            exec_parallel_backup_tasks($self, $tasklist, $concurrency);
+        } else {
+            $self->exec_backup_task($_) for $tasklist->@*;
         }
 
         $self->{'backup-provider'}->job_cleanup() if $self->{'backup-provider'};
         $self->run_hook_script('job-end', undef, $job_end_fd);
     };
     my $err = $@;
+    my $errcount = grep { $_->{state} ne 'ok' } $tasklist->@*;
 
     if ($err) {
         if ($self->{'backup-provider'}) {
diff --git a/test/Makefile b/test/Makefile
index 2dfd20a41..1eb94f4e0 100644
--- a/test/Makefile
+++ b/test/Makefile
@@ -18,7 +18,7 @@ replication%.t: replication_test%.pl
 	./$<
 
 .PHONY: test-vzdump
-test-vzdump: test-vzdump-guest-included test-vzdump-new
+test-vzdump: test-vzdump-guest-included test-vzdump-new test-vzdump-concurrency
 
 .PHONY: test-vzdump-guest-included
 test-vzdump-guest-included:
@@ -28,6 +28,10 @@ test-vzdump-guest-included:
 test-vzdump-new:
 	./vzdump_new_test.pl
 
+.PHONY: test-vzdump-concurrency
+test-vzdump-concurrency:
+	./vzdump_concurrency_test.pl
+
 .PHONY: test-osd
 test-osd:
 	./OSD_test.pl
diff --git a/test/vzdump_concurrency_test.pl b/test/vzdump_concurrency_test.pl
new file mode 100755
index 000000000..46d679490
--- /dev/null
+++ b/test/vzdump_concurrency_test.pl
@@ -0,0 +1,89 @@
+#!/usr/bin/perl
+
+use strict;
+use warnings;
+
+use lib '..';
+
+use Test::More;
+use Test::MockModule;
+
+use PVE::VZDump;
+
+my $node_config = {};
+my $datacenter_config = {};
+
+my $node_config_module = Test::MockModule->new('PVE::NodeConfig');
+$node_config_module->mock(load_config => sub { return $node_config; });
+
+my $vzdump_module = Test::MockModule->new('PVE::VZDump');
+$vzdump_module->mock(
+    cfs_read_file => sub {
+        my ($filename) = @_;
+        die "unexpected file '$filename'\n" if $filename ne 'datacenter.cfg';
+        return $datacenter_config;
+    },
+);
+
+my $tests = [
+    {
+        description => 'defaults keep backing up one guest at a time',
+        expected => 1,
+    },
+    {
+        description => 'datacenter default applies to the node',
+        datacenter => 4,
+        expected => 4,
+    },
+    {
+        description => 'node configuration overrides datacenter default',
+        datacenter => 4,
+        node => 2,
+        expected => 2,
+    },
+    {
+        description => 'node configuration applies without datacenter default',
+        node => 3,
+        expected => 3,
+    },
+    {
+        description => 'job limit lowers node limit',
+        node => 4,
+        job => 2,
+        expected => 2,
+    },
+    {
+        description => 'job limit cannot raise node limit',
+        node => 2,
+        job => 8,
+        expected => 2,
+    },
+    {
+        description => 'job limit does not raise the default',
+        job => 8,
+        expected => 1,
+    },
+    {
+        description => 'external backup providers back up one guest at a time',
+        node => 4,
+        job => 4,
+        provider => 1,
+        expected => 1,
+    },
+];
+
+for my $test ($tests->@*) {
+    $datacenter_config =
+        defined($test->{datacenter})
+        ? { 'backup-concurrency' => { node => $test->{datacenter} } }
+        : {};
+    $node_config = defined($test->{node}) ? { 'backup-concurrency' => $test->{node} } : {};
+
+    my $self = bless { opts => {} }, 'PVE::VZDump';
+    $self->{opts}->{concurrency} = $test->{job} if defined($test->{job});
+    $self->{'backup-provider'} = {} if $test->{provider};
+
+    is($self->guest_backup_concurrency(), $test->{expected}, $test->{description});
+}
+
+done_testing();
-- 
2.47.3