Add selectable PostgreSQL monitor backends

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,

plugins/postgresql/manifest.json

@@ -1,11 +1,12 @@
 {
   "name": "postgresql",
-  "version": 3,
+  "version": 5,
   "module": "plugin.py",
   "actions": [
     "monitor",
     "notify",
     "notify.target",
+    "promote.database",
     "demote.prepare",
     "demote.database"
   ],
@@ -19,6 +20,10 @@
   ],
   "optional_config": [
     "PG_RECONNECT_GRACE_SECONDS",
-    "PACEMAKER_NODE_NAME"
+    "PACEMAKER_NODE_NAME",
+    "MONITOR_BACKEND",
+    "CONTAINER_RUNTIME",
+    "CONTAINER_NAME",
+    "PG_OS_USER"
   ]
 }

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",
@@ -531,6 +1021,191 @@ def notify_target(config, runtime_state, request_parameters):
     }
 
 
+
+def promote_database(config, runtime_state, request_parameters):
+    """
+    Read-only PostgreSQL promotion preflight.
+
+    Validate that Pacemaker selected this local node for promotion, that the
+    local PostgreSQL instance is still an eligible standby, and that no
+    reachable configured peer is currently writable.
+
+    This action never promotes PostgreSQL.  The OCF resource agent remains
+    authoritative for the actual pg_ctl promote operation and for local
+    filesystem state such as the rejoin-required marker.
+    """
+    del request_parameters
+
+    node_name = str(
+        config.get("PACEMAKER_NODE_NAME", "")
+    ).strip().lower()
+
+    if not node_name:
+        raise ValueError(
+            "PACEMAKER_NODE_NAME is required for promote.database"
+        )
+
+    notification = _postgres_runtime_state(
+        runtime_state
+    ).get("notification")
+
+    if not isinstance(notification, dict):
+        return {
+            "ocf_rc": 1,
+            "ready": False,
+            "mutation": False,
+            "reason": "notification_missing",
+            "node": node_name,
+        }
+
+    if notification.get("type") != "pre":
+        return {
+            "ocf_rc": 1,
+            "ready": False,
+            "mutation": False,
+            "reason": "notification_type_invalid",
+            "node": node_name,
+        }
+
+    if notification.get("operation") != "promote":
+        return {
+            "ocf_rc": 1,
+            "ready": False,
+            "mutation": False,
+            "reason": "notification_operation_invalid",
+            "node": node_name,
+        }
+
+    received_at = notification.get("received_at")
+
+    if not isinstance(received_at, (int, float)):
+        return {
+            "ocf_rc": 1,
+            "ready": False,
+            "mutation": False,
+            "reason": "notification_timestamp_invalid",
+            "node": node_name,
+        }
+
+    age = time.time() - received_at
+
+    if age < 0 or age > NOTIFY_TARGET_MAX_AGE_SECONDS:
+        return {
+            "ocf_rc": 1,
+            "ready": False,
+            "mutation": False,
+            "reason": "notification_stale",
+            "age_seconds": age,
+            "node": node_name,
+        }
+
+    target_node = str(
+        notification.get("promote_uname", "")
+    ).strip().lower()
+
+    previous_primary_node = str(
+        notification.get("demote_uname", "")
+    ).strip().lower()
+
+    if not target_node:
+        return {
+            "ocf_rc": 1,
+            "ready": False,
+            "mutation": False,
+            "reason": "promotion_target_missing",
+            "node": node_name,
+        }
+
+    if target_node != node_name:
+        return {
+            "ocf_rc": 1,
+            "ready": False,
+            "mutation": False,
+            "reason": "local_node_not_target",
+            "node": node_name,
+            "target_node": target_node,
+            "previous_primary_node": previous_primary_node,
+        }
+
+    local = monitor(config, runtime_state)
+    local_ha = local.get("ha") or {}
+
+    checks = {
+        "notification": True,
+        "local_target": True,
+        "local_monitor_success": (
+            local_ha.get("ocf_rc") == OCF_SUCCESS
+        ),
+        "local_standby": (
+            local.get("role") == "standby"
+            and local.get("in_recovery") is True
+        ),
+        "promotion_eligible": (
+            local_ha.get("promotion_eligible") is True
+        ),
+    }
+
+    if not all(checks.values()):
+        return {
+            "ocf_rc": 1,
+            "ready": False,
+            "mutation": False,
+            "reason": "local_not_promotion_eligible",
+            "node": node_name,
+            "target_node": target_node,
+            "previous_primary_node": previous_primary_node,
+            "local": local_ha,
+            "checks": checks,
+        }
+
+    try:
+        password = read_secret(config["PGPASSWORD_FILE"])
+        peer_state = probe_replication_peers(config, password)
+    except (KeyError, TypeError, ValueError, psycopg2.Error) as exc:
+        return {
+            "ocf_rc": 1,
+            "ready": False,
+            "mutation": False,
+            "reason": "peer_probe_failed",
+            "node": node_name,
+            "target_node": target_node,
+            "previous_primary_node": previous_primary_node,
+            "error": str(exc).splitlines()[0],
+            "checks": checks,
+        }
+
+    writable_primaries = [
+        str(name).strip().lower()
+        for name in peer_state.get("writable_primaries", [])
+    ]
+
+    checks["no_reachable_writable_primary"] = (
+        len(writable_primaries) == 0
+    )
+
+    ready = all(checks.values())
+
+    return {
+        "ocf_rc": OCF_SUCCESS if ready else OCF_ERR_GENERIC,
+        "ready": ready,
+        "mutation": False,
+        "reason": (
+            "promotion_preflight_ready"
+            if ready
+            else "reachable_writable_primary"
+        ),
+        "node": node_name,
+        "target_node": target_node,
+        "previous_primary_node": previous_primary_node,
+        "writable_primaries": writable_primaries,
+        "writable_primary_count": len(writable_primaries),
+        "peer_state": peer_state,
+        "checks": checks,
+        "age_seconds": age,
+    }
+
+
+
 def demote_prepare(config, runtime_state, request_parameters):
     """
     Read-only demotion preparation.
@@ -617,6 +1292,47 @@ def demote_prepare(config, runtime_state, request_parameters):
             "successor_node": successor_node,
         }
 
+    promoted_nodes = {
+        str(name).strip().lower()
+        for name in str(
+            notification.get("promoted_uname", "")
+        ).split()
+        if str(name).strip()
+    }
+
+    if successor_node == node_name:
+        if node_name not in promoted_nodes:
+            return {
+                "ocf_rc": 1,
+                "ready": False,
+                "reason": "self_restart_not_current_primary",
+                "node": node_name,
+                "target_node": target_node,
+                "successor_node": successor_node,
+            }
+
+        return {
+            "ocf_rc": 0,
+            "ready": False,
+            "identity_validated": True,
+            "reason": "local_restart_checks_pending",
+            "mutation": False,
+            "node": node_name,
+            "target_node": target_node,
+            "successor_node": successor_node,
+            "handoff_mode": "local_restart",
+            "final_handoff_required": False,
+            "age_seconds": age,
+            "checks": {
+                "notification": True,
+                "local_target": True,
+                "self_successor": True,
+                "local_currently_promoted": True,
+                "postgresql_mutated": False,
+                "database_checks": False,
+            },
+        }
+
     if not successor_node:
         return {
             "ocf_rc": 1,
@@ -700,6 +1416,7 @@ def demote_database(config, runtime_state, request_parameters):
 
     if preparation.get("reason") not in (
         "database_checks_pending",
+        "local_restart_checks_pending",
     ):
         return preparation
 
@@ -713,6 +1430,26 @@ def demote_database(config, runtime_state, request_parameters):
             "local": local["ha"],
         }
 
+    if preparation.get("reason") == "local_restart_checks_pending":
+        return {
+            "ocf_rc": 0,
+            "ready": True,
+            "mutation": False,
+            "final_handoff_required": False,
+            "handoff_mode": "local_restart",
+            "reason": "local_restart_demotion_ready",
+            "node": preparation["node"],
+            "target_node": preparation["target_node"],
+            "successor_node": preparation["successor_node"],
+            "checks": {
+                "notification": True,
+                "local_target": True,
+                "local_primary": True,
+                "self_successor": True,
+                "remote_handoff_required": False,
+            },
+        }
+
     successor_node = preparation["successor_node"]
     peers = parse_replication_peers(
         config.get("PG_REPLICATION_PEERS", "")

tests/test-demote-database.py

@@ -137,3 +137,56 @@ assert result["reason"] == "successor_unreachable"
 print("successor_unreachable    rejected")
 
 print("DEMOTE_DATABASE_FAILURE_TEST_PASS=1")
+
+print()
+print("===== SELF-RESTART DATABASE GATE =====")
+
+plugin.monitor = lambda config, runtime_state: {
+    "ha": {"ocf_rc": plugin.OCF_RUNNING_PROMOTED},
+    "primary": {"current_wal_lsn": "14/1C07A920"},
+}
+
+self_restart_state = {
+    "postgresql": {
+        "notification": {
+            "type": "pre",
+            "operation": "demote",
+            "demote_uname": "david-prodesk",
+            "promote_uname": "david-prodesk",
+            "promoted_uname": "david-prodesk",
+            "unpromoted_uname": "david-eve",
+            "received_at": time.time(),
+        },
+    },
+}
+
+connect_called = False
+
+def must_not_connect(**kwargs):
+    global connect_called
+    connect_called = True
+    raise AssertionError(
+        "self-restart demotion must not probe a remote successor"
+    )
+
+plugin.psycopg2.connect = must_not_connect
+
+result = plugin.demote_database(
+    config,
+    self_restart_state,
+    {},
+)
+
+assert result["ocf_rc"] == 0
+assert result["ready"] is True
+assert result["mutation"] is False
+assert result["final_handoff_required"] is False
+assert result["handoff_mode"] == "local_restart"
+assert result["reason"] == "local_restart_demotion_ready"
+assert result["successor_node"] == "david-prodesk"
+assert result["checks"]["local_primary"] is True
+assert result["checks"]["remote_handoff_required"] is False
+assert connect_called is False
+
+print("self_restart_database     accepted")
+print("remote_probe_skipped      accepted")

tests/test-demote-prepare.py

@@ -124,3 +124,63 @@ result = plugin.demote_prepare(
 assert result["ocf_rc"] == 1
 assert result["reason"] == "successor_not_configured"
 print("unconfigured_successor  rejected")
+
+print()
+print("===== SELF-RESTART PREPARE =====")
+
+self_restart_state = {
+    "postgresql": {
+        "notification": {
+            "type": "pre",
+            "operation": "demote",
+            "demote_uname": "david-eve",
+            "promote_uname": "david-eve",
+            "promoted_uname": "david-eve",
+            "unpromoted_uname": "david-prodesk david-macbookpro",
+            "received_at": time.time(),
+        },
+    },
+}
+
+result = plugin.demote_prepare(
+    {
+        "PACEMAKER_NODE_NAME": "david-eve",
+        "PG_REPLICATION_PEERS": (
+            "david-prodesk=192.168.1.4:5432,"
+            "david-eve=192.168.1.3:5432"
+        ),
+    },
+    self_restart_state,
+    {},
+)
+
+assert result["ocf_rc"] == 0
+assert result["ready"] is False
+assert result["identity_validated"] is True
+assert result["reason"] == "local_restart_checks_pending"
+assert result["handoff_mode"] == "local_restart"
+assert result["final_handoff_required"] is False
+assert result["successor_node"] == "david-eve"
+assert result["checks"]["self_successor"] is True
+assert result["checks"]["local_currently_promoted"] is True
+
+print("self_restart_prepare      accepted")
+
+self_restart_state["postgresql"]["notification"]["promoted_uname"] = ""
+
+result = plugin.demote_prepare(
+    {
+        "PACEMAKER_NODE_NAME": "david-eve",
+        "PG_REPLICATION_PEERS": (
+            "david-prodesk=192.168.1.4:5432,"
+            "david-eve=192.168.1.3:5432"
+        ),
+    },
+    self_restart_state,
+    {},
+)
+
+assert result["ocf_rc"] == 1
+assert result["reason"] == "self_restart_not_current_primary"
+
+print("false_self_restart        rejected")

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")

tests/test-promote-database.py

@@ -0,0 +1,225 @@
+#!/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)
+
+
+def standby_monitor(*, eligible=True):
+    return {
+        "ha": {
+            "ocf_rc": plugin.OCF_SUCCESS,
+            "health": "healthy",
+            "ha_role": "unpromoted",
+            "state": "streaming",
+            "promotion_eligible": eligible,
+        },
+        "role": "standby",
+        "in_recovery": True,
+        "streaming": True,
+    }
+
+
+def peer_state(*primaries):
+    return {
+        "peers": [],
+        "writable_primaries": list(primaries),
+        "writable_primary_count": len(primaries),
+    }
+
+
+config = {
+    "PACEMAKER_NODE_NAME": "david-eve",
+    "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"
+    ),
+    "PGUSER": "david",
+    "PGDATABASE": "david",
+}
+
+
+def fresh_state():
+    return {
+        "postgresql": {
+            "notification": {
+                "type": "pre",
+                "operation": "promote",
+                "promote_uname": "david-eve",
+                "demote_uname": "david-prodesk",
+                "received_at": time.time(),
+            },
+        },
+    }
+
+
+plugin.read_secret = lambda path: "test-password"
+plugin.monitor = lambda config, runtime_state: standby_monitor()
+plugin.probe_replication_peers = (
+    lambda config, password: peer_state()
+)
+
+
+print("===== PROMOTE DATABASE ACTION =====")
+
+runtime_state = fresh_state()
+
+result = plugin.promote_database(
+    config,
+    runtime_state,
+    {},
+)
+
+assert result["ocf_rc"] == plugin.OCF_SUCCESS
+assert result["ready"] is True
+assert result["mutation"] is False
+assert result["reason"] == "promotion_preflight_ready"
+assert result["target_node"] == "david-eve"
+assert result["previous_primary_node"] == "david-prodesk"
+assert result["writable_primary_count"] == 0
+assert result["checks"]["local_target"] is True
+assert result["checks"]["local_standby"] is True
+assert result["checks"]["promotion_eligible"] is True
+assert result["checks"]["no_reachable_writable_primary"] is True
+
+print("valid_promotion_preflight accepted")
+
+
+runtime_state = fresh_state()
+runtime_state["postgresql"]["notification"]["promote_uname"] = (
+    "david-macbookpro"
+)
+
+result = plugin.promote_database(
+    config,
+    runtime_state,
+    {},
+)
+
+assert result["ocf_rc"] == 1
+assert result["ready"] is False
+assert result["reason"] == "local_node_not_target"
+
+print("wrong_target              rejected")
+
+
+runtime_state = fresh_state()
+runtime_state["postgresql"]["notification"]["received_at"] = (
+    time.time() - 121
+)
+
+result = plugin.promote_database(
+    config,
+    runtime_state,
+    {},
+)
+
+assert result["ocf_rc"] == 1
+assert result["reason"] == "notification_stale"
+
+print("stale_notification        rejected")
+
+
+result = plugin.promote_database(
+    config,
+    {},
+    {},
+)
+
+assert result["ocf_rc"] == 1
+assert result["reason"] == "notification_missing"
+
+print("missing_notification      rejected")
+
+
+runtime_state = fresh_state()
+
+plugin.monitor = lambda config, runtime_state: {
+    "ha": {
+        "ocf_rc": plugin.OCF_RUNNING_PROMOTED,
+        "promotion_eligible": True,
+    },
+    "role": "primary",
+    "in_recovery": False,
+}
+
+result = plugin.promote_database(
+    config,
+    runtime_state,
+    {},
+)
+
+assert result["ocf_rc"] == 1
+assert result["ready"] is False
+assert result["reason"] == "local_not_promotion_eligible"
+assert result["checks"]["local_standby"] is False
+
+print("local_primary             rejected")
+
+
+plugin.monitor = lambda config, runtime_state: standby_monitor(
+    eligible=False
+)
+
+result = plugin.promote_database(
+    config,
+    fresh_state(),
+    {},
+)
+
+assert result["ocf_rc"] == 1
+assert result["ready"] is False
+assert result["checks"]["promotion_eligible"] is False
+
+print("ineligible_standby        rejected")
+
+
+plugin.monitor = lambda config, runtime_state: standby_monitor()
+plugin.probe_replication_peers = (
+    lambda config, password: peer_state("david-prodesk")
+)
+
+result = plugin.promote_database(
+    config,
+    fresh_state(),
+    {},
+)
+
+assert result["ocf_rc"] == plugin.OCF_ERR_GENERIC
+assert result["ready"] is False
+assert result["mutation"] is False
+assert result["reason"] == "reachable_writable_primary"
+assert result["writable_primary_count"] == 1
+assert result["writable_primaries"] == ["david-prodesk"]
+assert result["checks"]["no_reachable_writable_primary"] is False
+
+print("existing_primary          rejected")
+
+
+print()
+print("PROMOTE_DATABASE_TEST_PASS=1")

tests/test-runtime-state.py

@@ -179,8 +179,8 @@ production_manifest = json.loads(
 )
 
 require(
-    production_manifest.get("version") == 3,
+    production_manifest.get("version") == 5,
-    "PostgreSQL manifest version is not 3",
+    "PostgreSQL manifest version is not 5",
 )
 require(
     "PG_REPLICATION_PEERS"