From 01695d1ac1f34b30eb164879d0dfe8cdbe30e528 Mon Sep 17 00:00:00 2001 From: Lyncoln Date: Fri, 18 Sep 2026 14:09:55 -0400 Subject: [PATCH] Add selectable PostgreSQL monitor backends --- daemon/registry.py | 14 + plugins/postgresql/manifest.json | 8 +- plugins/postgresql/plugin.py | 492 ++++++++++++++++++++++++++++++- tests/test-monitor-backends.py | 317 ++++++++++++++++++++ tests/test-runtime-state.py | 4 +- 5 files changed, 830 insertions(+), 5 deletions(-) create mode 100755 tests/test-monitor-backends.py diff --git a/daemon/registry.py b/daemon/registry.py index b1c1079..a7df29b 100644 --- a/daemon/registry.py +++ b/daemon/registry.py @@ -84,6 +84,20 @@ class Plugin: + ", ".join(missing) ) + plugin_validator = getattr( + self.module, + "validate_config", + None, + ) + + if plugin_validator is not None: + try: + plugin_validator(config) + except (TypeError, ValueError) as exc: + raise RegistryError( + f"invalid plugin configuration: {exc}" + ) from exc + def execute( self, action, diff --git a/plugins/postgresql/manifest.json b/plugins/postgresql/manifest.json index 3a0983b..67b531b 100644 --- a/plugins/postgresql/manifest.json +++ b/plugins/postgresql/manifest.json @@ -1,6 +1,6 @@ { "name": "postgresql", - "version": 4, + "version": 5, "module": "plugin.py", "actions": [ "monitor", @@ -20,6 +20,10 @@ ], "optional_config": [ "PG_RECONNECT_GRACE_SECONDS", - "PACEMAKER_NODE_NAME" + "PACEMAKER_NODE_NAME", + "MONITOR_BACKEND", + "CONTAINER_RUNTIME", + "CONTAINER_NAME", + "PG_OS_USER" ] } diff --git a/plugins/postgresql/plugin.py b/plugins/postgresql/plugin.py index 6e77ebb..25643da 100644 --- a/plugins/postgresql/plugin.py +++ b/plugins/postgresql/plugin.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 import time +import subprocess import psycopg2 import psycopg2.extras @@ -11,6 +12,111 @@ OCF_NOT_RUNNING = 7 OCF_RUNNING_PROMOTED = 8 +SUPPORTED_MONITOR_BACKENDS = {"api", "container"} +SUPPORTED_CONTAINER_RUNTIMES = {"podman", "docker"} + + +def monitor_backend(config): + backend = ( + config.get("MONITOR_BACKEND") + or "api" + ).strip().lower() + + if backend not in SUPPORTED_MONITOR_BACKENDS: + raise ValueError( + f"unsupported MONITOR_BACKEND: {backend!r}" + ) + + return backend + + +def container_runtime(config): + runtime = ( + config.get("CONTAINER_RUNTIME") + or "" + ).strip().lower() + + if not runtime: + raise ValueError( + "CONTAINER_RUNTIME is required when " + "MONITOR_BACKEND=container" + ) + + if runtime not in SUPPORTED_CONTAINER_RUNTIMES: + raise ValueError( + f"unsupported CONTAINER_RUNTIME: {runtime!r}" + ) + + return runtime + + +def container_monitor_config(config): + runtime = container_runtime(config) + + container = ( + config.get("CONTAINER_NAME") + or "" + ).strip() + + if not container: + raise ValueError( + "CONTAINER_NAME is required when " + "MONITOR_BACKEND=container" + ) + + os_user = ( + config.get("PG_OS_USER") + or "postgres" + ).strip() + + if not os_user: + raise ValueError( + "PG_OS_USER must not be empty" + ) + + return runtime, container, os_user + + +def container_exec( + config, + *command, + timeout=5, + check=True, +): + runtime, container, os_user = ( + container_monitor_config(config) + ) + + argv = [ + runtime, + "exec", + "--user", + os_user, + container, + *command, + ] + + return subprocess.run( + argv, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + timeout=timeout, + check=check, + ) + + +def validate_config(config): + backend = monitor_backend(config) + + if backend == "container": + # Option B: selection/delegation requires only a supported runtime. + # Direct container execution has stricter requirements and validates + # those separately through container_monitor_config(). + container_runtime(config) + + def read_secret(path): with open(path, "r", encoding="utf-8") as f: return f.read().strip() @@ -239,7 +345,7 @@ def normalize_ha_state( } -def monitor(config, runtime_state): +def monitor_api(config, runtime_state): password = read_secret(config["PGPASSWORD_FILE"]) try: @@ -382,6 +488,390 @@ def monitor(config, runtime_state): conn.close() +def monitor_container(config, runtime_state): + """ + Monitor local PostgreSQL through the selected container runtime. + + Container-backed monitoring intentionally favors compatibility with the + proven RA semantics over minimizing runtime exec calls. The API backend is + the low-overhead monitoring path. + """ + + runtime, container, os_user = container_monitor_config(config) + + def not_running(kind, message): + _clear_reconnect_grace(runtime_state) + _set_transition_state( + runtime_state, + "not_running", + ) + + return { + "ha": { + "health": "down", + "ha_role": "unknown", + "state": "not_running", + "ocf_rc": OCF_NOT_RUNNING, + "promotion_eligible": False, + }, + "role": "unknown", + "in_recovery": None, + "streaming": False, + "wal_receiver": None, + "recovery": None, + "primary": None, + "server": None, + "error": { + "kind": kind, + "message": message, + }, + "monitor_backend": "container", + "container_runtime": runtime, + } + + def monitor_error(kind, message): + _clear_reconnect_grace(runtime_state) + _set_transition_state( + runtime_state, + "monitor_error", + ) + + return { + "ha": { + "health": "degraded", + "ha_role": "unknown", + "state": "monitor_error", + "ocf_rc": OCF_ERR_GENERIC, + "promotion_eligible": False, + }, + "role": "unknown", + "in_recovery": None, + "streaming": False, + "wal_receiver": None, + "recovery": None, + "primary": None, + "server": None, + "error": { + "kind": kind, + "message": message, + }, + "monitor_backend": "container", + "container_runtime": runtime, + } + + # Establish whether PostgreSQL itself is running. + pg_status_command = ( + 'BINDIR="$(pg_config --bindir)" || exit 2; ' + '"$BINDIR/pg_ctl" status -D "$PGDATA" >/dev/null 2>&1' + ) + + try: + container_exec( + config, + "sh", + "-c", + pg_status_command, + timeout=3, + ) + + except subprocess.CalledProcessError as exc: + return not_running( + "postgresql_not_running", + f"pg_ctl status failed with rc={exc.returncode}", + ) + + except ( + subprocess.TimeoutExpired, + FileNotFoundError, + OSError, + ) as exc: + return monitor_error( + "container_runtime_failure", + str(exc), + ) + + # A running postmaster that is not ready is unhealthy, not stopped. + try: + container_exec( + config, + "pg_isready", + "-q", + "-U", + config["PGUSER"], + "-d", + config["PGDATABASE"], + timeout=3, + ) + + except subprocess.CalledProcessError as exc: + return monitor_error( + "postgresql_not_ready", + f"pg_isready failed with rc={exc.returncode}", + ) + + except ( + subprocess.TimeoutExpired, + FileNotFoundError, + OSError, + ) as exc: + return monitor_error( + "container_runtime_failure", + str(exc), + ) + + query = r""" + SELECT json_build_object( + 'in_recovery', + pg_is_in_recovery(), + + 'receive_lsn', + CASE + WHEN pg_is_in_recovery() + THEN pg_last_wal_receive_lsn()::text + ELSE NULL + END, + + 'replay_lsn', + CASE + WHEN pg_is_in_recovery() + THEN pg_last_wal_replay_lsn()::text + ELSE NULL + END, + + 'replay_paused', + CASE + WHEN pg_is_in_recovery() + THEN pg_is_wal_replay_paused() + ELSE false + END, + + 'current_wal_lsn', + CASE + WHEN NOT pg_is_in_recovery() + THEN pg_current_wal_lsn()::text + ELSE NULL + END, + + 'server_version', + current_setting('server_version'), + + 'cluster_name', + current_setting('cluster_name', true), + + 'current_user', + current_user, + + 'current_database', + current_database(), + + 'server_address', + inet_server_addr()::text, + + 'server_port', + inet_server_port(), + + 'receiver', + ( + SELECT row_to_json(receiver_state) + FROM ( + SELECT + status, + sender_host, + sender_port, + slot_name, + latest_end_lsn::text AS latest_end_lsn + FROM pg_stat_wal_receiver + LIMIT 1 + ) AS receiver_state + ) + )::text; + """ + + try: + completed = container_exec( + config, + "psql", + "-X", + "-w", + "-v", + "ON_ERROR_STOP=1", + "-U", + config["PGUSER"], + "-d", + config["PGDATABASE"], + "-Atqc", + query, + timeout=5, + ) + + except subprocess.CalledProcessError as exc: + return monitor_error( + "postgresql_query_failed", + f"monitor query failed with rc={exc.returncode}", + ) + + except ( + subprocess.TimeoutExpired, + FileNotFoundError, + OSError, + ) as exc: + return monitor_error( + "container_runtime_failure", + str(exc), + ) + + try: + import json + + payload = json.loads( + completed.stdout.strip() + ) + + in_recovery = bool( + payload["in_recovery"] + ) + + except ( + KeyError, + TypeError, + ValueError, + json.JSONDecodeError, + ) as exc: + return monitor_error( + "invalid_monitor_payload", + str(exc), + ) + + receiver = payload.get("receiver") + + if in_recovery: + recovery = { + "receive_lsn": payload.get( + "receive_lsn" + ), + "replay_lsn": payload.get( + "replay_lsn" + ), + "replay_paused": bool( + payload.get("replay_paused") + ), + } + + primary = None + + streaming = bool( + receiver + and receiver.get("status") == "streaming" + ) + + role = "standby" + + else: + recovery = { + "receive_lsn": None, + "replay_lsn": None, + "replay_paused": False, + } + + primary = { + "current_wal_lsn": payload.get( + "current_wal_lsn" + ), + } + + receiver = None + streaming = False + role = "primary" + + peer_state = None + + if in_recovery and not streaming: + password = read_secret( + config["PGPASSWORD_FILE"] + ) + + peer_state = probe_replication_peers( + config, + password, + ) + + ha = normalize_ha_state( + config=config, + runtime_state=runtime_state, + in_recovery=in_recovery, + streaming=streaming, + receiver=receiver, + recovery=recovery, + peer_state=peer_state, + ) + + server = { + "server_version": payload.get( + "server_version" + ), + "cluster_name": payload.get( + "cluster_name" + ), + "current_user": payload.get( + "current_user" + ), + "current_database": payload.get( + "current_database" + ), + "server_address": payload.get( + "server_address" + ), + "server_port": payload.get( + "server_port" + ), + } + + return { + "ha": ha, + "peer_state": peer_state, + "role": role, + "in_recovery": in_recovery, + "streaming": streaming, + "wal_receiver": receiver, + "recovery": recovery, + "primary": primary, + "server": server, + "monitor_backend": "container", + "container_runtime": runtime, + } + + + +def monitor(config, runtime_state): + backend = monitor_backend(config) + + if backend == "api": + result = monitor_api( + config, + runtime_state, + ) + + result["monitor_backend"] = "api" + return result + + if backend == "container": + # Option B transitional execution model: + # managed-containerd remains authoritative for backend selection, + # but delegates container execution to the RA's packaged runtime + # and application adapters. + # + # monitor_container() remains implemented for future Option A. + return { + "delegated": True, + "backend": "container", + "container_runtime": container_runtime(config), + } + + raise ValueError( + f"unsupported monitor backend: {backend!r}" + ) + + NOTIFY_FIELDS = ( "type", "operation", diff --git a/tests/test-monitor-backends.py b/tests/test-monitor-backends.py new file mode 100755 index 0000000..80e12ee --- /dev/null +++ b/tests/test-monitor-backends.py @@ -0,0 +1,317 @@ +#!/usr/bin/env python3 + +import importlib.util +import json +import subprocess +import sys +import types +from pathlib import Path + + +# Match the existing daemon unit-test convention: psycopg2 is supplied by the +# production container image, while host-side tests use a minimal fake module. +fake_psycopg2 = types.ModuleType("psycopg2") +fake_psycopg2.Error = Exception +fake_psycopg2.OperationalError = OSError + +fake_extras = types.ModuleType("psycopg2.extras") +fake_extras.RealDictCursor = object + +fake_psycopg2.extras = fake_extras + +sys.modules["psycopg2"] = fake_psycopg2 +sys.modules["psycopg2.extras"] = fake_extras + + +spec = importlib.util.spec_from_file_location( + "postgresql_plugin", + Path("plugins/postgresql/plugin.py"), +) +plugin = importlib.util.module_from_spec(spec) +spec.loader.exec_module(plugin) + + +def base_config(): + return { + "PGHOST": "127.0.0.1", + "PGPORT": "5432", + "PGUSER": "david", + "PGDATABASE": "david", + "PGPASSWORD_FILE": "/run/secrets/postgres_password", + "PG_REPLICATION_PEERS": ( + "node1=192.0.2.1:5432," + "node2=192.0.2.2:5432" + ), + "PACEMAKER_NODE_NAME": "node1", + } + + +print("===== DEFAULT BACKEND =====") + +cfg = base_config() + +assert plugin.monitor_backend(cfg) == "api" +plugin.validate_config(cfg) + +print("default_api accepted") + + +print() +print("===== EXPLICIT API =====") + +cfg = base_config() +cfg["MONITOR_BACKEND"] = "api" + +plugin.validate_config(cfg) +assert plugin.monitor_backend(cfg) == "api" + +print("explicit_api accepted") + + +print() +print("===== API DISPATCH =====") + +cfg = base_config() +cfg["MONITOR_BACKEND"] = "api" + +original_api = plugin.monitor_api +original_container = plugin.monitor_container + +try: + plugin.monitor_api = ( + lambda config, runtime_state: { + "ha": { + "ocf_rc": plugin.OCF_SUCCESS + }, + "role": "standby", + } + ) + + def must_not_run(*args, **kwargs): + raise AssertionError( + "container monitor executed for API backend" + ) + + plugin.monitor_container = must_not_run + + result = plugin.monitor(cfg, {}) + + assert result["ha"]["ocf_rc"] == plugin.OCF_SUCCESS + assert result["monitor_backend"] == "api" + +finally: + plugin.monitor_api = original_api + plugin.monitor_container = original_container + +print("api_dispatch accepted") + + +print() +print("===== CONTAINER VALIDATION =====") + +cfg = base_config() +cfg["MONITOR_BACKEND"] = "container" +cfg["CONTAINER_RUNTIME"] = "podman" +cfg["CONTAINER_NAME"] = "postgres" +cfg["PG_OS_USER"] = "postgres" + +plugin.validate_config(cfg) + +assert plugin.monitor_backend(cfg) == "container" +assert plugin.container_runtime(cfg) == "podman" + +print("container_podman accepted") + + +print() +print("===== DOCKER VALIDATION =====") + +cfg["CONTAINER_RUNTIME"] = "docker" + +plugin.validate_config(cfg) +assert plugin.container_runtime(cfg) == "docker" + +print("container_docker accepted") + + +print() +print("===== INVALID BACKEND =====") + +bad = base_config() +bad["MONITOR_BACKEND"] = "invalid" + +try: + plugin.validate_config(bad) +except ValueError: + print("invalid_backend rejected") +else: + raise AssertionError( + "invalid backend was accepted" + ) + + +print() +print("===== MISSING RUNTIME =====") + +bad = base_config() +bad["MONITOR_BACKEND"] = "container" +bad["CONTAINER_NAME"] = "postgres" + +try: + plugin.validate_config(bad) +except ValueError: + print("missing_runtime rejected") +else: + raise AssertionError( + "container backend without runtime was accepted" + ) + + +print() +print("===== INVALID RUNTIME =====") + +bad["CONTAINER_RUNTIME"] = "lxc" + +try: + plugin.validate_config(bad) +except ValueError: + print("invalid_runtime rejected") +else: + raise AssertionError( + "invalid runtime was accepted" + ) + + +print() +print("===== DELEGATION WITHOUT CONTAINER IDENTITY =====") + +cfg = base_config() +cfg["MONITOR_BACKEND"] = "container" +cfg["CONTAINER_RUNTIME"] = "podman" + +plugin.validate_config(cfg) + +print("delegation_config accepted") + + +print() +print("===== DIRECT EXECUTOR STILL REQUIRES CONTAINER =====") + +try: + plugin.container_monitor_config(cfg) +except ValueError: + print("direct_missing_container rejected") +else: + raise AssertionError( + "direct executor accepted missing container name" + ) + + +print() +print("===== CONTAINER DISPATCH DELEGATION =====") + +original_container = plugin.monitor_container + +try: + def must_not_execute(*args, **kwargs): + raise AssertionError( + "direct container monitor executed during delegation" + ) + + plugin.monitor_container = must_not_execute + + result = plugin.monitor(cfg, {}) + + assert result == { + "delegated": True, + "backend": "container", + "container_runtime": "podman", + }, result + +finally: + plugin.monitor_container = original_container + +print("container_delegation accepted") + + +print() +print("MONITOR_BACKEND_TEST_PASS=1") + + +print() +print("===== CONTAINER RUNTIME EXECUTION =====") + +original_run = plugin.subprocess.run +calls = [] + +try: + def fake_run(argv, **kwargs): + calls.append((argv, kwargs)) + + class Result: + stdout = "" + stderr = "" + returncode = 0 + + return Result() + + plugin.subprocess.run = fake_run + + podman_config = base_config() + podman_config.update({ + "MONITOR_BACKEND": "container", + "CONTAINER_RUNTIME": "podman", + "CONTAINER_NAME": "postgres", + "PG_OS_USER": "postgres", + }) + + plugin.container_exec( + podman_config, + "pg_isready", + "-q", + ) + + assert calls[-1][0] == [ + "podman", + "exec", + "--user", + "postgres", + "postgres", + "pg_isready", + "-q", + ], calls[-1][0] + + print("podman_exec_selection accepted") + + docker_config = base_config() + docker_config.update({ + "MONITOR_BACKEND": "container", + "CONTAINER_RUNTIME": "docker", + "CONTAINER_NAME": "postgres", + "PG_OS_USER": "postgres", + }) + + plugin.container_exec( + docker_config, + "pg_isready", + "-q", + ) + + assert calls[-1][0] == [ + "docker", + "exec", + "--user", + "postgres", + "postgres", + "pg_isready", + "-q", + ], calls[-1][0] + + print("docker_exec_selection accepted") + +finally: + plugin.subprocess.run = original_run + + +print() +print("CONTAINER_RUNTIME_SELECTION_TEST_PASS=1") diff --git a/tests/test-runtime-state.py b/tests/test-runtime-state.py index 92f4e07..f8af9d9 100755 --- a/tests/test-runtime-state.py +++ b/tests/test-runtime-state.py @@ -179,8 +179,8 @@ production_manifest = json.loads( ) require( - production_manifest.get("version") == 4, - "PostgreSQL manifest version is not 4", + production_manifest.get("version") == 5, + "PostgreSQL manifest version is not 5", ) require( "PG_REPLICATION_PEERS"