Proven working Postgresql monitor

Containerfile

@@ -0,0 +1,10 @@
+FROM registry.fedoraproject.org/fedora:44
+
+RUN dnf -y install \
+        python3 \
+        python3-psycopg2 \
+    && dnf clean all
+
+COPY daemon/ /opt/managed-containerd/daemon/
+
+ENTRYPOINT ["python3", "/opt/managed-containerd/daemon/server.py"]

daemon/registry.py

@@ -0,0 +1,174 @@
+#!/usr/bin/env python3
+
+import importlib.util
+import json
+import os
+from pathlib import Path
+
+
+class RegistryError(Exception):
+    pass
+
+
+class Plugin:
+    def __init__(self, plugin_root, name):
+        self.name = name
+        self.path = Path(plugin_root) / name
+        self.manifest_path = self.path / "manifest.json"
+
+        if not self.manifest_path.is_file():
+            raise RegistryError(
+                f"plugin manifest not found: {self.manifest_path}"
+            )
+
+        with self.manifest_path.open("r", encoding="utf-8") as f:
+            self.manifest = json.load(f)
+
+        module_name = self.manifest.get("module", "plugin.py")
+        module_path = self.path / module_name
+
+        if not module_path.is_file():
+            raise RegistryError(
+                f"plugin module not found: {module_path}"
+            )
+
+        spec = importlib.util.spec_from_file_location(
+            f"managed_containerd_plugin_{name}",
+            module_path,
+        )
+
+        if spec is None or spec.loader is None:
+            raise RegistryError(
+                f"unable to load plugin: {name}"
+            )
+
+        self.module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(self.module)
+
+    @property
+    def actions(self):
+        return list(self.manifest.get("actions", []))
+
+    @property
+    def required_config(self):
+        return list(
+            self.manifest.get("required_config", [])
+        )
+
+    def validate_config(self, config):
+        missing = [
+            key
+            for key in self.required_config
+            if not config.get(key)
+        ]
+
+        if missing:
+            raise RegistryError(
+                "missing required configuration: "
+                + ", ".join(missing)
+            )
+
+    def execute(self, action, config):
+        if action not in self.actions:
+            raise RegistryError(
+                f"plugin {self.name} does not support action {action}"
+            )
+
+        handler = getattr(self.module, action, None)
+
+        if handler is None:
+            raise RegistryError(
+                f"plugin {self.name} advertises {action} "
+                "but does not implement it"
+            )
+
+        return handler(config)
+
+
+class Resource:
+    def __init__(self, resource_id, plugin, config):
+        self.resource_id = resource_id
+        self.plugin = plugin
+        self.config = config
+
+        self.plugin.validate_config(config)
+
+    def describe(self):
+        return {
+            "resource": self.resource_id,
+            "plugin": self.plugin.name,
+            "plugin_version": self.plugin.manifest.get(
+                "version"
+            ),
+            "actions": self.plugin.actions,
+            "configured": True,
+        }
+
+
+class Registry:
+    def __init__(self):
+        self.plugin_root = os.environ.get(
+            "MANAGED_CONTAINERD_PLUGIN_ROOT",
+            "/opt/managed-containerd/plugins",
+        )
+
+        self.resources = {}
+        self._load_environment_resource()
+
+    def _load_environment_resource(self):
+        resource_id = os.environ.get("RESOURCE_ID")
+        plugin_name = os.environ.get("PLUGIN")
+
+        if not resource_id:
+            raise RegistryError(
+                "RESOURCE_ID is not configured"
+            )
+
+        if not plugin_name:
+            raise RegistryError(
+                "PLUGIN is not configured"
+            )
+
+        plugin = Plugin(
+            self.plugin_root,
+            plugin_name,
+        )
+
+        config = {
+            key: os.environ.get(key)
+            for key in plugin.required_config
+        }
+
+        self.resources[resource_id] = Resource(
+            resource_id,
+            plugin,
+            config,
+        )
+
+    def list_resources(self):
+        return {
+            resource_id: resource.describe()
+            for resource_id, resource
+            in self.resources.items()
+        }
+
+    def execute(self, resource_id, action):
+        resource = self.resources.get(resource_id)
+
+        if resource is None:
+            raise RegistryError(
+                f"unknown resource: {resource_id}"
+            )
+
+        result = resource.plugin.execute(
+            action,
+            resource.config,
+        )
+
+        return {
+            "ok": True,
+            "resource": resource_id,
+            "plugin": resource.plugin.name,
+            "action": action,
+            "result": result,
+        }

daemon/server.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+
+import json
+import os
+import signal
+import socket
+import traceback
+
+from registry import Registry, RegistryError
+
+
+SOCKET_PATH = os.environ.get(
+    "MANAGED_CONTAINERD_SOCKET",
+    "/run/managed-containerd/managed-containerd.sock",
+)
+
+server_socket = None
+registry = None
+
+
+def handle_request(req):
+    action = req.get("action")
+
+    if action == "ping":
+        return {
+            "ok": True,
+            "daemon": "managed-containerd",
+            "protocol_version": 1,
+        }
+
+    if action == "registry.list":
+        return {
+            "ok": True,
+            "resources": registry.list_resources(),
+        }
+
+    resource = req.get("resource")
+
+    if not resource:
+        return {
+            "ok": False,
+            "error": "resource is required",
+        }
+
+    return registry.execute(
+        resource,
+        action,
+    )
+
+
+def cleanup_socket():
+    try:
+        os.unlink(SOCKET_PATH)
+    except FileNotFoundError:
+        pass
+
+
+def shutdown(signum, frame):
+    global server_socket
+
+    print(
+        f"managed-containerd received signal {signum}; shutting down",
+        flush=True,
+    )
+
+    if server_socket is not None:
+        try:
+            server_socket.close()
+        except OSError:
+            pass
+
+    cleanup_socket()
+    raise SystemExit(0)
+
+
+def serve():
+    global server_socket
+    global registry
+
+    registry = Registry()
+
+    os.makedirs(
+        os.path.dirname(SOCKET_PATH),
+        exist_ok=True,
+    )
+
+    cleanup_socket()
+
+    signal.signal(signal.SIGTERM, shutdown)
+    signal.signal(signal.SIGINT, shutdown)
+
+    server_socket = socket.socket(
+        socket.AF_UNIX,
+        socket.SOCK_STREAM,
+    )
+
+    server_socket.bind(SOCKET_PATH)
+    os.chmod(SOCKET_PATH, 0o660)
+    server_socket.listen(16)
+
+    print(
+        f"managed-containerd listening on {SOCKET_PATH}",
+        flush=True,
+    )
+
+    print(
+        "registered resources: "
+        + ", ".join(registry.resources),
+        flush=True,
+    )
+
+    try:
+        while True:
+            conn, _ = server_socket.accept()
+
+            with conn:
+                try:
+                    raw = b""
+
+                    while b"\n" not in raw:
+                        chunk = conn.recv(65536)
+
+                        if not chunk:
+                            break
+
+                        raw += chunk
+
+                    req = json.loads(
+                        raw.decode("utf-8").strip()
+                    )
+
+                    response = handle_request(req)
+
+                except (
+                    RegistryError,
+                    ValueError,
+                    KeyError,
+                ) as exc:
+                    response = {
+                        "ok": False,
+                        "error": str(exc),
+                    }
+
+                except Exception as exc:
+                    traceback.print_exc()
+
+                    response = {
+                        "ok": False,
+                        "error": str(exc),
+                    }
+
+                conn.sendall(
+                    json.dumps(
+                        response,
+                        default=str,
+                    ).encode("utf-8")
+                    + b"\n"
+                )
+
+    finally:
+        cleanup_socket()
+
+
+if __name__ == "__main__":
+    serve()

plugins/postgresql/manifest.json

@@ -0,0 +1,15 @@
+{
+  "name": "postgresql",
+  "version": 1,
+  "module": "plugin.py",
+  "actions": [
+    "monitor"
+  ],
+  "required_config": [
+    "PGHOST",
+    "PGPORT",
+    "PGUSER",
+    "PGDATABASE",
+    "PGPASSWORD_FILE"
+  ]
+}

plugins/postgresql/plugin.py

@@ -0,0 +1,139 @@
+#!/usr/bin/env python3
+
+import psycopg2
+import psycopg2.extras
+
+
+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"]
+    )
+
+    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",
+    )
+
+    try:
+        conn.autocommit = True
+
+        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"]
+            )
+
+            if not in_recovery:
+                cur.execute("""
+                    SELECT
+                        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
+                """)
+
+                recovery = cur.fetchone()
+
+                cur.execute("""
+                    SELECT
+                        status,
+                        sender_host,
+                        sender_port,
+                        slot_name,
+                        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_user AS current_user,
+                    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"
+            )
+
+        else:
+            role = "primary"
+            streaming = False
+
+        return {
+            "role": role,
+            "in_recovery": in_recovery,
+            "streaming": streaming,
+            "wal_receiver": (
+                dict(receiver)
+                if receiver
+                else None
+            ),
+            "recovery": dict(recovery),
+            "primary": (
+                dict(primary)
+                if primary
+                else None
+            ),
+            "server": dict(server),
+        }
+
+    finally:
+        conn.close()

postgres.env

@@ -0,0 +1,11 @@
+RESOURCE_ID=postgresql-ha
+PLUGIN=postgresql
+
+MANAGED_CONTAINERD_SOCKET=/run/managed-containerd/managed-containerd.sock
+MANAGED_CONTAINERD_PLUGIN_ROOT=/opt/managed-containerd/plugins
+
+PGHOST=192.168.1.4
+PGPORT=5432
+PGUSER=david
+PGDATABASE=david
+PGPASSWORD_FILE=/run/secrets/postgres_password

server.py

@@ -0,0 +1,255 @@
+#!/usr/bin/env python3
+
+import json
+import os
+import signal
+import socket
+import traceback
+
+import psycopg2
+import psycopg2.extras
+
+
+SOCKET_PATH = os.environ.get(
+    "MANAGED_CONTAINERD_SOCKET",
+    "/run/managed-containerd/managed-containerd.sock",
+)
+
+PGHOST = os.environ.get("PGHOST", "postgres")
+PGPORT = int(os.environ.get("PGPORT", "5432"))
+PGUSER = os.environ.get("PGUSER", "david")
+PGDATABASE = os.environ.get("PGDATABASE", "david")
+PGPASSWORD_FILE = os.environ.get("PGPASSWORD_FILE")
+
+server_socket = None
+
+
+def read_secret(path):
+    if not path:
+        return None
+
+    with open(path, "r", encoding="utf-8") as f:
+        return f.read().strip()
+
+
+def postgres_status():
+    password = read_secret(PGPASSWORD_FILE)
+
+    conn = psycopg2.connect(
+        host=PGHOST,
+        port=PGPORT,
+        user=PGUSER,
+        dbname=PGDATABASE,
+        password=password,
+        connect_timeout=5,
+        application_name="managed-containerd",
+    )
+
+    try:
+        conn.autocommit = True
+
+        with conn.cursor(
+            cursor_factory=psycopg2.extras.RealDictCursor
+        ) as cur:
+
+            #
+            # Establish role first.
+            #
+            cur.execute("""
+                SELECT pg_is_in_recovery() AS in_recovery
+            """)
+            in_recovery = bool(cur.fetchone()["in_recovery"])
+
+            #
+            # Primary-only state.
+            #
+            if not in_recovery:
+                cur.execute("""
+                    SELECT
+                        pg_current_wal_lsn()::text AS current_wal_lsn
+                """)
+                primary = cur.fetchone()
+            else:
+                primary = None
+
+            #
+            # Recovery state.
+            #
+            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
+                """)
+                recovery = cur.fetchone()
+
+                cur.execute("""
+                    SELECT
+                        status,
+                        sender_host,
+                        sender_port,
+                        slot_name,
+                        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
+
+            #
+            # General server state.
+            #
+            cur.execute("""
+                SELECT
+                    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
+            """)
+            server = cur.fetchone()
+
+        if in_recovery:
+            role = "standby"
+            streaming = bool(
+                receiver and receiver.get("status") == "streaming"
+            )
+        else:
+            role = "primary"
+            streaming = False
+
+        return {
+            "ok": True,
+            "application": "postgresql",
+            "role": role,
+            "in_recovery": in_recovery,
+            "streaming": streaming,
+            "wal_receiver": dict(receiver) if receiver else None,
+            "recovery": dict(recovery),
+            "primary": dict(primary) if primary else None,
+            "server": dict(server),
+        }
+
+    finally:
+        conn.close()
+
+
+def handle_request(req):
+    action = req.get("action")
+
+    if action == "ping":
+        return {
+            "ok": True,
+            "daemon": "managed-containerd",
+            "version": 1,
+        }
+
+    if action == "postgresql.status":
+        return postgres_status()
+
+    return {
+        "ok": False,
+        "error": f"unknown action: {action}",
+    }
+
+
+def cleanup_socket():
+    try:
+        os.unlink(SOCKET_PATH)
+    except FileNotFoundError:
+        pass
+
+
+def shutdown(signum, frame):
+    global server_socket
+
+    print(
+        f"managed-containerd received signal {signum}; shutting down",
+        flush=True,
+    )
+
+    if server_socket is not None:
+        try:
+            server_socket.close()
+        except OSError:
+            pass
+
+    cleanup_socket()
+    raise SystemExit(0)
+
+
+def serve():
+    global server_socket
+
+    os.makedirs(os.path.dirname(SOCKET_PATH), exist_ok=True)
+    cleanup_socket()
+
+    signal.signal(signal.SIGTERM, shutdown)
+    signal.signal(signal.SIGINT, shutdown)
+
+    server_socket = socket.socket(
+        socket.AF_UNIX,
+        socket.SOCK_STREAM,
+    )
+
+    server_socket.bind(SOCKET_PATH)
+    os.chmod(SOCKET_PATH, 0o660)
+    server_socket.listen(16)
+
+    print(
+        f"managed-containerd listening on {SOCKET_PATH}",
+        flush=True,
+    )
+
+    try:
+        while True:
+            conn, _ = server_socket.accept()
+
+            with conn:
+                try:
+                    raw = b""
+
+                    while b"\n" not in raw:
+                        chunk = conn.recv(65536)
+
+                        if not chunk:
+                            break
+
+                        raw += chunk
+
+                    req = json.loads(
+                        raw.decode("utf-8").strip()
+                    )
+
+                    response = handle_request(req)
+
+                except Exception as exc:
+                    traceback.print_exc()
+
+                    response = {
+                        "ok": False,
+                        "error": str(exc),
+                    }
+
+                conn.sendall(
+                    json.dumps(
+                        response,
+                        default=str,
+                    ).encode("utf-8") + b"\n"
+                )
+
+    finally:
+        cleanup_socket()
+
+
+if __name__ == "__main__":
+    serve()