diff --git a/daemon/registry.py b/daemon/registry.py index 56d3c06..1141d9b 100644 --- a/daemon/registry.py +++ b/daemon/registry.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 import importlib.util +import inspect import json import os from pathlib import Path @@ -55,6 +56,18 @@ class Plugin: self.manifest.get("required_config", []) ) + @property + def optional_config(self): + return list( + self.manifest.get("optional_config", []) + ) + + @property + def config_keys(self): + return list(dict.fromkeys( + self.required_config + self.optional_config + )) + def validate_config(self, config): missing = [ key @@ -68,7 +81,7 @@ class Plugin: + ", ".join(missing) ) - def execute(self, action, config): + def execute(self, action, config, runtime_state): if action not in self.actions: raise RegistryError( f"plugin {self.name} does not support action {action}" @@ -82,6 +95,14 @@ class Plugin: "but does not implement it" ) + # Backward-compatible generic state injection. Existing one-argument + # plugins continue to receive only config; state-aware plugins can + # declare a second positional parameter named runtime_state. + parameters = list(inspect.signature(handler).parameters.values()) + + if len(parameters) >= 2: + return handler(config, runtime_state) + return handler(config) @@ -91,6 +112,11 @@ class Resource: self.plugin = plugin self.config = config + # Generic daemon-lifetime state storage. The daemon does not interpret + # plugin keys or values; applications may use this for transition + # history, grace windows, caches, or other resource-local state. + self.runtime_state = {} + self.plugin.validate_config(config) def describe(self): @@ -136,7 +162,7 @@ class Registry: config = { key: os.environ.get(key) - for key in plugin.required_config + for key in plugin.config_keys } self.resources[resource_id] = Resource( @@ -163,6 +189,7 @@ class Registry: result = resource.plugin.execute( action, resource.config, + resource.runtime_state, ) return { diff --git a/plugins/postgresql/manifest.json b/plugins/postgresql/manifest.json index e30969a..721db19 100644 --- a/plugins/postgresql/manifest.json +++ b/plugins/postgresql/manifest.json @@ -1,15 +1,17 @@ { "name": "postgresql", - "version": 1, + "version": 3, "module": "plugin.py", - "actions": [ - "monitor" - ], + "actions": ["monitor"], "required_config": [ "PGHOST", "PGPORT", "PGUSER", "PGDATABASE", - "PGPASSWORD_FILE" + "PGPASSWORD_FILE", + "PG_REPLICATION_PEERS" + ], + "optional_config": [ + "PG_RECONNECT_GRACE_SECONDS" ] } diff --git a/plugins/postgresql/plugin.py b/plugins/postgresql/plugin.py index 417fe27..10dba31 100644 --- a/plugins/postgresql/plugin.py +++ b/plugins/postgresql/plugin.py @@ -1,28 +1,288 @@ #!/usr/bin/env python3 +import time + import psycopg2 import psycopg2.extras +OCF_SUCCESS = 0 +OCF_ERR_GENERIC = 1 +OCF_NOT_RUNNING = 7 +OCF_RUNNING_PROMOTED = 8 + def read_secret(path): with open(path, "r", encoding="utf-8") as f: return f.read().strip() -def monitor(config): - password = read_secret( - config["PGPASSWORD_FILE"] - ) +def parse_replication_peers(value): + peers = [] + for item in (value or "").split(","): + item = item.strip() + if not item: + continue + name, sep, endpoint = item.partition("=") + if not sep: + raise ValueError(f"invalid replication peer entry: {item!r}") + host, sep, port = endpoint.rpartition(":") + if not sep or not host or not port: + raise ValueError(f"invalid replication peer endpoint: {endpoint!r}") + peers.append({ + "name": name.strip(), + "host": host.strip(), + "port": int(port), + }) + return peers - conn = psycopg2.connect( - host=config["PGHOST"], - port=int(config["PGPORT"]), - user=config["PGUSER"], - dbname=config["PGDATABASE"], - password=password, - connect_timeout=5, - application_name="managed-containerd", - ) + +def probe_replication_peers(config, password): + results = [] + writable_primaries = [] + + for peer in parse_replication_peers(config["PG_REPLICATION_PEERS"]): + result = dict(peer) + try: + conn = psycopg2.connect( + host=peer["host"], + port=peer["port"], + user=config["PGUSER"], + dbname=config["PGDATABASE"], + password=password, + connect_timeout=2, + application_name="managed-containerd-peer-probe", + ) + try: + conn.autocommit = True + with conn.cursor() as cur: + cur.execute("SELECT pg_is_in_recovery()") + in_recovery = bool(cur.fetchone()[0]) + result.update({ + "reachable": True, + "in_recovery": in_recovery, + "writable_primary": not in_recovery, + }) + if not in_recovery: + writable_primaries.append(peer["name"]) + finally: + conn.close() + except psycopg2.Error as exc: + result.update({ + "reachable": False, + "in_recovery": None, + "writable_primary": False, + "error": str(exc).splitlines()[0], + }) + results.append(result) + + return { + "peers": results, + "writable_primaries": writable_primaries, + "writable_primary_count": len(writable_primaries), + } + + +def reconnect_grace_seconds(config): + raw = config.get("PG_RECONNECT_GRACE_SECONDS") or "15" + try: + value = float(raw) + except (TypeError, ValueError): + raise ValueError( + f"invalid PG_RECONNECT_GRACE_SECONDS: {raw!r}" + ) + + if value < 0: + raise ValueError( + "PG_RECONNECT_GRACE_SECONDS must be >= 0" + ) + + return value + + +def _postgres_runtime_state(runtime_state): + return runtime_state.setdefault("postgresql", {}) + + +def _set_transition_state(runtime_state, state): + pg_state = _postgres_runtime_state(runtime_state) + previous = pg_state.get("state") + + if previous != state: + pg_state["state"] = state + pg_state["state_changed_at"] = time.monotonic() + + return pg_state + + +def _clear_reconnect_grace(runtime_state): + pg_state = _postgres_runtime_state(runtime_state) + pg_state.pop("reconnect_started_at", None) + + +def normalize_ha_state( + *, + config, + runtime_state, + in_recovery, + streaming, + receiver, + recovery, + peer_state=None, +): + """Translate only validated PostgreSQL states into OCF semantics.""" + if not in_recovery: + _clear_reconnect_grace(runtime_state) + _set_transition_state(runtime_state, "primary") + return { + "health": "healthy", + "ha_role": "promoted", + "state": "primary", + "ocf_rc": OCF_RUNNING_PROMOTED, + "promotion_eligible": True, + } + + replay_paused = bool((recovery or {}).get("replay_paused")) + if streaming and not replay_paused: + _clear_reconnect_grace(runtime_state) + _set_transition_state(runtime_state, "streaming") + return { + "health": "healthy", + "ha_role": "unpromoted", + "state": "streaming", + "ocf_rc": OCF_SUCCESS, + "promotion_eligible": True, + } + + receiver_status = (receiver or {}).get("status") + state = receiver_status or "not_streaming" + writable_primary_count = None + + if peer_state is not None: + writable_primary_count = peer_state.get( + "writable_primary_count" + ) + + # Validated against the authoritative RA on 2026-09-17: if the local + # server remains a reachable standby, its WAL receiver is not streaming, + # and no writable primary is reachable, the RA returns OCF_SUCCESS and + # keeps the node as a reduced-priority election candidate. + if state == "not_streaming" and writable_primary_count == 0: + _clear_reconnect_grace(runtime_state) + _set_transition_state(runtime_state, "disconnected_candidate") + return { + "health": "degraded", + "ha_role": "unpromoted", + "state": "disconnected_candidate", + "ocf_rc": OCF_SUCCESS, + "promotion_eligible": True, + } + + # Reconnect grace is PostgreSQL-specific policy layered on generic daemon + # runtime state. It is entered only after this daemon has actually observed + # disconnected_candidate and then sees exactly one writable primary return + # while the local WAL receiver has not yet resumed streaming. + if state == "not_streaming" and writable_primary_count == 1: + pg_state = _postgres_runtime_state(runtime_state) + previous_state = pg_state.get("state") + started_at = pg_state.get("reconnect_started_at") + now = time.monotonic() + grace = reconnect_grace_seconds(config) + + if previous_state == "disconnected_candidate": + started_at = now + pg_state["reconnect_started_at"] = started_at + + if started_at is not None: + elapsed = max(0.0, now - float(started_at)) + + if elapsed < grace: + _set_transition_state(runtime_state, "reconnecting") + return { + "health": "degraded", + "ha_role": "unpromoted", + "state": "reconnecting", + "ocf_rc": OCF_SUCCESS, + "promotion_eligible": False, + "grace": { + "seconds": grace, + "elapsed_seconds": round(elapsed, 3), + "remaining_seconds": round( + max(0.0, grace - elapsed), 3 + ), + }, + } + + _clear_reconnect_grace(runtime_state) + + # Already validated for the steady-state failure case: a reachable + # standby whose WAL receiver is not streaming while exactly one remote + # writable primary is available returns OCF_ERR_GENERIC. + _set_transition_state(runtime_state, "not_streaming") + return { + "health": "degraded", + "ha_role": "unpromoted", + "state": "not_streaming", + "ocf_rc": OCF_ERR_GENERIC, + "promotion_eligible": False, + } + + # Multiple writable primaries and other degraded states remain deliberately + # unclassified until independently validated. + _clear_reconnect_grace(runtime_state) + _set_transition_state(runtime_state, state) + return { + "health": "degraded", + "ha_role": "unpromoted", + "state": state, + "ocf_rc": None, + "promotion_eligible": False, + } + + +def monitor(config, runtime_state): + password = read_secret(config["PGPASSWORD_FILE"]) + + try: + conn = psycopg2.connect( + host=config["PGHOST"], + port=int(config["PGPORT"]), + user=config["PGUSER"], + dbname=config["PGDATABASE"], + password=password, + connect_timeout=5, + application_name="managed-containerd", + ) + except psycopg2.OperationalError as exc: + message = str(exc) + + # Validated against the authoritative RA on 2026-09-17: + # an explicitly refused local PostgreSQL connection corresponds to + # OCF_NOT_RUNNING. Other OperationalError classes remain unclassified. + if "connection refused" in message.lower(): + _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": "connection_refused", + "message": message, + }, + } + + raise try: conn.autocommit = True @@ -30,38 +290,27 @@ def monitor(config): with conn.cursor( cursor_factory=psycopg2.extras.RealDictCursor ) as cur: - cur.execute(""" SELECT pg_is_in_recovery() AS in_recovery """) - - in_recovery = bool( - cur.fetchone()["in_recovery"] - ) + in_recovery = bool(cur.fetchone()["in_recovery"]) if not in_recovery: cur.execute(""" SELECT - pg_current_wal_lsn()::text - AS current_wal_lsn + pg_current_wal_lsn()::text AS current_wal_lsn """) - primary = cur.fetchone() - else: primary = None if in_recovery: cur.execute(""" SELECT - pg_last_wal_receive_lsn()::text - AS receive_lsn, - pg_last_wal_replay_lsn()::text - AS replay_lsn, - pg_is_wal_replay_paused() - AS replay_paused + pg_last_wal_receive_lsn()::text AS receive_lsn, + pg_last_wal_replay_lsn()::text AS replay_lsn, + pg_is_wal_replay_paused() AS replay_paused """) - recovery = cur.fetchone() cur.execute(""" @@ -70,68 +319,62 @@ def monitor(config): sender_host, sender_port, slot_name, - latest_end_lsn::text - AS latest_end_lsn + latest_end_lsn::text AS latest_end_lsn FROM pg_stat_wal_receiver LIMIT 1 """) - receiver = cur.fetchone() - else: recovery = { "receive_lsn": None, "replay_lsn": None, "replay_paused": False, } - receiver = None cur.execute(""" SELECT - current_setting('server_version') - AS server_version, - current_setting('cluster_name', true) - AS cluster_name, + current_setting('server_version') AS server_version, + current_setting('cluster_name', true) AS cluster_name, current_user AS current_user, - current_database() - AS current_database, - inet_server_addr()::text - AS server_address, - inet_server_port() - AS server_port + current_database() AS current_database, + inet_server_addr()::text AS server_address, + inet_server_port() AS server_port """) - server = cur.fetchone() if in_recovery: role = "standby" - streaming = bool( - receiver - and receiver.get("status") - == "streaming" + receiver and receiver.get("status") == "streaming" ) - else: role = "primary" streaming = False + peer_state = None + if in_recovery and not streaming: + 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, + ) + return { + "ha": ha, + "peer_state": peer_state, "role": role, "in_recovery": in_recovery, "streaming": streaming, - "wal_receiver": ( - dict(receiver) - if receiver - else None - ), + "wal_receiver": dict(receiver) if receiver else None, "recovery": dict(recovery), - "primary": ( - dict(primary) - if primary - else None - ), + "primary": dict(primary) if primary else None, "server": dict(server), } diff --git a/postgres.env b/postgres.env index b835300..69e033d 100644 --- a/postgres.env +++ b/postgres.env @@ -7,8 +7,10 @@ PLUGIN=postgresql MANAGED_CONTAINERD_SOCKET=/run/managed-containerd/managed-containerd.sock MANAGED_CONTAINERD_PLUGIN_ROOT=/opt/managed-containerd/plugins -PGHOST=192.168.1.4 +PGHOST=postgres PGPORT=5432 PGUSER=david PGDATABASE=david PGPASSWORD_FILE=/run/secrets/postgres_password +PG_REPLICATION_PEERS=david-prodesk=192.168.1.4:5432,david-macbookpro=192.168.1.5:5432,david-eve=192.168.1.3:5432 +PG_RECONNECT_GRACE_SECONDS=15 diff --git a/tests/test-runtime-state.py b/tests/test-runtime-state.py new file mode 100755 index 0000000..119e824 --- /dev/null +++ b/tests/test-runtime-state.py @@ -0,0 +1,234 @@ +#!/usr/bin/env python3 + +import ast +import json +import os +import sys +import tempfile +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, "daemon") + +from registry import Plugin, Registry, RegistryError, Resource + + +def require(condition, message): + if not condition: + raise SystemExit("FAIL: " + message) + + +def write_plugin( + root, + name, + manifest, + source, +): + plugin_root = Path(root) / name + plugin_root.mkdir() + + (plugin_root / "manifest.json").write_text( + json.dumps(manifest) + ) + (plugin_root / "plugin.py").write_text(source) + + +print("===== GENERIC PLUGIN RUNTIME STATE =====") + +with tempfile.TemporaryDirectory() as temporary_root: + stateful_manifest = { + "name": "stateful", + "version": 1, + "module": "plugin.py", + "actions": ["monitor"], + "required_config": ["REQUIRED_VALUE"], + "optional_config": ["OPTIONAL_VALUE"], + } + + write_plugin( + temporary_root, + "stateful", + stateful_manifest, + """ +def monitor(config, runtime_state): + counter = runtime_state.get("counter", 0) + 1 + runtime_state["counter"] = counter + + return { + "counter": counter, + "required": config["REQUIRED_VALUE"], + "optional": config["OPTIONAL_VALUE"], + } +""", + ) + + environment = { + "MANAGED_CONTAINERD_PLUGIN_ROOT": ( + temporary_root + ), + "RESOURCE_ID": "stateful-resource", + "PLUGIN": "stateful", + "REQUIRED_VALUE": "required", + "OPTIONAL_VALUE": "optional", + } + + with patch.dict( + os.environ, + environment, + clear=False, + ): + registry = Registry() + + first = registry.execute( + "stateful-resource", + "monitor", + ) + second = registry.execute( + "stateful-resource", + "monitor", + ) + + require( + first["result"]["counter"] == 1, + "first stateful invocation has wrong counter", + ) + require( + second["result"]["counter"] == 2, + "runtime state did not persist", + ) + require( + second["result"]["required"] == "required", + "required configuration was not injected", + ) + require( + second["result"]["optional"] == "optional", + "optional configuration was not injected", + ) + + resource = registry.resources[ + "stateful-resource" + ] + + require( + resource.runtime_state == {"counter": 2}, + "resource runtime state has unexpected contents", + ) + + print("stateful_handler persisted") + print("optional_config injected") + + legacy_manifest = { + "name": "legacy", + "version": 1, + "module": "plugin.py", + "actions": ["monitor"], + "required_config": ["REQUIRED_VALUE"], + } + + write_plugin( + temporary_root, + "legacy", + legacy_manifest, + """ +def monitor(config): + return { + "legacy": True, + "required": config["REQUIRED_VALUE"], + } +""", + ) + + legacy_plugin = Plugin( + temporary_root, + "legacy", + ) + + legacy_result = legacy_plugin.execute( + "monitor", + {"REQUIRED_VALUE": "required"}, + {}, + ) + + require( + legacy_result == { + "legacy": True, + "required": "required", + }, + "one-argument plugin compatibility failed", + ) + + print("legacy_handler compatible") + + try: + Resource( + "invalid-resource", + legacy_plugin, + {"REQUIRED_VALUE": None}, + ) + except RegistryError: + print("required_config enforced") + else: + raise SystemExit( + "FAIL: missing required configuration accepted" + ) + +production_manifest = json.loads( + Path( + "plugins/postgresql/manifest.json" + ).read_text() +) + +require( + production_manifest.get("version") == 3, + "PostgreSQL manifest version is not 3", +) +require( + "PG_REPLICATION_PEERS" + in production_manifest.get( + "required_config", + [], + ), + "replication peers are not required", +) +require( + "PG_RECONNECT_GRACE_SECONDS" + in production_manifest.get( + "optional_config", + [], + ), + "reconnect grace is not optional", +) + +plugin_tree = ast.parse( + Path( + "plugins/postgresql/plugin.py" + ).read_text() +) + +monitor_functions = [ + node + for node in plugin_tree.body + if isinstance(node, ast.FunctionDef) + and node.name == "monitor" +] + +require( + len(monitor_functions) == 1, + "PostgreSQL monitor entry point is ambiguous", +) + +monitor_arguments = [ + argument.arg + for argument + in monitor_functions[0].args.args +] + +require( + monitor_arguments[:2] + == ["config", "runtime_state"], + "PostgreSQL monitor does not accept runtime state", +) + +print("postgresql_contract validated") +print() +print("RUNTIME_STATE_TEST_PASS=1")