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 1FFE51FF0C2 for ; Thu, 22 Oct 2026 16:24:02 +0200 (CEST) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id 12783217C0; Thu, 22 Oct 2026 16:23:03 +0200 (CEST) From: Hannes Laimer To: pve-devel@lists.proxmox.com Subject: [PATCH pve-cluster 10/12] cfs: add perl client for the change notification socket Date: Thu, 22 Oct 2026 16:22:03 +0200 Message-ID: <20261022142205.930536-11-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: 1792678938518 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.341 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) 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: LX5YCT2LW5GWTP2ODJHUYVIBCUNUPYS2 X-Message-ID-Hash: LX5YCT2LW5GWTP2ODJHUYVIBCUNUPYS2 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: 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 --- 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