diff --git a/src/literals.py b/src/literals.py index 90a5d352..46b90852 100644 --- a/src/literals.py +++ b/src/literals.py @@ -28,6 +28,7 @@ TOPOLOGY_OBSERVER_LOG_FILENAME = "topology_observer.log" TOPOLOGY_OBSERVER_TLS_CA_FILENAME = "valkey_ca.pem" TOPOLOGY_OBSERVER_PID_FILENAME = "topology_observer.pid" +TOPOLOGY_OBSERVER_SIGNATURE_FILENAME = "topology_observer.signature" PEER_RELATION = "valkey-peers" STATUS_PEERS_RELATION = "status-peers" diff --git a/src/managers/topology.py b/src/managers/topology.py index 950110d9..a5e3b15a 100644 --- a/src/managers/topology.py +++ b/src/managers/topology.py @@ -8,6 +8,7 @@ import os import signal import subprocess +from hashlib import sha256 from pathlib import Path from sys import version_info @@ -18,6 +19,7 @@ SENTINEL_TLS_PORT, TOPOLOGY_OBSERVER_LOG_FILENAME, TOPOLOGY_OBSERVER_PID_FILENAME, + TOPOLOGY_OBSERVER_SIGNATURE_FILENAME, TOPOLOGY_OBSERVER_TLS_CA_FILENAME, CharmUsers, ) @@ -50,16 +52,64 @@ def _pid_file_path(self) -> Path: """Return the path to the topology observer pid file.""" return self.state.charm.charm_dir / TOPOLOGY_OBSERVER_PID_FILENAME + @property + def _signature_file_path(self) -> Path: + """Return the path to the topology observer signature file.""" + return self.state.charm.charm_dir / TOPOLOGY_OBSERVER_SIGNATURE_FILENAME + + @property + def _observer_hosts(self) -> str: + """Return the Sentinel host list the observer subprocess is launched with.""" + started_servers = [ + unit.get_endpoint(self.state.substrate) + for unit in self.state.servers + if unit.is_active + ] + port = SENTINEL_TLS_PORT if self.state.unit_server.is_tls_enabled else SENTINEL_PORT + return ",".join(sorted([f"{server}:{port}" for server in started_servers])) + + def observer_signature(self) -> str: + """Return a digest of the arguments the observer subprocess is launched with. + + Recorded next to the PID so a re-delivered event can tell an observer that + is already watching the right topology from one that has to be relaunched. + Hashed because the Sentinel password is one of those arguments. + """ + parts = [ + self._observer_hosts, + str(self.state.unit_server.is_tls_enabled), + CharmUsers.SENTINEL_CHARM_ADMIN.value, + self.state.cluster.internal_users_credentials.get( + CharmUsers.SENTINEL_CHARM_ADMIN.value, "" + ), + ] + if self.state.unit_server.is_tls_enabled: + # the observer is handed its own copy of the CA, so a rotation has to relaunch + # it; an unreadable CA yields "" and reads as a change, which is the safe way + # round -- start_observer surfaces the real error when it re-reads the file + try: + parts.append(self.workload.read_file(self.workload.tls_paths.client_ca)) + except Exception: # noqa: BLE001 + logger.debug("Could not read the client CA while signing the observer") + parts.append("") + + return sha256("|".join(parts).encode()).hexdigest() + + def _is_observer_running(self) -> bool: + """Return whether the recorded observer process is still alive.""" + if (observer_pid := self._read_observer_pid()) == 0: + return False + try: + os.kill(observer_pid, 0) + return True + except OSError: + logger.debug("Topology observer not running") + return False + def start_observer(self) -> None: """Start the topology observer as a subprocess.""" - if (observer_pid := self._read_observer_pid()) != 0: - try: - # check if the process already runs - os.kill(int(observer_pid), 0) - return - except OSError: - logger.debug("Topology observer not running") - pass + if self._is_observer_running(): + return # Generate the venv path based on the existing lib path env = os.environ.copy() @@ -79,13 +129,9 @@ def start_observer(self) -> None: break # Gather Valkey hosts for connection - started_servers = [ - unit.get_endpoint(self.state.substrate) - for unit in self.state.servers - if unit.is_active - ] - port = SENTINEL_TLS_PORT if self.state.unit_server.is_tls_enabled else SENTINEL_PORT - hosts = ",".join(sorted([f"{server}:{port}" for server in started_servers])) + hosts = self._observer_hosts + # captured before the spawn so the record matches what the process was given + signature = self.observer_signature() if self.state.unit_server.is_tls_enabled: # Store current TLS CA cert on operator container @@ -113,10 +159,13 @@ def start_observer(self) -> None: ).pid self._pid_file_path.write_text(str(pid)) + self._signature_file_path.write_text(signature) logging.info(f"Started topology observer process with PID {pid}") def stop_observer(self) -> None: """Stop the topology observer.""" + self._signature_file_path.unlink(missing_ok=True) + if (observer_pid := self._read_observer_pid()) == 0: logger.debug("Topology observer already stopped") return @@ -128,10 +177,24 @@ def stop_observer(self) -> None: except OSError: pass finally: - self._pid_file_path.unlink() + self._pid_file_path.unlink(missing_ok=True) def restart_observer(self) -> None: - """Stop and start the topology observer to pickup host changes.""" + """Relaunch the topology observer if it is gone or watching a stale topology. + + The leader calls this on every peer relation-changed, and the observer only + reports a primary change against the one it saw last -- a value it keeps in + memory. Relaunching a healthy observer therefore throws that value away and + leaves the cluster unwatched for the length of a Python start-up, so a + failover landing in that window is never dispatched. Leave it alone unless + an argument it was launched with has actually changed. + """ + if self._is_observer_running() and self._read_observer_signature() == ( + self.observer_signature() + ): + logger.debug("Topology observer already watching the current topology") + return + self.stop_observer() self.start_observer() @@ -139,5 +202,12 @@ def _read_observer_pid(self) -> int: """Read the pid file of the topology observer and return the pid, or 0 if none.""" try: return int(self._pid_file_path.read_text()) - except (FileNotFoundError, PermissionError): + except (FileNotFoundError, PermissionError, ValueError): return 0 + + def _read_observer_signature(self) -> str: + """Read the signature of the running observer, or "" if there is none recorded.""" + try: + return self._signature_file_path.read_text() + except (FileNotFoundError, PermissionError): + return "" diff --git a/tests/integration/clients/test_client_relation.py b/tests/integration/clients/test_client_relation.py index 248208d7..328c3e3a 100644 --- a/tests/integration/clients/test_client_relation.py +++ b/tests/integration/clients/test_client_relation.py @@ -28,6 +28,8 @@ get_cluster_addresses, get_password, get_primary_ip, + has_leader_settled_since, + utc_now, wait_for_failover, ) @@ -186,6 +188,10 @@ def test_failover_topology_update(juju: jubilant.Juju) -> None: old_primary_ip = get_primary_ip(juju, APP_NAME) logger.info("Initiate failover through Sentinel %s", ip_address) + # Anchor before the failover: the gate below has to wait for the charm to react to it, + # and until the leader is woken it still reports the idle it entered beforehand. + failover_requested_at = utc_now() + failover_result = exec_valkey_cli( hostname=ip_address, username=CharmUsers.SENTINEL_CHARM_ADMIN, @@ -197,8 +203,12 @@ def test_failover_topology_update(juju: jubilant.Juju) -> None: assert failover_result == "OK", "Failover not successful" wait_for_failover(juju, APP_NAME, old_primary_ip=old_primary_ip, unit_count=NUM_UNITS) + # Sentinel has switched, but the client still reaches Valkey through whatever the charm + # last published; `topology_changed` is what re-points it, and only the leader handles it. juju.wait( - lambda status: are_agents_idle(status, APP_NAME, idle_period=60, unit_count=NUM_UNITS), + lambda status: has_leader_settled_since( + status, APP_NAME, since=failover_requested_at, idle_period=60 + ), timeout=600, ) diff --git a/tests/integration/helpers.py b/tests/integration/helpers.py index 348d09c5..5cf1bb17 100644 --- a/tests/integration/helpers.py +++ b/tests/integration/helpers.py @@ -7,7 +7,7 @@ import re import subprocess from contextlib import contextmanager -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from pathlib import Path from typing import List, Literal, NamedTuple @@ -193,15 +193,48 @@ def are_agents_idle( ) +def utc_now() -> datetime: + """Return the current UTC time, naive, to compare against Juju status timestamps. + + Juju reports timestamps in UTC and they are parsed with `ignoretz`, so the value they + are compared against has to be UTC too. `datetime.now()` is local: anywhere east of + Greenwich it runs ahead, and every elapsed-time check passes by that offset alone. + """ + return datetime.now(timezone.utc).replace(tzinfo=None) + + def _check_apps_idle_period(status: jubilant.Status, *apps: str, idle_period: int) -> bool: return all( - parse(unit.juju_status.since, ignoretz=True) + timedelta(seconds=idle_period) - < datetime.now() + parse(unit.juju_status.since, ignoretz=True) + timedelta(seconds=idle_period) < utc_now() for app in apps for unit in status.get_units(app).values() ) +def has_leader_settled_since( + status: jubilant.Status, app: str, since: datetime, idle_period: int = 0 +) -> bool: + """Whether the leader has run a hook and gone idle again after `since` (naive UTC). + + An `idle_period` on its own cannot express "the charm has reacted to X". An agent that + has not been woken yet still carries the idle timestamp it had before X, and that + timestamp already satisfies any period, so the wait returns on its first poll and the + test races ahead of the charm. Anchoring on `since` waits for the leader to actually + pick the event up; the period then lets any deferral it left behind drain. + + Only the leader is considered: handlers such as `topology_changed` are leader-only and + write no peer data, so the other units are never woken and would never settle. + """ + for unit in status.get_units(app).values(): + if not unit.leader: + continue + if unit.juju_status.current != "idle": + return False + idle_since = parse(unit.juju_status.since, ignoretz=True) + return idle_since >= since and idle_since + timedelta(seconds=idle_period) < utc_now() + return False + + def verify_unit_count( status: jubilant.Status, *apps: str, unit_count: int | dict[str, int] | None = None ): diff --git a/tests/unit/test_topology_observer.py b/tests/unit/test_topology_observer.py new file mode 100644 index 00000000..c5ea7e18 --- /dev/null +++ b/tests/unit/test_topology_observer.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +# Copyright 2026 Canonical Ltd. +# See LICENSE file for licensing details. + +"""Unit tests for the TopologyManager observer lifecycle.""" + +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from literals import ( + TOPOLOGY_OBSERVER_PID_FILENAME, + TOPOLOGY_OBSERVER_SIGNATURE_FILENAME, + CharmUsers, +) +from managers.topology import TopologyManager + + +def _make_manager( + charm_dir: Path, + pid: int | None = 1234, + signature: str | None = None, + endpoints: tuple[str, ...] = ("10.0.0.1", "10.0.0.2"), + tls: bool = False, + password: str = "sentinel-password", + client_ca: str = "ca-pem", +): + """Build a TopologyManager whose observer inputs are fully determined. + + `pid` and `signature` are what the unit recorded for the observer it last + launched; `signature` defaults to the digest of the very inputs given here, + i.e. the running observer was launched with exactly the current topology. + `pid=None` records no observer at all. + """ + state = MagicMock() + state.charm.charm_dir = charm_dir + state.unit_server.is_tls_enabled = tls + servers = [] + for endpoint in endpoints: + server = MagicMock(is_active=True) + server.get_endpoint.return_value = endpoint + servers.append(server) + state.servers = servers + state.cluster.internal_users_credentials = {CharmUsers.SENTINEL_CHARM_ADMIN.value: password} + + workload = MagicMock() + workload.read_file.return_value = client_ca + manager = TopologyManager(state=state, workload=workload) + + if pid is not None: + (charm_dir / TOPOLOGY_OBSERVER_PID_FILENAME).write_text(str(pid)) + (charm_dir / TOPOLOGY_OBSERVER_SIGNATURE_FILENAME).write_text( + manager.observer_signature() if signature is None else signature + ) + return manager, state + + +@pytest.fixture +def charm_dir(tmp_path: Path) -> Path: + """Return a throwaway charm directory to hold the observer's bookkeeping files.""" + return tmp_path + + +def test_restart_is_noop_when_running_and_inputs_unchanged(charm_dir: Path): + """A re-delivered peer event must not churn the observer. + + The observer only reports a primary change against the one it saw last, and + it keeps that in memory -- a needless relaunch throws it away and leaves the + cluster unwatched while the replacement starts up. + """ + manager, _ = _make_manager(charm_dir) + + with ( + patch("managers.topology.os.kill") as mock_kill, + patch.object(TopologyManager, "stop_observer") as mock_stop, + patch.object(TopologyManager, "start_observer") as mock_start, + ): + manager.restart_observer() + + mock_stop.assert_not_called() + mock_start.assert_not_called() + # only the liveness probe, never a signal + mock_kill.assert_called_once_with(1234, 0) + + +def test_restart_when_topology_changed(charm_dir: Path): + """A changed host set must relaunch the observer with the new arguments.""" + manager, _ = _make_manager(charm_dir, signature="stale-digest-from-a-different-topology") + + with ( + patch("managers.topology.os.kill"), + patch.object(TopologyManager, "stop_observer") as mock_stop, + patch.object(TopologyManager, "start_observer") as mock_start, + ): + manager.restart_observer() + + mock_stop.assert_called_once() + mock_start.assert_called_once() + + +def test_restart_when_observer_process_is_gone(charm_dir: Path): + """A dead observer is relaunched even though its inputs are unchanged.""" + manager, _ = _make_manager(charm_dir) + + with ( + patch("managers.topology.os.kill", side_effect=OSError), + patch.object(TopologyManager, "stop_observer") as mock_stop, + patch.object(TopologyManager, "start_observer") as mock_start, + ): + manager.restart_observer() + + mock_stop.assert_called_once() + mock_start.assert_called_once() + + +def test_restart_when_no_observer_recorded(charm_dir: Path): + """A unit that never started an observer starts one.""" + manager, _ = _make_manager(charm_dir, pid=None) + + with ( + patch("managers.topology.os.kill") as mock_kill, + patch.object(TopologyManager, "stop_observer") as mock_stop, + patch.object(TopologyManager, "start_observer") as mock_start, + ): + manager.restart_observer() + + mock_kill.assert_not_called() + mock_stop.assert_called_once() + mock_start.assert_called_once() + + +def test_restart_when_pid_file_predates_the_signature_file(charm_dir: Path): + """An observer launched by an older revision has no signature; relaunch it once.""" + manager, _ = _make_manager(charm_dir) + (charm_dir / TOPOLOGY_OBSERVER_SIGNATURE_FILENAME).unlink() + + with ( + patch("managers.topology.os.kill"), + patch.object(TopologyManager, "stop_observer") as mock_stop, + patch.object(TopologyManager, "start_observer") as mock_start, + ): + manager.restart_observer() + + mock_stop.assert_called_once() + mock_start.assert_called_once() + + +def test_stop_observer_clears_both_records(charm_dir: Path): + """Stopping must leave nothing behind that a later check could mistake for a live observer.""" + manager, _ = _make_manager(charm_dir) + + with patch("managers.topology.os.kill") as mock_kill: + manager.stop_observer() + + mock_kill.assert_called_once() + assert not (charm_dir / TOPOLOGY_OBSERVER_PID_FILENAME).exists() + assert not (charm_dir / TOPOLOGY_OBSERVER_SIGNATURE_FILENAME).exists() + + +def test_signature_tracks_every_launch_argument(charm_dir: Path): + """Host set, TLS mode and the Sentinel password each change the digest.""" + baseline, _ = _make_manager(charm_dir) + digest = baseline.observer_signature() + + changed_hosts, _ = _make_manager(charm_dir, endpoints=("10.0.0.1", "10.0.0.9")) + changed_tls, _ = _make_manager(charm_dir, tls=True) + changed_password, _ = _make_manager(charm_dir, password="rotated-password") + + assert changed_hosts.observer_signature() != digest + assert changed_tls.observer_signature() != digest + assert changed_password.observer_signature() != digest + + +def test_signature_tracks_the_ca_when_tls_is_on(charm_dir: Path): + """A CA rotation must relaunch the observer, which holds its own copy of the CA.""" + before, _ = _make_manager(charm_dir, tls=True, client_ca="old-ca-pem") + after, _ = _make_manager(charm_dir, tls=True, client_ca="rotated-ca-pem") + + assert before.observer_signature() != after.observer_signature() + + +def test_signature_ignores_the_ca_when_tls_is_off(charm_dir: Path): + """Without TLS the observer is never handed a CA, so it must not be read.""" + manager, _ = _make_manager(charm_dir, tls=False) + + manager.observer_signature() + + manager.workload.read_file.assert_not_called() + + +def test_signature_survives_an_unreadable_ca(charm_dir: Path): + """An unreadable CA must not crash the leader's peer hook.""" + manager, _ = _make_manager(charm_dir, tls=True) + manager.workload.read_file.side_effect = OSError("gone") + + assert manager.observer_signature() + + +def test_signature_is_order_independent(charm_dir: Path): + """Peer ordering is not a topology change; the digest must not move.""" + one, _ = _make_manager(charm_dir, endpoints=("10.0.0.1", "10.0.0.2")) + other, _ = _make_manager(charm_dir, endpoints=("10.0.0.2", "10.0.0.1")) + + assert one.observer_signature() == other.observer_signature() + + +def test_signature_does_not_leak_the_password(charm_dir: Path): + """The digest is written to a file on the unit, so it must not carry the secret.""" + manager, _ = _make_manager(charm_dir, password="super-secret-password") + + assert "super-secret-password" not in manager.observer_signature()