[PATCH pve-cluster 10/12] cfs: add perl client for the change notification socket

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

Daemons that want to react to cluster-wide config changes all need to
connect to the pmxcfs notification socket, subscribe with path
templates, wait for events, and recover when pmxcfs restarts. A shared
client module handles that, so every consumer gets the reconnect
handling right instead of reimplementing it.

The client retries a lost connection and a failed connect on its own,
with a growing backoff between failed attempts. An error reply from
the daemon or an unsupported protocol version makes the call die
instead, since a retry would get the same answer. Two processes
reading one stream would split its lines between them, so a forked
child opens a connection of its own.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 debian/pve-cluster.install    |   1 +
 src/PVE/Cluster/Makefile      |   2 +-
 src/PVE/Cluster/Watch.pm      | 325 ++++++++++++++++++++++++++++++++++
 src/test/Makefile             |   6 +-
 src/test/watch_client_test.pl | 229 ++++++++++++++++++++++++
 5 files changed, 561 insertions(+), 2 deletions(-)
 create mode 100644 src/PVE/Cluster/Watch.pm
 create mode 100644 src/test/watch_client_test.pl

diff --git a/debian/pve-cluster.install b/debian/pve-cluster.install
index f66cd06..77e3244 100644
--- a/debian/pve-cluster.install
+++ b/debian/pve-cluster.install
@@ -5,4 +5,5 @@ usr/lib/
 usr/share/man/man8/pmxcfs.8
 usr/share/perl5/PVE/Cluster.pm
 usr/share/perl5/PVE/Cluster/IPCConst.pm
+usr/share/perl5/PVE/Cluster/Watch.pm
 usr/share/perl5/PVE/IPCC.pm
diff --git a/src/PVE/Cluster/Makefile b/src/PVE/Cluster/Makefile
index 3f920cb..7beb976 100644
--- a/src/PVE/Cluster/Makefile
+++ b/src/PVE/Cluster/Makefile
@@ -1,6 +1,6 @@
 PVEDIR=$(DESTDIR)/usr/share/perl5/PVE
 
-SOURCES=IPCConst.pm Setup.pm
+SOURCES=IPCConst.pm Setup.pm Watch.pm
 
 .PHONY: install
 install: $(SOURCES)
diff --git a/src/PVE/Cluster/Watch.pm b/src/PVE/Cluster/Watch.pm
new file mode 100644
index 0000000..dde8711
--- /dev/null
+++ b/src/PVE/Cluster/Watch.pm
@@ -0,0 +1,325 @@
+package PVE::Cluster::Watch;
+
+use strict;
+use warnings;
+
+use IO::Select;
+use IO::Socket::UNIX;
+use JSON;
+use Socket qw(SOCK_STREAM MSG_NOSIGNAL);
+use Time::HiRes qw(time sleep);
+
+my $default_socket = '/run/pve-cluster/pmxcfs.sock';
+my $protocol_version = 1;
+my $io_timeout = 10;
+my $min_backoff = 1;
+my $max_backoff = 30;
+
+sub new {
+    my ($class, %param) = @_;
+
+    my $self = bless {
+        socket_path => $param{socket} // $default_socket,
+        patterns => {},
+        pid => $$,
+    }, $class;
+    $self->reset();
+
+    return $self;
+}
+
+# Disconnected, with a connect due right away.
+sub reset {
+    my ($self) = @_;
+
+    $self->{sock} = undef;
+    $self->{buf} = '';
+    $self->{queue} = [];
+    $self->{established} = 0;
+    $self->{next_connect} = 0;
+    $self->{backoff} = $min_backoff;
+    $self->{warned} = 0;
+
+    return;
+}
+
+# Two processes reading one connection would split its lines between them,
+# so a forked child connects on its own, as IPCC does. It closes the
+# inherited copy without a shutdown, which would cut off the parent as well.
+# A child that keeps running should close the watch, otherwise that copy
+# stays open until the child next calls a method of the watch.
+sub check_fork {
+    my ($self) = @_;
+
+    return if $self->{pid} == $$;
+
+    $self->{sock}->close() if $self->{sock};
+    $self->{pid} = $$;
+    $self->reset();
+
+    return;
+}
+
+# Takes effect when connected and is replayed on every reconnect. Dies if
+# the daemon refuses it. A connection lost while sending it is dropped, and
+# the next wait reconnects.
+sub subscribe {
+    my ($self, %param) = @_;
+
+    $self->check_fork();
+
+    my $previous = $self->{patterns};
+    $self->{patterns} = { ($param{patterns} // {})->%* };
+
+    return if !$self->{sock};
+
+    my $error = eval { $self->send_subscription() };
+    if ($@) {
+        chomp(my $reason = $@);
+        $self->disconnect($reason) if $self->{sock};
+        return;
+    }
+    # the daemon keeps serving the previous subscription, so keep its
+    # patterns for the next reconnect
+    if (defined($error)) {
+        $self->{patterns} = $previous;
+        die "subscription refused: $error\n";
+    }
+
+    return;
+}
+
+sub connected {
+    my ($self) = @_;
+
+    $self->check_fork();
+
+    return defined($self->{sock});
+}
+
+# A reconnect replaces the socket, possibly under the same number, so a
+# caller watching it has to look it up again after every wait. It is undef
+# while disconnected, and only wait reconnects.
+sub fd {
+    my ($self) = @_;
+
+    $self->check_fork();
+
+    return $self->{sock} ? fileno($self->{sock}) : undef;
+}
+
+sub close {
+    my ($self) = @_;
+
+    $self->check_fork();
+
+    $self->{sock}->close() if $self->{sock};
+    $self->{sock} = undef;
+    $self->{established} = 0;
+    $self->{buf} = '';
+
+    return;
+}
+
+# Returns the queued events as soon as there are any, or none once $timeout
+# seconds have passed or a signal has arrived. A due reconnect also happens
+# here and is the one way a zero timeout can block. Dies if the daemon
+# refuses the handshake, since a retry would not change that.
+sub wait {
+    my ($self, $timeout) = @_;
+
+    $self->check_fork();
+
+    my $deadline = time() + ($timeout // 0);
+
+    while (1) {
+        if (scalar($self->{queue}->@*)) {
+            my @events = $self->{queue}->@*;
+            $self->{queue} = [];
+            return @events;
+        }
+
+        my $now = time();
+        if (!$self->{sock}) {
+            if ($now >= $self->{next_connect}) {
+                $self->connect();
+                next;
+            }
+            my $until = $self->{next_connect} < $deadline ? $self->{next_connect} : $deadline;
+            return if $until <= $now;
+            sleep($until - $now);
+            # a signal cut the sleep short, and a caller whose handler only
+            # sets a flag needs control back to act on it
+            return if time() < $until;
+            next;
+        }
+
+        my $remaining = $deadline - $now;
+        $remaining = 0 if $remaining < 0;
+
+        return if !IO::Select->new($self->{sock})->can_read($remaining);
+        $self->read_messages();
+    }
+}
+
+sub connect {
+    my ($self) = @_;
+
+    # connect blocks while the daemon's backlog is full, as when it cannot
+    # accept for a while. The timeout bounds that, and a failed attempt is
+    # retried later
+    my $sock = IO::Socket::UNIX->new(
+        Type => SOCK_STREAM,
+        Peer => $self->{socket_path},
+        Timeout => $io_timeout,
+    );
+    if (!$sock) {
+        $self->connect_failed("connect to $self->{socket_path} failed: $!");
+        return;
+    }
+
+    $self->{sock} = $sock;
+
+    my $error = eval { $self->handshake() };
+    if (my $reason = $@) {
+        $self->close();
+        $self->connect_failed("handshake with $self->{socket_path} failed: $reason");
+        return;
+    }
+    if (defined($error)) {
+        $self->close();
+        $self->delay_retry();
+        die "handshake with $self->{socket_path} refused: $error\n";
+    }
+
+    $self->{established} = 1;
+    $self->{backoff} = $min_backoff;
+    $self->{warned} = 0;
+
+    return;
+}
+
+sub handshake {
+    my ($self) = @_;
+
+    my ($hello, $error) = $self->request('hello');
+    return $error if defined($error);
+
+    my $version = ref($hello) eq 'HASH' ? $hello->{protocol} : undef;
+    return "unsupported protocol version " . ($version // 'unknown')
+        if !defined($version) || $version != $protocol_version;
+
+    return $self->send_subscription();
+}
+
+sub connect_failed {
+    my ($self, $msg) = @_;
+
+    if (!$self->{warned}) {
+        chomp $msg;
+        warn "$msg, retrying\n";
+        $self->{warned} = 1;
+    }
+
+    $self->delay_retry();
+
+    return;
+}
+
+sub delay_retry {
+    my ($self) = @_;
+
+    $self->{next_connect} = time() + $self->{backoff};
+    $self->{backoff} *= 2;
+    $self->{backoff} = $max_backoff if $self->{backoff} > $max_backoff;
+
+    return;
+}
+
+# A connection lost before its handshake completed is reported by the
+# connect attempt, so a daemon that accepts and then closes causes one
+# warning, not two.
+sub disconnect {
+    my ($self, $reason) = @_;
+
+    my $established = $self->{established};
+    $self->close();
+    return if !$established;
+
+    warn "notification socket $reason, reconnecting\n";
+    $self->{next_connect} = time();
+    $self->{backoff} = $min_backoff;
+
+    return;
+}
+
+sub send_subscription {
+    my ($self) = @_;
+
+    my (undef, $error) = $self->request('subscribe', { patterns => $self->{patterns} });
+
+    return $error;
+}
+
+# Sends a request and returns its reply as (data, error), queueing any events
+# that arrive meanwhile. Dies when no reply comes in time or the connection
+# is lost.
+sub request {
+    my ($self, $command, $args) = @_;
+
+    my $line = encode_json({ command => $command, ($args ? (args => $args) : ()) }) . "\n";
+    my $written = 0;
+    while ($written < length($line)) {
+        my $len = send($self->{sock}, substr($line, $written), MSG_NOSIGNAL);
+        next if !defined($len) && $!{EINTR};
+        die "write failed: $!\n" if !defined($len);
+        $written += $len;
+    }
+
+    my $deadline = time() + $io_timeout;
+    while (1) {
+        my $remaining = $deadline - time();
+        die "timeout waiting for reply to '$command'\n" if $remaining <= 0;
+        # bounds the wait, since the read itself would block for as long as
+        # the daemon stays silent
+        next if !IO::Select->new($self->{sock})->can_read($remaining);
+
+        for my $msg ($self->read_messages()) {
+            return ($msg->{ok}, undef) if exists $msg->{ok};
+            return (undef, $msg->{error} // 'unknown error') if exists $msg->{error};
+        }
+        die "connection lost while waiting for reply to '$command'\n" if !$self->{sock};
+    }
+}
+
+sub read_messages {
+    my ($self) = @_;
+
+    my $len;
+    do {
+        # appended to what the last read left unfinished
+        $len = sysread($self->{sock}, $self->{buf}, 65536, length($self->{buf}));
+    } while (!defined($len) && $!{EINTR});
+    if (!$len) {
+        $self->disconnect(defined($len) ? 'closed' : "read failed: $!");
+        return;
+    }
+
+    my @replies;
+    while ($self->{buf} =~ s/^([^\n]*)\n//) {
+        my $msg = eval { decode_json($1) };
+        if (ref($msg) ne 'HASH') {
+            $self->disconnect('sent a malformed line');
+            return;
+        }
+        if (my $event = $msg->{event}) {
+            push $self->{queue}->@*, $event;
+        } else {
+            push @replies, $msg;
+        }
+    }
+
+    return @replies;
+}
+
+1;
diff --git a/src/test/Makefile b/src/test/Makefile
index cdd37d0..7b36fca 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
+check: corosync-parser-test test-mac-prefix watch-client-test
 
 .PHONY: corosync-parser-test
 corosync-parser-test:
@@ -14,5 +14,9 @@ corosync-parser-test:
 test-mac-prefix:
 	perl test_mac_prefix.pl
 
+.PHONY: watch-client-test
+watch-client-test:
+	perl watch_client_test.pl
+
 distclean: clean
 clean:
diff --git a/src/test/watch_client_test.pl b/src/test/watch_client_test.pl
new file mode 100644
index 0000000..ef8159e
--- /dev/null
+++ b/src/test/watch_client_test.pl
@@ -0,0 +1,229 @@
+#!/usr/bin/perl
+
+use lib '..';
+
+use strict;
+use warnings;
+
+use File::Temp qw(tempdir);
+use IO::Socket::UNIX;
+use JSON;
+use Socket qw(SOCK_STREAM);
+use Test::More;
+use Time::HiRes qw(sleep time);
+
+use PVE::Cluster::Watch;
+
+my $dir = tempdir(CLEANUP => 1);
+my $path = "$dir/sock";
+
+my $listener = IO::Socket::UNIX->new(Type => SOCK_STREAM, Local => $path, Listen => 1)
+    or die "listen failed: $!\n";
+
+# A stand-in for pmxcfs that serves ten connections in a row. The fourth
+# refuses the subscription, the fifth hangs up after answering the hello,
+# the seventh refuses a second subscription on a live connection and the
+# last three hang up before answering anything. The others answer the
+# handshake, send the resync, mirror the subscription back inside two
+# events and hang up, which forces the client through its reconnect path.
+my $server = fork() // die "fork failed: $!\n";
+if (!$server) {
+    for my $round (1 .. 10) {
+        my $conn = $listener->accept() or die "accept failed: $!\n";
+        if ($round >= 8) {
+            close($conn);
+            next;
+        }
+        my $subscribed = 0;
+        while (my $line = <$conn>) {
+            my $req = decode_json($line);
+            if ($req->{command} eq 'hello') {
+                print $conn encode_json({ ok => { protocol => 1 } }), "\n";
+            } elsif ($round == 5) {
+                last;
+            } elsif ($req->{command} eq 'subscribe') {
+                if ($round == 4) {
+                    print $conn encode_json({ error => "pattern 'guest': no good" }), "\n";
+                    last;
+                }
+                if ($round == 7 && $subscribed) {
+                    print $conn encode_json({ error => "pattern 'bad': no good" }), "\n";
+                    next;
+                }
+                $subscribed = 1;
+                my $patterns = $req->{args}->{patterns};
+                print $conn encode_json({ ok => undef }), "\n";
+                print $conn encode_json({ event => { type => 'resync' } }), "\n";
+                next if $round == 7;
+                print $conn encode_json({
+                    event => {
+                        type => 'write',
+                        path => 'x',
+                        params =>
+                            { map { $_ => [{ template => $patterns->{$_} }] } keys %$patterns },
+                    },
+                    }),
+                    "\n";
+                print $conn encode_json({
+                    event => {
+                        type => 'rename',
+                        path => 'old',
+                        to => join(',', sort keys %$patterns),
+                    },
+                    }),
+                    "\n";
+                last;
+            } else {
+                print $conn encode_json({ error => "unknown command '$req->{command}'" }), "\n";
+            }
+        }
+        close($conn);
+    }
+    exit(0);
+}
+close($listener);
+
+my $collect = sub {
+    my ($watch, $count, $timeout) = @_;
+    my @events;
+    my $deadline = time() + 10;
+    while (scalar(@events) < $count && time() < $deadline) {
+        push @events, $watch->wait($timeout);
+        sleep(0.05) if !$timeout;
+    }
+    return \@events;
+};
+
+my $watch = PVE::Cluster::Watch->new(socket => $path);
+ok(!$watch->connected(), 'not connected before the first wait');
+
+my $template = 'nodes/{node}/qemu-server/{vmid}.conf';
+$watch->subscribe(patterns => { guest => $template });
+
+my $write = {
+    type => 'write',
+    path => 'x',
+    params => { guest => [{ template => $template }] },
+};
+my $rename = { type => 'rename', path => 'old', to => 'guest' };
+my $expected = [{ type => 'resync' }, $write, $rename];
+
+is_deeply($collect->($watch, 3, 1), $expected, 'connect gives a resync followed by the events');
+ok(defined($watch->fd()), 'a descriptor is available while connected');
+
+$watch->{pid} = -1; # pretend the object was inherited across a fork
+is_deeply(
+    $collect->($watch, 3, 0),
+    $expected,
+    'a forked child gets a connection of its own, a zero timeout drains it',
+);
+
+is_deeply(
+    $collect->($watch, 3, 1),
+    $expected,
+    'reconnect after hangup replays the subscription and gives a resync again',
+);
+
+# the server has hung up on that connection by now. Writing to it must
+# neither raise SIGPIPE nor die, and the next wait reconnects
+sleep(0.2);
+$watch->subscribe(patterns => { guest => $template });
+ok(!$watch->connected(), 'a subscription on a lost connection drops it');
+
+eval { $watch->wait(1) };
+like($@, qr/refused: pattern 'guest': no good/, 'a refused subscription makes the wait die');
+ok(!$watch->connected(), 'and its connection is closed');
+
+# a handshake cut after the hello, rounds five and six
+{
+    my @warnings;
+    local $SIG{__WARN__} = sub { push @warnings, $_[0] };
+    my $half = PVE::Cluster::Watch->new(socket => $path);
+    $half->subscribe(patterns => { guest => $template });
+    is_deeply(
+        $collect->($half, 3, 1),
+        $expected,
+        'a connect that is cut after the hello is retried and gives a resync',
+    );
+    is(scalar(@warnings), 1, 'and the cut handshake is reported');
+}
+
+# a refused subscription on a live connection, round seven. The connection
+# stays up after the refusal, as it does with pmxcfs
+{
+    my $live = PVE::Cluster::Watch->new(socket => $path);
+    $live->subscribe(patterns => { guest => $template });
+    is_deeply(
+        $collect->($live, 1, 1),
+        [{ type => 'resync' }],
+        'connected with the first subscription',
+    );
+    is_deeply(
+        [$live->request('bogus')],
+        [undef, "unknown command 'bogus'"],
+        'an error reply outside the handshake comes back as the error',
+    );
+    eval { $live->subscribe(patterns => { bad => 'x/{y}' }) };
+    like($@, qr/subscription refused: pattern 'bad': no good/, 'a refusal makes subscribe die');
+    is_deeply($live->{patterns}, { guest => $template }, 'and the previous patterns stay');
+    ok($live->connected(), 'on a connection that stays up');
+}
+
+# a daemon that accepts and hangs up immediately, rounds eight to ten
+{
+    my @warnings;
+    local $SIG{__WARN__} = sub { push @warnings, $_[0] };
+    my $cut = PVE::Cluster::Watch->new(socket => $path);
+    $cut->subscribe(patterns => { guest => $template });
+    my @none = $cut->wait(5);
+    is(scalar(@none), 0, 'nothing arrives from a daemon that hangs up immediately');
+    is(scalar(@warnings), 1, 'which is reported once and retried with backoff');
+    ok(!$cut->connected(), 'and leaves the client disconnected');
+}
+
+# a signal that cuts a wait short while the client waits for its next
+# connect attempt
+{
+    my @warnings;
+    local $SIG{__WARN__} = sub { push @warnings, $_[0] };
+    my $flagged = 0;
+    local $SIG{USR1} = sub { $flagged = 1 };
+
+    my $parent = $$;
+    my $signaller = fork() // die "fork failed: $!\n";
+    if (!$signaller) {
+        sleep(0.3);
+        kill 'USR1', $parent;
+        exit(0);
+    }
+
+    my $gone = PVE::Cluster::Watch->new(socket => "$dir/gone");
+    my $started = time();
+    my @none = $gone->wait(3);
+    my $elapsed = time() - $started;
+    waitpid($signaller, 0);
+
+    is(scalar(@none), 0, 'a wait cut short by a signal returns no events');
+    ok($flagged, 'the handler of the caller has run');
+    # a signal that lands after the first backoff sleep ended only cuts the
+    # second one short, so the bound has to sit above one backoff and still
+    # below the deadline
+    ok(
+        $elapsed < 2.5,
+        sprintf('and the wait gave up after %.2fs, short of its deadline', $elapsed),
+    );
+    is(scalar(@warnings), 1, 'the socket that is not there is reported once');
+}
+
+# a shortfall in connections must fail the test, not hang it
+local $SIG{ALRM} = sub { kill 'KILL', $server; die "fake server did not finish\n" };
+alarm(30);
+waitpid($server, 0);
+alarm(0);
+is($?, 0, 'fake server ran all rounds cleanly');
+
+my @none = $watch->wait(1);
+is(scalar(@none), 0, 'no events once the server is gone');
+ok(!$watch->connected(), 'disconnected after the server went away');
+
+done_testing();
-- 
2.47.3