[PATCH pve-cluster 12/12] cfs: add runner executing cluster change hooks in children

[Hannes Laimer <h.laimer@proxmox.com>]

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 <h.laimer@proxmox.com>
---
 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