From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from gate001.proxmox.com (gate001.proxmox.com [45.144.208.40]) by lore.proxmox.com (Postfix) with ESMTPS id AD9191FF0C2 for ; Thu, 22 Oct 2026 16:23:33 +0200 (CEST) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id E5B182172C; Thu, 22 Oct 2026 16:23:01 +0200 (CEST) From: Hannes Laimer To: pve-devel@lists.proxmox.com Subject: [PATCH pve-cluster 12/12] cfs: add runner executing cluster change hooks in children Date: Thu, 22 Oct 2026 16:22:05 +0200 Message-ID: <20261022142205.930536-13-h.laimer@proxmox.com> X-Mailer: git-send-email 2.47.3 In-Reply-To: <20261022142205.930536-1-h.laimer@proxmox.com> References: <20261022142205.930536-1-h.laimer@proxmox.com> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit X-Bm-Milter-Handled: 55990f41-d878-4baa-be0a-ee34c49e34d2 X-Bm-Transport-Timestamp: 1792678940709 X-SPAM-LEVEL: Spam detection results: 0 AWL -0.812 Adjusted score from AWL reputation of From: address DMARC_MISSING 0.1 Missing DMARC policy KAM_DMARC_STATUS 0.01 Test Rule for DKIM or SPF Failure with Strict Alignment (newer systems) KAM_MAILER 2 Automated Mailer Tag Left in Email POISEN_SPAM_PILL 0.1 Meta: its spam POISEN_SPAM_PILL_1 0.1 random spam to be learned in bayes POISEN_SPAM_PILL_3 0.1 random spam to be learned in bayes RCVD_IN_DNSWL_MED -2.3 Sender listed at https://www.dnswl.org/, medium trust SPF_HELO_NONE 0.001 SPF: HELO does not publish an SPF Record SPF_PASS -0.001 SPF: sender matches SPF record Message-ID-Hash: QUS6S4K7EPRVGDVZM5G74HRNQD4CWQQ6 X-Message-ID-Hash: QUS6S4K7EPRVGDVZM5G74HRNQD4CWQQ6 X-MailFrom: h.laimer@proxmox.com X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; loop; banned-address; emergency; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header X-Mailman-Version: 3.3.10 Precedence: list List-Id: Proxmox VE development discussion List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: The runner subscribes to the hook registry's path templates over one notification socket connection and executes the runs the registry derives from each event. Every run gets a forked child, so a hook that dies or hangs cannot take down the runner or stall its connection. The registry keys a run by its hook and parameters, and one run per key is in flight at a time. Events for that key that arrive meanwhile collapse into one rerun after the run ends. Runs of different keys go in parallel up to a worker limit. A run that runs too long stops counting against that limit, so the other runs keep going. The next event for its key kills it and starts a fresh run. A cap on all children keeps such runs from piling up. A stop terminates its children and kills what has not exited after a grace period. A child that survives even the kill is logged and left behind, since waiting on it any longer has no point. Signed-off-by: Hannes Laimer --- debian/pve-cluster.install | 1 + src/PVE/Cluster/HookRunner.pm | 268 ++++++++++++++++++++++++++++++ src/PVE/Cluster/Makefile | 2 +- src/test/Makefile | 6 +- src/test/hook_runner_test.pl | 302 ++++++++++++++++++++++++++++++++++ 5 files changed, 577 insertions(+), 2 deletions(-) create mode 100644 src/PVE/Cluster/HookRunner.pm create mode 100644 src/test/hook_runner_test.pl diff --git a/debian/pve-cluster.install b/debian/pve-cluster.install index b8ebff7..8321f70 100644 --- a/debian/pve-cluster.install +++ b/debian/pve-cluster.install @@ -4,6 +4,7 @@ usr/bin/pmxcfs usr/lib/ usr/share/man/man8/pmxcfs.8 usr/share/perl5/PVE/Cluster.pm +usr/share/perl5/PVE/Cluster/HookRunner.pm usr/share/perl5/PVE/Cluster/Hooks.pm usr/share/perl5/PVE/Cluster/IPCConst.pm usr/share/perl5/PVE/Cluster/Watch.pm diff --git a/src/PVE/Cluster/HookRunner.pm b/src/PVE/Cluster/HookRunner.pm new file mode 100644 index 0000000..f6fe34a --- /dev/null +++ b/src/PVE/Cluster/HookRunner.pm @@ -0,0 +1,268 @@ +package PVE::Cluster::HookRunner; + +use strict; +use warnings; + +use POSIX qw(WNOHANG); +use Time::HiRes qw(sleep time); + +use PVE::Cluster; +use PVE::Cluster::Hooks; +use PVE::Cluster::Watch; +use PVE::SafeSyslog; + +# Executes the runs the hook registry derives from cluster changes, one run +# per key at a time in a forked child. + +my $stop_grace = 5; + +sub new { + my ($class, %param) = @_; + + my $self = bless { + watch => $param{watch} // PVE::Cluster::Watch->new(), + max_workers => $param{max_workers} // 4, + stall_age => $param{stall_age} // 120, + running => {}, + by_key => {}, + pending => {}, + queue => [], + queued => {}, + terminate => 0, + fork_failed => 0, + }, $class; + + return $self; +} + +sub run { + my ($self) = @_; + + local $SIG{TERM} = sub { $self->{terminate} = 1 }; + local $SIG{INT} = $SIG{TERM}; + # a child's exit has to interrupt the wait, under the default + # disposition the wait runs on to its timeout + local $SIG{CHLD} = sub { }; + + # a warning on stderr is lost in a daemon, so the watch's warnings about + # its connection and those of hooks in the children go to syslog + local $SIG{__WARN__} = sub { + my ($msg) = @_; + chomp $msg; + syslog('warning', $msg); + }; + + $self->{watch}->subscribe(PVE::Cluster::Hooks->subscription()->%*); + + # the wait dies on a refused handshake. The children are stopped before + # the error propagates, so none of them outlives the runner + eval { + # with nothing running or queued only an event or a signal needs + # handling, and both end the wait early. Runs in flight need the short + # wait so a stall is noticed + while (!$self->{terminate}) { + $self->run_once($self->idle() ? 60 : 1); + } + }; + my $err = $@; + $self->stop_children(); + die $err if $err; + + return; +} + +sub run_once { + my ($self, $timeout) = @_; + + for my $event ($self->{watch}->wait($timeout)) { + $self->schedule($_) for PVE::Cluster::Hooks->dispatch($event)->@*; + } + + $self->reap(); + $self->check_stalls(); + $self->start_queued() if !$self->{terminate}; + + return; +} + +sub stalled { + my ($self, $run) = @_; + + return time() - $run->{started} >= $self->{stall_age}; +} + +sub check_stalls { + my ($self) = @_; + + for my $pid (keys $self->{running}->%*) { + my $run = $self->{running}->{$pid}; + if ($run->{killed_at} && time() - $run->{killed_at} > $stop_grace && !$run->{forced}) { + syslog('warning', "killing unresponsive hook run '$run->{key}' as $pid"); + kill 'KILL', $pid; + $run->{forced} = 1; + } elsif (!$run->{stalled} && $self->stalled($run)) { + $run->{stalled} = 1; + my $age = int(time() - $run->{started}); + syslog('warning', "hook run '$run->{key}' as $pid still running after ${age}s"); + } + } + + return; +} + +sub idle { + my ($self) = @_; + + return !scalar(keys $self->{running}->%*) && !scalar($self->{queue}->@*); +} + +sub schedule { + my ($self, $run) = @_; + + my $key = $run->{key}; + + if (my $pid = $self->{by_key}->{$key}) { + $self->{pending}->{$key} = $run; + my $current = $self->{running}->{$pid}; + if ($self->stalled($current) && !$current->{killed_at}) { + syslog('warning', "replacing stalled hook run '$key' as $pid"); + kill 'TERM', $pid; + $current->{killed_at} = time(); + } + } elsif ($self->{queued}->{$key}) { + $self->{queued}->{$key} = $run; + } else { + push $self->{queue}->@*, $key; + $self->{queued}->{$key} = $run; + } + + return; +} + +# A stalled run gives its slot back so the other runs keep going, and the cap +# on all runs keeps stalled ones from piling up. +sub start_queued { + my ($self) = @_; + + while (scalar($self->{queue}->@*)) { + my $running = scalar(keys $self->{running}->%*); + my $active = scalar(grep { !$_->{stalled} } values $self->{running}->%*); + last if $active >= $self->{max_workers} || $running >= 2 * $self->{max_workers}; + my $key = shift $self->{queue}->@*; + my $run = delete $self->{queued}->{$key} // next; + next if $self->fork_run($run); + + unshift $self->{queue}->@*, $key; + $self->{queued}->{$key} = $run; + last; + } + + return; +} + +sub reap { + my ($self) = @_; + + for my $pid (keys $self->{running}->%*) { + my $res = waitpid($pid, WNOHANG); + next if $res == 0; + + my $run = delete $self->{running}->{$pid}; + delete $self->{by_key}->{ $run->{key} }; + + if ($run->{stalled}) { + my $age = int(time() - $run->{started}); + syslog('info', "hook run '$run->{key}' ended after ${age}s"); + } + # the child logs a hook that dies, so only a signal the runner did not + # send is logged here + if ($res == $pid && ($? & 127) && !$run->{killed_at}) { + syslog('err', "hook run '$run->{key}' killed by signal " . ($? & 127)); + } + + if (my $again = delete $self->{pending}->{ $run->{key} }) { + $self->schedule($again); + } + } + + return; +} + +# Returns whether the run was started. +sub fork_run { + my ($self, $run) = @_; + + my $pid = fork(); + if (!defined($pid)) { + syslog('err', "fork for hook run '$run->{key}' failed: $!") if !$self->{fork_failed}++; + return 0; + } + $self->{fork_failed} = 0; + + if ($pid == 0) { + $self->{watch}->close(); + $SIG{$_} = 'DEFAULT' for qw(CHLD HUP INT TERM QUIT); + + my $rc = 0; + eval { + # the hook reads the current state, the cache inherited from the + # parent may be older + PVE::Cluster::cfs_update(); + $run->{hook}->{code}->($run->{param}, $run->{event}); + }; + if (my $err = $@) { + chomp $err; + syslog('err', "hook run '$run->{key}' failed: $err"); + $rc = 1; + } + POSIX::_exit($rc); + } + + $run->{started} = time(); + $self->{running}->{$pid} = $run; + $self->{by_key}->{ $run->{key} } = $pid; + + return 1; +} + +sub reap_within_grace { + my ($self) = @_; + + my $deadline = time() + $stop_grace; + while (scalar(keys $self->{running}->%*) && time() < $deadline) { + sleep(0.1); + $self->reap(); + } + + return; +} + +sub stop_children { + my ($self) = @_; + + my @pids = keys $self->{running}->%*; + return if !scalar(@pids); + + my $now = time(); + $_->{killed_at} = $now for values $self->{running}->%*; + kill 'TERM', @pids; + $self->reap_within_grace(); + + if (my @left = keys $self->{running}->%*) { + syslog('warning', "killing unresponsive hook runs: " . join(', ', @left)); + kill 'KILL', @left; + # a child blocked on the cluster file system stays in + # uninterruptible sleep until pmxcfs answers, so the kill alone + # does not bound this wait + $self->reap_within_grace(); + + if (my @stuck = keys $self->{running}->%*) { + my $runs = join(', ', map { "'$self->{running}->{$_}->{key}' as $_" } @stuck); + syslog('warning', "hook runs left outstanding by the stop: $runs"); + } + } + + return; +} + +1; diff --git a/src/PVE/Cluster/Makefile b/src/PVE/Cluster/Makefile index 0da7838..47666b2 100644 --- a/src/PVE/Cluster/Makefile +++ b/src/PVE/Cluster/Makefile @@ -1,6 +1,6 @@ PVEDIR=$(DESTDIR)/usr/share/perl5/PVE -SOURCES=Hooks.pm IPCConst.pm Setup.pm Watch.pm +SOURCES=HookRunner.pm Hooks.pm IPCConst.pm Setup.pm Watch.pm .PHONY: install install: $(SOURCES) diff --git a/src/test/Makefile b/src/test/Makefile index de07608..ab4eba2 100644 --- a/src/test/Makefile +++ b/src/test/Makefile @@ -4,7 +4,7 @@ cpgtest: cpgtest.c gcc -Wall cpgtest.c $(shell pkg-config --cflags --libs libcpg libqb) -o cpgtest .PHONY: check install clean distclean -check: corosync-parser-test test-mac-prefix watch-client-test hooks-test +check: corosync-parser-test test-mac-prefix watch-client-test hooks-test hook-runner-test .PHONY: corosync-parser-test corosync-parser-test: @@ -22,5 +22,9 @@ watch-client-test: hooks-test: perl hooks_test.pl +.PHONY: hook-runner-test +hook-runner-test: + perl hook_runner_test.pl + distclean: clean clean: diff --git a/src/test/hook_runner_test.pl b/src/test/hook_runner_test.pl new file mode 100644 index 0000000..571197c --- /dev/null +++ b/src/test/hook_runner_test.pl @@ -0,0 +1,302 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use lib '..'; + +use File::Temp qw(tempdir); +use Test::MockModule; +use Test::More; +use Time::HiRes qw(sleep time); + +use PVE::Cluster::Hooks; +use PVE::Cluster::HookRunner; + +my $dir = tempdir(CLEANUP => 1); +my $log = "$dir/runs"; + +# The runner never talks to pmxcfs here. A scripted watch hands out event +# batches as the daemon would deliver them, parameters included, and +# cfs_update is stubbed. The forks and the bookkeeping around them are real. +my $cluster_mock = Test::MockModule->new('PVE::Cluster'); +$cluster_mock->redefine(cfs_update => sub { }); + +# nothing here belongs in the host's journal +my @logged; +my $runner_mock = Test::MockModule->new('PVE::Cluster::HookRunner'); +$runner_mock->redefine(syslog => sub { push @logged, [@_] }); + +package FakeWatch { + + sub new { + my ($class, @batches) = @_; + return bless { batches => [@batches] }, $class; + } + + sub subscribe { + my ($self, %param) = @_; + $self->{subscription} = \%param; + return; + } + + sub wait { + my ($self, $timeout) = @_; + warn delete($self->{warn}) if $self->{warn}; + my $batch = shift $self->{batches}->@*; + return $batch->@* if $batch; + die $self->{fail} if $self->{fail}; + Time::HiRes::sleep($timeout); + return; + } + + sub close { + my ($self) = @_; + return; + } +} + +my $record = sub { + my ($name, $param, $event) = @_; + open(my $fh, '>>', $log) or die "open: $!"; + my $args = join(',', map { "$_=$param->{$_}" } sort keys $param->%*); + print $fh "$name($args) $event->{type}\n"; + close($fh); + return; +}; + +PVE::Cluster::Hooks->register_hook({ + name => 'guest', + path => 'nodes/{node}/qemu-server/{vmid}.conf', + code => sub { + $record->('guest', @_); + sleep(0.5); + }, +}); + +PVE::Cluster::Hooks->register_hook({ + name => 'dc', + path => 'datacenter.cfg', + code => sub { + $record->('dc', @_); + sleep(0.5); + }, +}); + +PVE::Cluster::Hooks->register_hook({ + name => 'broken', + path => 'broken.cfg', + code => sub { die "on purpose\n" }, +}); + +# hangs on a rename, records only once it got past that +PVE::Cluster::Hooks->register_hook({ + name => 'slow', + path => 'slow/{id}.conf', + code => sub { + my ($param, $event) = @_; + sleep(30) if $event->{type} eq 'rename'; + $record->('slow', @_); + }, +}); + +my $drain = sub { + my ($runner) = @_; + my $deadline = time() + 10; + while (!$runner->idle() && time() < $deadline) { + $runner->run_once(0.05); + } + ok($runner->idle(), 'runner drained'); + return; +}; + +my $lines = sub { + open(my $fh, '<', $log) or return []; + my @lines = <$fh>; + chomp @lines; + unlink($log); + return \@lines; +}; + +my $ev = sub { + my ($type, $path, $to, $params) = @_; + return { + type => $type, + path => $path, + ($to ? (to => $to) : ()), + ($params ? (params => $params) : ()), + }; +}; + +my $guest = sub { + my ($vmid) = @_; + return { guest => [{ node => 'n1', vmid => $vmid }] }; +}; + +# saves of one guest that arrive while the run for the first is still in +# flight collapse into one run plus one rerun +my $save = sub { + return [ + $ev->( + 'rename', + 'nodes/n1/qemu-server/100.conf.tmp.1', + 'nodes/n1/qemu-server/100.conf', + $guest->(100), + ), + ]; +}; +my $watch = FakeWatch->new($save->(), $save->(), $save->()); +my $runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 4); +$runner->run_once(0.05); +is(scalar(keys $runner->{running}->%*), 1, 'one run in flight per key'); +$runner->run_once(0.05); +$runner->run_once(0.05); +is(scalar(keys $runner->{running}->%*), 1, 'later saves do not start a second run'); +is(scalar(keys $runner->{pending}->%*), 1, 'they are folded into one pending rerun'); +$drain->($runner); +is_deeply( + $lines->(), + ['guest(node=n1,vmid=100) rename', 'guest(node=n1,vmid=100) rename'], + 'three saves became one run plus one rerun', +); + +# different keys run concurrently up to the worker cap +$watch = FakeWatch->new([ + $ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100)), + $ev->('write', 'nodes/n1/qemu-server/101.conf', undef, $guest->(101)), + $ev->('write', 'nodes/n1/qemu-server/102.conf', undef, $guest->(102)), + $ev->('write', 'datacenter.cfg', undef, { dc => [{}] }), +]); +$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 2); +$runner->run_once(0.05); +is(scalar(keys $runner->{running}->%*), 2, 'the worker cap holds'); +is(scalar($runner->{queue}->@*), 2, 'the rest waits in the queue'); +$drain->($runner); +is_deeply( + [sort $lines->()->@*], + [ + 'dc() write', + 'guest(node=n1,vmid=100) write', + 'guest(node=n1,vmid=101) write', + 'guest(node=n1,vmid=102) write', + ], + 'every key ran exactly once', +); + +# a resync runs every hook once without parameters +$watch = FakeWatch->new([$ev->('resync')]); +$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 4); +$runner->run_once(0.05); +$drain->($runner); +is_deeply( + [sort $lines->()->@*], + ['dc() resync', 'guest() resync', 'slow() resync'], + 'resync reaches every hook, a failing hook does not disturb the others', +); + +# the runner subscribes to what the registry derived +$watch = FakeWatch->new(); +$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 1); +$runner->{terminate} = 1; +$runner->run(); +is_deeply( + $watch->{subscription}, + PVE::Cluster::Hooks->subscription(), + 'subscription is derived from the registered hooks', +); + +# a watch that dies ends the runner, and the runner stops its children first +$watch = FakeWatch->new([$ev->('write', 'datacenter.cfg', undef, { dc => [{}] })]); +$watch->{fail} = "handshake refused\n"; +$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 4); +eval { $runner->run() }; +like($@, qr/handshake refused/, 'a failing watch ends the runner'); +is(scalar(keys $runner->{running}->%*), 0, 'and no child is left behind'); +$lines->(); + +@logged = (); +$watch = FakeWatch->new([$ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100))]); +$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 4); +$runner->run_once(0.05); +$runner->stop_children(); +$runner->run_once(0.05); +is(scalar(keys $runner->{running}->%*), 0, 'the run was stopped'); +ok(!(grep { $_->[0] eq 'err' } @logged), 'a run interrupted by the stop is not an error'); +$lines->(); + +@logged = (); +$watch = FakeWatch->new(); +$watch->{warn} = "notification socket closed, reconnecting\n"; +$watch->{fail} = "stop\n"; +$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 1); +eval { $runner->run() }; +is_deeply( + [grep { $_->[0] eq 'warning' } @logged], + [['warning', 'notification socket closed, reconnecting']], + 'a warning from the watch is logged', +); + +# a stalled run frees its slot, is logged, and is killed by the next event +# for its key while the other keys keep running +@logged = (); +my $slow = sub { + my ($hang) = @_; + my $params = { slow => [{ id => 1 }] }; + return $ev->('rename', 'slow/1.conf.tmp', 'slow/1.conf', $params) if $hang; + return $ev->('write', 'slow/1.conf', undef, $params); +}; +$watch = FakeWatch->new( + [$slow->(1)], + [$ev->('write', 'datacenter.cfg', undef, { dc => [{}] })], + [$slow->()], +); +$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 1, stall_age => 0.5); +$runner->run_once(0.05); +is(scalar(keys $runner->{running}->%*), 1, 'the hanging run holds the only slot'); +sleep(0.6); +$runner->run_once(0.05); +is(scalar(keys $runner->{running}->%*), 2, 'past the stall age another key runs beside it'); +ok( + (grep { $_->[0] eq 'warning' && $_->[1] =~ /^hook run 'slow:.*still running after/ } @logged), + 'the stall is logged', +); +$runner->run_once(0.05); +$drain->($runner); +is_deeply( + [sort $lines->()->@*], + ['dc() write', 'slow(id=1) write'], + 'the next event for the key replaced the stalled run with a rerun', +); +ok( + (grep { $_->[0] eq 'warning' && $_->[1] =~ /^replacing stalled hook run 'slow:/ } @logged), + 'and said so', +); +ok((grep { $_->[0] eq 'info' && $_->[1] =~ /ended after/ } @logged), 'the end is logged too'); +ok(!(grep { $_->[0] eq 'err' } @logged), 'a run the runner killed is not an error'); + +# a child in uninterruptible sleep outlives the kill. A clock that jumps +# past every deadline leaves the child unreaped the same way, without +# waiting +@logged = (); +$watch = FakeWatch->new([$ev->('write', 'nodes/n1/qemu-server/100.conf', undef, $guest->(100))]); +$runner = PVE::Cluster::HookRunner->new(watch => $watch, max_workers => 4); +$runner->run_once(0.05); +my ($stuck) = keys $runner->{running}->%*; +my $clock = time(); +$runner_mock->redefine(time => sub () { $clock += 10 }); +$runner->stop_children(); +$runner_mock->unmock('time'); +is_deeply([keys $runner->{running}->%*], [$stuck], 'the stop gives up on a run it cannot reap'); +ok( + ( + grep { + $_->[0] eq 'warning' + && $_->[1] =~ /^hook runs left outstanding by the stop: 'guest:/ + } @logged + ), + 'and says which ones are left', +); +waitpid($stuck, 0); +$lines->(); + +done_testing(); -- 2.47.3