From 5fcd0bf93182c3f4e7fda3a2b4645a0ea2882b3f Mon Sep 17 00:00:00 2001 From: Lyncoln Date: Thu, 17 Sep 2026 18:15:39 -0400 Subject: [PATCH] Add read-only PostgreSQL demotion database probe --- plugins/postgresql/manifest.json | 3 +- plugins/postgresql/plugin.py | 155 +++++++++++++++++++++++++++++++ tests/test-demote-database.py | 139 +++++++++++++++++++++++++++ 3 files changed, 296 insertions(+), 1 deletion(-) create mode 100755 tests/test-demote-database.py diff --git a/plugins/postgresql/manifest.json b/plugins/postgresql/manifest.json index a19b696..e4f753e 100644 --- a/plugins/postgresql/manifest.json +++ b/plugins/postgresql/manifest.json @@ -6,7 +6,8 @@ "monitor", "notify", "notify.target", - "demote.prepare" + "demote.prepare", + "demote.database" ], "required_config": [ "PGHOST", diff --git a/plugins/postgresql/plugin.py b/plugins/postgresql/plugin.py index 8fc44ad..6ac4523 100644 --- a/plugins/postgresql/plugin.py +++ b/plugins/postgresql/plugin.py @@ -676,3 +676,158 @@ def demote_prepare(config, runtime_state, request_parameters): "database_checks": False, }, } + + +def _lsn_to_int(value): + if not value: + return None + + upper, lower = value.split("/", 1) + return (int(upper, 16) << 32) + int(lower, 16) + + +def demote_database(config, runtime_state, request_parameters): + """ + Read-only PostgreSQL successor validation for demotion. + """ + del request_parameters + + preparation = demote_prepare( + config, + runtime_state, + {}, + ) + + if preparation.get("reason") not in ( + "database_checks_pending", + ): + return preparation + + local = monitor(config, runtime_state) + + if local["ha"]["ocf_rc"] != OCF_RUNNING_PROMOTED: + return { + "ocf_rc": 1, + "ready": False, + "reason": "local_not_primary", + "local": local["ha"], + } + + successor_node = preparation["successor_node"] + peers = parse_replication_peers( + config.get("PG_REPLICATION_PEERS", "") + ) + + successor = next( + ( + peer for peer in peers + if peer["name"].strip().lower() == successor_node + ), + None, + ) + + if successor is None: + return { + "ocf_rc": 1, + "ready": False, + "reason": "successor_not_configured", + "successor_node": successor_node, + } + + try: + password = read_secret(config["PGPASSWORD_FILE"]) + + conn = psycopg2.connect( + host=successor["host"], + port=successor["port"], + user=config["PGUSER"], + dbname=config["PGDATABASE"], + password=password, + connect_timeout=3, + application_name="managed-containerd-demote-check", + ) + + try: + conn.autocommit = True + + with conn.cursor( + cursor_factory=psycopg2.extras.RealDictCursor + ) as cur: + cur.execute(""" + SELECT + pg_is_in_recovery() AS in_recovery, + 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, + ( + SELECT status + FROM pg_stat_wal_receiver + LIMIT 1 + ) AS receiver_status, + ( + SELECT slot_name + FROM pg_stat_wal_receiver + LIMIT 1 + ) AS slot_name + """) + successor_state = dict(cur.fetchone()) + finally: + conn.close() + + except psycopg2.Error as exc: + return { + "ocf_rc": 1, + "ready": False, + "reason": "successor_unreachable", + "successor_node": successor_node, + "successor_host": successor["host"], + "successor_port": successor["port"], + "error": str(exc).splitlines()[0], + } + + primary_lsn = local["primary"]["current_wal_lsn"] + replay_lsn = successor_state["replay_lsn"] + + primary_value = _lsn_to_int(primary_lsn) + replay_value = _lsn_to_int(replay_lsn) + + wal_lag_bytes = None + if primary_value is not None and replay_value is not None: + wal_lag_bytes = max(0, primary_value - replay_value) + + checks = { + "local_primary": True, + "successor_reachable": True, + "successor_in_recovery": ( + successor_state["in_recovery"] is True + ), + "successor_streaming": ( + successor_state["receiver_status"] == "streaming" + ), + "successor_replay_not_paused": ( + successor_state["replay_paused"] is False + ), + "successor_slot_present": bool( + successor_state["slot_name"] + ), + } + + ready = all(checks.values()) + + return { + "ocf_rc": 0 if ready else 1, + "ready": ready, + "mutation": False, + "final_handoff_required": True, + "node": preparation["node"], + "target_node": preparation["target_node"], + "successor_node": successor_node, + "successor_host": successor["host"], + "successor_port": successor["port"], + "primary_wal_lsn": primary_lsn, + "successor_receive_lsn": successor_state["receive_lsn"], + "successor_replay_lsn": replay_lsn, + "wal_lag_bytes": wal_lag_bytes, + "successor_slot_name": successor_state["slot_name"], + "checks": checks, + } diff --git a/tests/test-demote-database.py b/tests/test-demote-database.py new file mode 100755 index 0000000..96cb717 --- /dev/null +++ b/tests/test-demote-database.py @@ -0,0 +1,139 @@ +#!/usr/bin/env python3 + +import importlib.util +import sys +import time +import types +from pathlib import Path + +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) + + +class Cursor: + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def execute(self, query): + pass + + def fetchone(self): + return { + "in_recovery": True, + "replay_paused": False, + "receive_lsn": "14/1C07A910", + "replay_lsn": "14/1C07A910", + "receiver_status": "streaming", + "slot_name": "ha_pg_david_eve", + } + + +class Connection: + autocommit = False + + def cursor(self, cursor_factory=None): + return Cursor() + + def close(self): + pass + + +plugin.monitor = lambda config, runtime_state: { + "ha": {"ocf_rc": plugin.OCF_RUNNING_PROMOTED}, + "primary": {"current_wal_lsn": "14/1C07A920"}, +} + +plugin.read_secret = lambda path: "test-password" +plugin.psycopg2.connect = lambda **kwargs: Connection() + +runtime_state = { + "postgresql": { + "notification": { + "type": "pre", + "operation": "demote", + "demote_uname": "david-prodesk", + "promote_uname": "david-eve", + "received_at": time.time(), + }, + }, +} + +config = { + "PACEMAKER_NODE_NAME": "david-prodesk", + "PG_REPLICATION_PEERS": ( + "david-prodesk=192.168.1.4:5432," + "david-eve=192.168.1.3:5432" + ), + "PGPASSWORD_FILE": "/run/secrets/postgres_password", + "PGUSER": "david", + "PGDATABASE": "david", +} + +result = plugin.demote_database(config, runtime_state, {}) + +assert result["ocf_rc"] == 0 +assert result["ready"] is True +assert result["mutation"] is False +assert result["final_handoff_required"] is True +assert result["successor_node"] == "david-eve" +assert result["successor_slot_name"] == "ha_pg_david_eve" +assert result["checks"]["successor_streaming"] is True +assert result["checks"]["successor_replay_not_paused"] is True +assert result["wal_lag_bytes"] == 16 + +print("valid_database_probe accepted") +print("wal_lag_reported accepted") +print() +print("DEMOTE_DATABASE_TEST_PASS=1") + + +plugin.monitor = lambda config, runtime_state: { + "ha": {"ocf_rc": plugin.OCF_SUCCESS}, + "primary": None, +} + +result = plugin.demote_database(config, runtime_state, {}) + +assert result["ocf_rc"] == 1 +assert result["reason"] == "local_not_primary" +print("local_not_primary rejected") + + +plugin.monitor = lambda config, runtime_state: { + "ha": {"ocf_rc": plugin.OCF_RUNNING_PROMOTED}, + "primary": {"current_wal_lsn": "14/1C07A920"}, +} + + +def unavailable(**kwargs): + raise plugin.psycopg2.OperationalError( + "connection refused" + ) + + +plugin.psycopg2.connect = unavailable + +result = plugin.demote_database(config, runtime_state, {}) + +assert result["ocf_rc"] == 1 +assert result["reason"] == "successor_unreachable" +print("successor_unreachable rejected") + +print("DEMOTE_DATABASE_FAILURE_TEST_PASS=1")