[PATCH manager 16/19] vzdump: honor backup concurrency limits of storages

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

Guest backups now also count against the backup concurrency limit of
the storages they use, both the backup target and the storages of the
included guest disks. This protects a backup server or a Ceph cluster
from more parallel backups than it can handle, which a limit for the
whole cluster cannot express once several such systems are in use.

Limits of shared storages count for the whole cluster and use the
cluster-wide slots. Limits of local storages count on each node and
use slot locks like the node limit. A guest whose local storage has no
free slot lets other guests of the job go first.

With several limits, a waiting guest backup can be blocked by a full
storage while younger ones with room in their limits get slots. This
may happen a few times, then a slot of each limit it needs is kept
free for it, so it cannot starve.

The storages are taken from the guest configuration before the guest
gets locked for backup, so waiting for slots does not block changes to
the guest. If its disks moved to another storage with a limit by the
time it got locked, the backup fails instead of running without a slot
for that storage. If the storages cannot be determined, the guest is
skipped for the same reason. Nothing of this happens as long as no
storage has a limit. The task log tells which limit a guest waits for.

Signed-off-by: Thomas Lamprecht <t.lamprecht@proxmox.com>
---
 PVE/VZDump.pm                     | 95 +++++++++++++++++++++++++++++--
 PVE/VZDump/ClusterSlots.pm        | 44 +++++++++++++-
 test/vzdump_cluster_slots_test.pl | 79 ++++++++++++++++++++++++-
 3 files changed, 209 insertions(+), 9 deletions(-)

diff --git a/PVE/VZDump.pm b/PVE/VZDump.pm
index 525a593c6..59809d85c 100644
--- a/PVE/VZDump.pm
+++ b/PVE/VZDump.pm
@@ -933,6 +933,34 @@ sub get_backup_file_list {
     return $bklist;
 }
 
+my sub limited_backup_storages {
+    my ($self, $task) = @_;
+
+    my $storage_cfg = PVE::Storage::config();
+    my $limited = {
+        map { $_ => 1 }
+        grep { defined($storage_cfg->{ids}->{$_}->{'backup-concurrency'}) }
+            keys $storage_cfg->{ids}->%*
+    };
+    return [] if !scalar(keys $limited->%*);
+
+    my $storeids = {};
+    $storeids->{ $self->{opts}->{storage} } = 1 if $self->{opts}->{storage};
+
+    if (my $plugin = $task->{plugin}) {
+        my $config_class = $plugin->type() eq 'qemu' ? 'PVE::QemuConfig' : 'PVE::LXC::Config';
+        my $conf = $config_class->load_config($task->{vmid});
+        for my $volume ($config_class->get_backup_volumes($conf)->@*) {
+            next if !$volume->{included};
+            my $volid = $volume->{volume_config}->{file} // $volume->{volume_config}->{volume};
+            my ($storeid) = PVE::Storage::parse_volume_id($volid, 1);
+            $storeids->{$storeid} = 1 if defined($storeid);
+        }
+    }
+
+    return [sort grep { $limited->{$_} } keys $storeids->%*];
+}
+
 sub exec_backup_task {
     my ($self, $task) = @_;
 
@@ -1116,6 +1144,13 @@ sub exec_backup_task {
 
         $cleanup->{unlock} = 1;
 
+        my $taken = { map { $_ => 1 } $task->{storages}->@* };
+        if (my @missing = grep { !$taken->{$_} } limited_backup_storages($self, $task)->@*) {
+            my $list = join(', ', map { "'$_'" } @missing);
+            die "no backup slot for storage $list - guest disks or storage limits changed while"
+                . " waiting for backup slots\n";
+        }
+
         # prepare
 
         my $mode = $running ? $task->{mode} : 'stop';
@@ -1717,6 +1752,25 @@ my sub try_node_slot {
     return try_slot("$lockdir/slot", node_backup_concurrency());
 }
 
+# Limits of local storages count on each node, shared ones are handled cluster-wide.
+my sub try_local_storage_slots {
+    my ($storeids) = @_;
+
+    my $storage_cfg = PVE::Storage::config();
+
+    my $locks = [];
+    for my $storeid ($storeids->@*) {
+        my $scfg = $storage_cfg->{ids}->{$storeid} or next;
+        my $limit = $scfg->{'backup-concurrency'};
+        next if $scfg->{shared} || !defined($limit);
+
+        my $lock = try_slot("$lockdir/storage-$storeid", $limit) or return (undef, $storeid);
+        push $locks->@*, $lock;
+    }
+
+    return ($locks);
+}
+
 my sub log_wait {
     my ($self, $reason) = @_;
 
@@ -1766,10 +1820,25 @@ my sub claim_next_guest {
             skip_task($task, "guest moved to node '$node'");
             next;
         }
+
+        $task->{storages} = eval { limited_backup_storages($self, $task) };
+        if (my $err = $@) {
+            die $err if $err =~ m/interrupted by signal$/;
+            chomp($err);
+            splice($queue->@*, $i--, 1);
+            skip_task($task, "unable to check storage backup limits - $err");
+            next;
+        }
+
+        my ($storage_locks, $busy_storage) = try_local_storage_slots($task->{storages});
+        if (!$storage_locks) {
+            log_wait($self, "a free backup slot of storage '$busy_storage'");
+            next;
+        }
         delete $self->{'waiting-for'};
 
         splice($queue->@*, $i--, 1);
-        return ($task, [$guest_lock, $slot]);
+        return ($task, [$guest_lock, $slot, $storage_locks->@*]);
     }
 
     return;
@@ -1785,7 +1854,7 @@ my sub skip_after_error {
 }
 
 my sub cluster_slot_resources {
-    my ($self) = @_;
+    my ($self, $task) = @_;
 
     my $limits = PVE::VZDump::ClusterSlots::limits();
     return undef if !scalar(keys $limits->%*);
@@ -1798,7 +1867,15 @@ my sub cluster_slot_resources {
         return undef;
     }
 
-    return ['cluster'];
+    my $storage_cfg = PVE::Storage::config();
+    my $shared = [grep { $storage_cfg->{ids}->{$_}->{shared} } $task->{storages}->@*];
+
+    return ['cluster', map { "storage/$_" } $shared->@*];
+}
+
+my sub render_cluster_limits {
+    my ($resources) = @_;
+    return join(', ', map { m!^storage/(.*)$! ? "storage '$1'" : "the $_" } $resources->@*);
 }
 
 my sub release_cluster_slots {
@@ -1824,7 +1901,7 @@ my sub next_guest_to_start {
     my $pending = $schedule->{pending};
     if (!$pending) {
         my ($task, $locks) = claim_next_guest($self, $queue, $schedule->{deferred}) or return;
-        my $resources = eval { cluster_slot_resources($self) };
+        my $resources = eval { cluster_slot_resources($self, $task) };
         if (my $err = $@) {
             unshift $queue->@*, $task;
             die $err;
@@ -1880,8 +1957,14 @@ my sub next_guest_to_start {
         my $reason = 'timeout waiting for cluster-wide backup slot';
         $reason .= " - $pending->{'request-error'}" if defined($pending->{'request-error'});
         skip_task($task, $reason);
-    } elsif (!$pending->{logged}++) {
-        debugmsg('info', "VM $task->{vmid} waits for a cluster-wide backup slot");
+    } elsif (defined($pending->{key}) && !$pending->{logged}++) {
+        my $reached = eval { PVE::VZDump::ClusterSlots::reached_limits($pending->{key}) };
+        die $@ if $@ =~ m/interrupted by signal$/;
+        my $reason =
+            scalar(($reached // [])->@*)
+            ? "limit reached for " . render_cluster_limits($reached)
+            : "queued behind other guest backups";
+        debugmsg('info', "VM $task->{vmid} waits for a cluster-wide backup slot, $reason");
     }
 
     return;
diff --git a/PVE/VZDump/ClusterSlots.pm b/PVE/VZDump/ClusterSlots.pm
index 6537bdce5..f4cac71c7 100644
--- a/PVE/VZDump/ClusterSlots.pm
+++ b/PVE/VZDump/ClusterSlots.pm
@@ -14,6 +14,7 @@ use JSON;
 use PVE::Cluster qw(cfs_read_file);
 use PVE::INotify;
 use PVE::QemuServer::Helpers;
+use PVE::Storage;
 use PVE::Tools;
 
 my $state_file = '/etc/pve/priv/backup-slots.json';
@@ -21,6 +22,9 @@ my $state_file = '/etc/pve/priv/backup-slots.json';
 # first release of pve-manager that takes part in cluster-wide limits
 my $min_version = [9, 2, 26];
 
+# How often younger requests may get ahead of a waiting one before its limits are reserved for it.
+use constant MAX_BYPASS => 3;
+
 sub limits {
     my $limits = {};
 
@@ -29,6 +33,13 @@ sub limits {
         $limits->{cluster} = $cluster;
     }
 
+    my $storage_cfg = PVE::Storage::config();
+    for my $storeid (keys $storage_cfg->{ids}->%*) {
+        my $scfg = $storage_cfg->{ids}->{$storeid};
+        next if !$scfg->{shared} || !defined($scfg->{'backup-concurrency'});
+        $limits->{"storage/$storeid"} = $scfg->{'backup-concurrency'};
+    }
+
     return $limits;
 }
 
@@ -127,7 +138,8 @@ my sub remove_entries {
 
 # Moves waiters that fit into all limits they count against to the holders, oldest first. Only
 # resources with a limit matter, the others are tracked for information. Waiters of offline nodes
-# neither get granted nor hold up others.
+# neither get granted nor hold up others. A waiter that younger ones got ahead of too often
+# reserves a slot of each limit it counts against, so it cannot starve.
 sub grant_waiters {
     my ($state, $limits, $members) = @_;
 
@@ -135,13 +147,29 @@ sub grant_waiters {
 
     my $used = used_slots($state);
 
+    my $passed_over = [];
+
     my $waiters = $state->{waiters};
     for my $key (sort { $waiters->{$a}->{seq} <=> $waiters->{$b}->{seq} } keys $waiters->%*) {
         my $waiter = $waiters->{$key};
         next if !$members->{ $waiter->{node} }->{online};
 
         my $limited = [grep { defined($limits->{$_}) } $waiter->{resources}->@*];
-        next if grep { ($used->{$_} // 0) >= $limits->{$_} } $limited->@*;
+        my $fits = !grep { ($used->{$_} // 0) >= $limits->{$_} } $limited->@*;
+
+        if (!$fits) {
+            if (($waiter->{bypassed} // 0) >= MAX_BYPASS) {
+                $used->{$_}++ for $limited->@*;
+            }
+            push $passed_over->@*, [$waiter, $limited];
+            next;
+        }
+
+        my %mine = map { $_ => 1 } $limited->@*;
+        for my $older ($passed_over->@*) {
+            my ($older_waiter, $older_limited) = $older->@*;
+            $older_waiter->{bypassed}++ if grep { $mine{$_} } $older_limited->@*;
+        }
 
         $used->{$_}++ for $waiter->{resources}->@*;
         $state->{holders}->{$key} = delete $waiters->{$key};
@@ -208,6 +236,18 @@ sub granted {
     return undef;
 }
 
+sub reached_limits {
+    my ($key) = @_;
+
+    my $state = read_state();
+    my $waiter = $state->{waiters}->{$key} or return [];
+
+    my $used = used_slots($state);
+    my $limits = limits();
+    return [grep { defined($limits->{$_}) && ($used->{$_} // 0) >= $limits->{$_} }
+        $waiter->{resources}->@*];
+}
+
 # Grants waiters that fit by now, for example after limits got raised.
 sub refresh {
     update_state(sub { });
diff --git a/test/vzdump_cluster_slots_test.pl b/test/vzdump_cluster_slots_test.pl
index 85608b5be..5e13512b2 100755
--- a/test/vzdump_cluster_slots_test.pl
+++ b/test/vzdump_cluster_slots_test.pl
@@ -27,11 +27,12 @@ my sub make_state {
         $state->{holders}->{$key} = { node => $node, resources => $resources };
     }
     for my $waiter ($waiters->@*) {
-        my ($key, $node, $resources) = $waiter->@*;
+        my ($key, $node, $resources, $bypassed) = $waiter->@*;
         $state->{waiters}->{$key} = {
             node => $node,
             resources => $resources,
             seq => ++$state->{seq},
+            bypassed => $bypassed // 0,
         };
     }
     return $state;
@@ -85,6 +86,54 @@ my $tests = [
         expected => ['b'],
         'expected-waiters' => [],
     },
+    {
+        description => 'younger waiter with free resources gets ahead of a blocked one',
+        limits => { x => 1, y => 1 },
+        holders => [['h', 'node1', ['x']]],
+        waiters => [['a', 'node1', ['x', 'y']], ['b', 'node2', ['y']]],
+        expected => ['b', 'h'],
+        bypassed => { a => 1 },
+    },
+    {
+        description => 'waiter passed over too often reserves its resources',
+        limits => { x => 1, y => 1 },
+        holders => [['h', 'node1', ['x']]],
+        waiters => [['a', 'node1', ['x', 'y'], 3], ['b', 'node2', ['y']]],
+        expected => ['h'],
+    },
+    {
+        description => 'waiter passed over too often reserves only one slot of each resource',
+        limits => { cluster => 3, x => 1 },
+        holders => [['h', 'node1', ['cluster', 'x']]],
+        waiters => [
+            ['a', 'node1', ['cluster', 'x'], 3],
+            ['b', 'node2', ['cluster']],
+            ['c', 'node2', ['cluster']],
+        ],
+        expected => ['b', 'h'],
+    },
+    {
+        description => 'reserved resources do not hold up unrelated waiters',
+        limits => { x => 1, y => 1, z => 1 },
+        holders => [['h', 'node1', ['x']]],
+        waiters => [['a', 'node1', ['x', 'y'], 3], ['b', 'node2', ['z']]],
+        expected => ['b', 'h'],
+    },
+    {
+        description => 'resources without limit do not count as passed over',
+        limits => { x => 1 },
+        holders => [['h', 'node1', ['cluster', 'x']]],
+        waiters => [['a', 'node1', ['cluster', 'x']], ['b', 'node2', ['cluster']]],
+        expected => ['b', 'h'],
+        bypassed => { a => 0 },
+    },
+    {
+        description => 'resources without limit do not get reserved',
+        limits => { x => 1 },
+        holders => [['h', 'node1', ['cluster', 'x']]],
+        waiters => [['a', 'node1', ['cluster', 'x'], 3], ['b', 'node2', ['cluster', 'y']]],
+        expected => ['b', 'h'],
+    },
 ];
 
 for my $test ($tests->@*) {
@@ -101,6 +150,34 @@ for my $test ($tests->@*) {
             "$test->{description} (waiters)",
         );
     }
+    for my $key (sort keys(($test->{bypassed} // {})->%*)) {
+        is(
+            $state->{waiters}->{$key}->{bypassed},
+            $test->{bypassed}->{$key},
+            "$test->{description} ($key bypassed)",
+        );
+    }
+}
+
+{
+    my $limits = { cluster => 2, x => 1 };
+    my $state = make_state([['h', 'node1', ['cluster', 'x']]], [['a', 'node1', ['cluster', 'x']]]);
+    my $add_younger = sub {
+        my ($key) = @_;
+        $state->{waiters}->{$key} =
+            { node => 'node2', resources => ['cluster'], seq => ++$state->{seq} };
+        PVE::VZDump::ClusterSlots::grant_waiters($state, $limits, $members);
+    };
+    for my $younger (qw(b c d)) {
+        $add_younger->($younger);
+        delete $state->{holders}->{$younger};
+    }
+    $add_younger->('e');
+    is_deeply(granted($state), ['h'], 'younger waiters stop getting ahead after a few times');
+
+    delete $state->{holders}->{h};
+    PVE::VZDump::ClusterSlots::grant_waiters($state, $limits, $members);
+    is_deeply(granted($state), ['a', 'e'], 'passed over waiter gets its slots once they are free');
 }
 
 # Request, grant, release and cleanup against an in-memory table.
-- 
2.47.3