Extract managed-containerd proxy client

checksums/SHA256SUMS

@@ -15,7 +15,8 @@ d7d2fd7ab651d1f2f026bdb11d62e2c0c6ce6ee3128c9814d09550cf4304b3f2  ocf/loxthelion
 3a27ac9c63c61310da8b8711fa2c4400b37ea8eae2ee2a79e28bc63180b06c7c  ocf/loxthelion/managed-container.d/container.sh
 70431e0d390a3612339c7b5887286c1be4d6b305cbe8898e7895a211c26b6087  ocf/loxthelion/managed-container.d/engine.sh
 6c92c095ce10ac95d843fdc1c49e31c19491ca22fdac960affd8649606d76d82  ocf/loxthelion/managed-container.d/lifecycles/systemd.sh
-9f934833ac7d1ad845244f2fe9c3ce3d93787e36f3c36b0c5896364cbd82fd77  ocf/loxthelion/managed-container.d/proxy.sh
+c5d99182ce1e4cbff4fc7127a0727776729210ae7616ac6cfad2e6967d50bb8c  ocf/loxthelion/managed-container.d/proxy.sh
+16a060b64288de59ab60a4e2cb0ee233db8969f86b07f758003a38cda78e65c1  ocf/loxthelion/managed-container.d/proxy_client.py
 f82e2571d309bce8e4d7e6aee6b61b68feb7d99dc85bb8b688b6123c13ccf6a7  ocf/loxthelion/managed-container.d/runtimes/podman.sh
 4a99efea3d4a185e0630db89cfacc28aa8777bad9c747794ae69f70d87cd0327  pacemaker/README.md
 a42bb59a058ffeb59c1baa96e403856910bbbaea98f9104ff93daf4b5970c780  postgres-host/postgres-ha-bootstrap

ocf/loxthelion/managed-container.d/proxy.sh

@@ -1,99 +1,13 @@
 # Generic managed-containerd client helpers.
-# Shadow calls are observational only and must never affect OCF action results.
 
 MC_PROXY_SOCKET_DEFAULT="/run/user/1000/managed-containerd/managed-containerd.sock"
 
-mc_proxy_shadow_monitor()
+MC_PROXY_MODULE_DIR="$(
-{
+    cd "$(dirname "${BASH_SOURCE[0]}")" &&
-    local authoritative_rc="$1"
+    pwd
-    local socket_path="${MC_PROXY_SOCKET:-${MC_PROXY_SOCKET_DEFAULT}}"
+)"
-    local resource="${OCF_RESOURCE_INSTANCE:-}"
-    local response
 
-    resource="${resource%%:*}"
+MC_PROXY_CLIENT_DEFAULT="${MC_PROXY_MODULE_DIR}/proxy_client.py"
-
-    [ -n "${resource}" ] || return 0
-    [ -S "${socket_path}" ] || return 0
-    command -v python3 >/dev/null 2>&1 || return 0
-
-    response="$(
-        timeout --kill-after=1s 3s python3 - "${socket_path}" "${resource}" "${authoritative_rc}" <<'PY'
-import json
-import socket
-import sys
-
-path, resource, authoritative_raw = sys.argv[1], sys.argv[2], sys.argv[3]
-authoritative_rc = int(authoritative_raw)
-request = {"action": "monitor", "resource": resource}
-
-s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-s.settimeout(2.0)
-s.connect(path)
-s.sendall(json.dumps(request).encode("utf-8") + b"\n")
-
-raw = b""
-while b"\n" not in raw:
-    chunk = s.recv(65536)
-    if not chunk:
-        break
-    raw += chunk
-s.close()
-
-obj = json.loads(raw.decode("utf-8").strip())
-if not obj.get("ok"):
-    print("ERROR|" + str(obj.get("error", "unknown proxy error")))
-    raise SystemExit(0)
-
-result = obj.get("result") or {}
-ha = result.get("ha") or {}
-proxy_rc = ha.get("ocf_rc")
-
-if proxy_rc is None:
-    comparison = "UNCLASSIFIED"
-else:
-    comparison = "MATCH" if int(proxy_rc) == authoritative_rc else "MISMATCH"
-
-receiver = result.get("wal_receiver") or {}
-print("OK|{}|{}|{}|{}|{}|{}|{}|{}|{}|{}".format(
-    comparison,
-    "null" if proxy_rc is None else proxy_rc,
-    ha.get("health", "unknown"),
-    ha.get("ha_role", "unknown"),
-    ha.get("state", "unknown"),
-    str(ha.get("promotion_eligible", "unknown")).lower(),
-    result.get("role", "unknown"),
-    str(result.get("streaming", "unknown")).lower(),
-    receiver.get("slot_name") or "",
-    receiver.get("sender_host") or "",
-))
-PY
-    )" || response="TRANSPORT_ERROR"
-
-    case "${response}" in
-        OK\|MATCH\|*)
-            ocf_log info \
-                "managed-containerd shadow monitor: resource=${resource} authoritative_rc=${authoritative_rc} proxy=${response}"
-            ;;
-        OK\|UNCLASSIFIED\|*)
-            ocf_log warn \
-                "managed-containerd shadow monitor unclassified: resource=${resource} authoritative_rc=${authoritative_rc} proxy=${response}"
-            ;;
-        OK\|MISMATCH\|*)
-            ocf_log err \
-                "managed-containerd shadow monitor MISMATCH: resource=${resource} authoritative_rc=${authoritative_rc} proxy=${response}"
-            ;;
-        ERROR\|*)
-            ocf_log warn \
-                "managed-containerd shadow monitor failed: resource=${resource} authoritative_rc=${authoritative_rc} proxy=${response}"
-            ;;
-        *)
-            ocf_log warn \
-                "managed-containerd shadow monitor transport failure: resource=${resource} authoritative_rc=${authoritative_rc}"
-            ;;
-    esac
-
-    return 0
-}
 
 
 # Query managed-containerd for the authoritative monitor result.
@@ -133,116 +47,12 @@ mc_proxy_monitor()
 
     response="$(
         timeout --kill-after=1s 3s \
-        python3 - \
+            python3 \
+            "${MC_PROXY_CLIENT:-${MC_PROXY_CLIENT_DEFAULT}}" \
             "${socket_path}" \
             "${resource}" \
             "${MC_PROXY_RC_TRANSPORT}" \
-            "${MC_PROXY_RC_HARD_FAILURE}" <<'PY'
+            "${MC_PROXY_RC_HARD_FAILURE}"
-import json
-import socket
-import sys
-
-path, resource = sys.argv[1], sys.argv[2]
-TRANSPORT_FAILURE = int(sys.argv[3])
-PROTOCOL_FAILURE = int(sys.argv[4])
-
-PROTOCOL_VERSION = 1
-
-request = {
-    "protocol_version": PROTOCOL_VERSION,
-    "action": "monitor",
-    "resource": resource,
-}
-
-try:
-    s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-    s.settimeout(2.0)
-
-    try:
-        s.connect(path)
-        s.sendall(json.dumps(request).encode("utf-8") + b"\n")
-
-        raw = b""
-
-        while b"\n" not in raw:
-            chunk = s.recv(65536)
-
-            if not chunk:
-                break
-
-            raw += chunk
-    finally:
-        s.close()
-
-except (socket.timeout, ConnectionError, OSError):
-    # Local daemon/control-plane reachability failure.
-    # Recent authoritative state may be used for bounded grace.
-    raise SystemExit(TRANSPORT_FAILURE)
-
-# Transport succeeded. Invalid daemon output is therefore a
-# protocol/result failure and must never be hidden by cached health.
-if not raw:
-    raise SystemExit(PROTOCOL_FAILURE)
-
-try:
-    obj = json.loads(raw.decode("utf-8").strip())
-except (UnicodeDecodeError, json.JSONDecodeError):
-    raise SystemExit(PROTOCOL_FAILURE)
-
-if not isinstance(obj, dict):
-    raise SystemExit(PROTOCOL_FAILURE)
-
-# The daemon is reachable at this point, so any protocol mismatch is a hard
-# control-plane failure. It must never be hidden by cached application health.
-if obj.get("protocol_version") != PROTOCOL_VERSION:
-    raise SystemExit(PROTOCOL_FAILURE)
-
-if not obj.get("ok"):
-    raise SystemExit(PROTOCOL_FAILURE)
-
-if obj.get("resource") != resource:
-    raise SystemExit(PROTOCOL_FAILURE)
-
-if obj.get("action") != "monitor":
-    raise SystemExit(PROTOCOL_FAILURE)
-
-result = obj.get("result")
-if not isinstance(result, dict):
-    raise SystemExit(PROTOCOL_FAILURE)
-
-ha = result.get("ha")
-if not isinstance(ha, dict):
-    raise SystemExit(PROTOCOL_FAILURE)
-
-rc = ha.get("ocf_rc")
-
-try:
-    rc = int(rc)
-except (TypeError, ValueError):
-    raise SystemExit(PROTOCOL_FAILURE)
-
-if rc not in (0, 1, 7, 8):
-    raise SystemExit(PROTOCOL_FAILURE)
-
-receiver = result.get("wal_receiver") or {}
-
-if not isinstance(receiver, dict):
-    raise SystemExit(PROTOCOL_FAILURE)
-
-payload = "|".join([
-    str(rc),
-    str(ha.get("health", "unknown")),
-    str(ha.get("ha_role", "unknown")),
-    str(ha.get("state", "unknown")),
-    str(ha.get("promotion_eligible", "unknown")).lower(),
-    str(result.get("role", "unknown")),
-    str(result.get("streaming", "unknown")).lower(),
-    str(receiver.get("slot_name") or ""),
-    str(receiver.get("sender_host") or ""),
-])
-
-print(payload)
-PY
     )"
     rc=$?
 

ocf/loxthelion/managed-container.d/proxy_client.py

@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+
+import json
+import socket
+import sys
+
+
+PROTOCOL_VERSION = 1
+VALID_OCF_RCS = {0, 1, 7, 8}
+
+
+def main():
+    if len(sys.argv) != 5:
+        raise SystemExit(11)
+
+    socket_path = sys.argv[1]
+    resource = sys.argv[2]
+    transport_failure = int(sys.argv[3])
+    protocol_failure = int(sys.argv[4])
+
+    request = {
+        "protocol_version": PROTOCOL_VERSION,
+        "action": "monitor",
+        "resource": resource,
+    }
+
+    try:
+        sock = socket.socket(
+            socket.AF_UNIX,
+            socket.SOCK_STREAM,
+        )
+        sock.settimeout(2.0)
+
+        try:
+            sock.connect(socket_path)
+
+            sock.sendall(
+                json.dumps(request).encode("utf-8")
+                + b"\n"
+            )
+
+            raw = b""
+
+            while b"\n" not in raw:
+                chunk = sock.recv(65536)
+
+                if not chunk:
+                    break
+
+                raw += chunk
+
+        finally:
+            sock.close()
+
+    except (
+        socket.timeout,
+        ConnectionError,
+        OSError,
+    ):
+        raise SystemExit(transport_failure)
+
+    # A connection was established. From this point onward, absence of a
+    # valid response is a protocol/result failure rather than transport loss.
+    if not raw:
+        raise SystemExit(protocol_failure)
+
+    try:
+        obj = json.loads(
+            raw.decode("utf-8").strip()
+        )
+
+    except (
+        UnicodeDecodeError,
+        json.JSONDecodeError,
+    ):
+        raise SystemExit(protocol_failure)
+
+    if not isinstance(obj, dict):
+        raise SystemExit(protocol_failure)
+
+    if obj.get("protocol_version") != PROTOCOL_VERSION:
+        raise SystemExit(protocol_failure)
+
+    if obj.get("ok") is not True:
+        raise SystemExit(protocol_failure)
+
+    if obj.get("resource") != resource:
+        raise SystemExit(protocol_failure)
+
+    if obj.get("action") != "monitor":
+        raise SystemExit(protocol_failure)
+
+    result = obj.get("result")
+
+    if not isinstance(result, dict):
+        raise SystemExit(protocol_failure)
+
+    ha = result.get("ha")
+
+    if not isinstance(ha, dict):
+        raise SystemExit(protocol_failure)
+
+    try:
+        ocf_rc = int(ha.get("ocf_rc"))
+
+    except (
+        TypeError,
+        ValueError,
+    ):
+        raise SystemExit(protocol_failure)
+
+    if ocf_rc not in VALID_OCF_RCS:
+        raise SystemExit(protocol_failure)
+
+    receiver = result.get("wal_receiver") or {}
+
+    if not isinstance(receiver, dict):
+        raise SystemExit(protocol_failure)
+
+    payload = "|".join([
+        str(ocf_rc),
+        str(ha.get("health", "unknown")),
+        str(ha.get("ha_role", "unknown")),
+        str(ha.get("state", "unknown")),
+        str(
+            ha.get(
+                "promotion_eligible",
+                "unknown",
+            )
+        ).lower(),
+        str(result.get("role", "unknown")),
+        str(
+            result.get(
+                "streaming",
+                "unknown",
+            )
+        ).lower(),
+        str(receiver.get("slot_name") or ""),
+        str(receiver.get("sender_host") or ""),
+    ])
+
+    print(payload)
+
+
+if __name__ == "__main__":
+    main()