Generalize protocol v1 proxy client actions

checksums/SHA256SUMS

@@ -15,14 +15,15 @@ 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
-c5d99182ce1e4cbff4fc7127a0727776729210ae7616ac6cfad2e6967d50bb8c  ocf/loxthelion/managed-container.d/proxy.sh
+5f3528932ea72ef42c18465e7f31f9a430407e8d4ff7c03c705fe8cae39dc26e  ocf/loxthelion/managed-container.d/proxy.sh
-16a060b64288de59ab60a4e2cb0ee233db8969f86b07f758003a38cda78e65c1  ocf/loxthelion/managed-container.d/proxy_client.py
+fb01359a84062ec3e910a8b27e460953a0ae1641a69167b3497f2594911d2a00  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
 957a5a72ed277d0216e070abb80733483be3d421b9fa488e27fdd75bb2919548  postgres-host/postgres-ha-peer-config.sh
 deb09e3ae40629f43b262ba4f9a7ebf59a1dca3da9d8c94478f670aea28bd0e1  quadlet/postgres.container
 04a125068677bbee6e71d313ca9d95d35faa32851b79de781cd7b9f466561445  quadlet/postgres.env.example
-45f398c0e494c86efb92bfa9381617c7dbb4b929c9efb5ff86473d2dc6fe729e  tests/run-all.sh
+0fb32ab5a9a51044ae9e4ae75b7d70bfb7f5e640c41a3d0de02277c9a8c44c91  tests/run-all.sh
 5e5d4fdd2b6c7e0b8b146de7aad06c1f96b221a10fec525b00d40e0d85709fb0  tests/test-entrypoint-structure.sh
+b83f0415c09f0be34c51722370e5b07c51b9a9756d305d3b3f50827f484aa4dd  tests/test-proxy-client-actions.py
 c3b988d7b2b1b138bb2530df9172f96bb0f5edc2f1561f38f0055695f4126134  tests/test-proxy-protocol.sh

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

@@ -51,6 +51,7 @@ mc_proxy_monitor()
             "${MC_PROXY_CLIENT:-${MC_PROXY_CLIENT_DEFAULT}}" \
             "${socket_path}" \
             "${resource}" \
+            "monitor" \
             "${MC_PROXY_RC_TRANSPORT}" \
             "${MC_PROXY_RC_HARD_FAILURE}"
     )"

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

@@ -9,18 +9,18 @@ PROTOCOL_VERSION = 1
 VALID_OCF_RCS = {0, 1, 7, 8}
 
 
-def main():
+class TransportFailure(Exception):
-    if len(sys.argv) != 5:
+    pass
-        raise SystemExit(11)
 
-    socket_path = sys.argv[1]
-    resource = sys.argv[2]
-    transport_failure = int(sys.argv[3])
-    protocol_failure = int(sys.argv[4])
 
+class ProtocolFailure(Exception):
+    pass
+
+
+def send_request(socket_path, resource, action):
     request = {
         "protocol_version": PROTOCOL_VERSION,
-        "action": "monitor",
+        "action": action,
         "resource": resource,
     }
 
@@ -35,7 +35,10 @@ def main():
             sock.connect(socket_path)
 
             sock.sendall(
-                json.dumps(request).encode("utf-8")
+                json.dumps(
+                    request,
+                    separators=(",", ":"),
+                ).encode("utf-8")
                 + b"\n"
             )
 
@@ -56,49 +59,60 @@ def main():
         socket.timeout,
         ConnectionError,
         OSError,
-    ):
+    ) as exc:
-        raise SystemExit(transport_failure)
+        raise TransportFailure from exc
 
-    # A connection was established. From this point onward, absence of a
+    return raw
-    # valid response is a protocol/result failure rather than transport loss.
+
+
+def decode_response(raw):
     if not raw:
-        raise SystemExit(protocol_failure)
+        raise ProtocolFailure
 
     try:
-        obj = json.loads(
+        response = json.loads(
             raw.decode("utf-8").strip()
         )
 
     except (
         UnicodeDecodeError,
         json.JSONDecodeError,
-    ):
+    ) as exc:
-        raise SystemExit(protocol_failure)
+        raise ProtocolFailure from exc
 
-    if not isinstance(obj, dict):
+    if not isinstance(response, dict):
-        raise SystemExit(protocol_failure)
+        raise ProtocolFailure
 
-    if obj.get("protocol_version") != PROTOCOL_VERSION:
+    return response
-        raise SystemExit(protocol_failure)
 
-    if obj.get("ok") is not True:
-        raise SystemExit(protocol_failure)
 
-    if obj.get("resource") != resource:
+def validate_response(response, resource, action):
-        raise SystemExit(protocol_failure)
+    if response.get("protocol_version") != PROTOCOL_VERSION:
+        raise ProtocolFailure
 
-    if obj.get("action") != "monitor":
+    if response.get("ok") is not True:
-        raise SystemExit(protocol_failure)
+        raise ProtocolFailure
 
-    result = obj.get("result")
+    if response.get("resource") != resource:
+        raise ProtocolFailure
 
+    if response.get("action") != action:
+        raise ProtocolFailure
+
+    if "result" not in response:
+        raise ProtocolFailure
+
+    return response["result"]
+
+
+def format_monitor_result(result):
     if not isinstance(result, dict):
-        raise SystemExit(protocol_failure)
+        raise ProtocolFailure
 
     ha = result.get("ha")
 
     if not isinstance(ha, dict):
-        raise SystemExit(protocol_failure)
+        raise ProtocolFailure
 
     try:
         ocf_rc = int(ha.get("ocf_rc"))
@@ -106,18 +120,18 @@ def main():
     except (
         TypeError,
         ValueError,
-    ):
+    ) as exc:
-        raise SystemExit(protocol_failure)
+        raise ProtocolFailure from exc
 
     if ocf_rc not in VALID_OCF_RCS:
-        raise SystemExit(protocol_failure)
+        raise ProtocolFailure
 
     receiver = result.get("wal_receiver") or {}
 
     if not isinstance(receiver, dict):
-        raise SystemExit(protocol_failure)
+        raise ProtocolFailure
 
-    payload = "|".join([
+    return "|".join([
         str(ocf_rc),
         str(ha.get("health", "unknown")),
         str(ha.get("ha_role", "unknown")),
@@ -139,7 +153,62 @@ def main():
         str(receiver.get("sender_host") or ""),
     ])
 
-    print(payload)
+
+def format_result(action, result):
+    if action == "monitor":
+        return format_monitor_result(result)
+
+    return json.dumps(
+        result,
+        separators=(",", ":"),
+        sort_keys=True,
+    )
+
+
+def main():
+    if len(sys.argv) != 6:
+        raise SystemExit(11)
+
+    socket_path = sys.argv[1]
+    resource = sys.argv[2]
+    action = sys.argv[3]
+
+    try:
+        transport_failure = int(sys.argv[4])
+        protocol_failure = int(sys.argv[5])
+
+    except ValueError:
+        raise SystemExit(11)
+
+    if not resource or not action:
+        raise SystemExit(protocol_failure)
+
+    try:
+        raw = send_request(
+            socket_path,
+            resource,
+            action,
+        )
+
+    except TransportFailure:
+        raise SystemExit(transport_failure)
+
+    try:
+        response = decode_response(raw)
+        result = validate_response(
+            response,
+            resource,
+            action,
+        )
+        output = format_result(
+            action,
+            result,
+        )
+
+    except ProtocolFailure:
+        raise SystemExit(protocol_failure)
+
+    print(output)
 
 
 if __name__ == "__main__":

tests/run-all.sh

@@ -20,6 +20,10 @@ echo
 echo ">>> Proxy protocol contract"
 ./tests/test-proxy-protocol.sh
 
+echo
+echo ">>> Generic proxy client actions"
+python3 ./tests/test-proxy-client-actions.py
+
 echo
 echo ">>> Python syntax"
 

tests/test-proxy-client-actions.py

@@ -0,0 +1,186 @@
+#!/usr/bin/env python3
+
+import json
+from pathlib import Path
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+
+
+ROOT = Path(__file__).resolve().parents[1]
+CLIENT = (
+    ROOT
+    / "ocf"
+    / "loxthelion"
+    / "managed-container.d"
+    / "proxy_client.py"
+)
+
+
+def require(condition, message):
+    if not condition:
+        raise SystemExit("FAIL: " + message)
+
+
+def run_exchange(response, action="promote"):
+    with tempfile.TemporaryDirectory() as temp_dir:
+        socket_path = str(
+            Path(temp_dir) / "managed-containerd.sock"
+        )
+
+        listener = socket.socket(
+            socket.AF_UNIX,
+            socket.SOCK_STREAM,
+        )
+        listener.bind(socket_path)
+        listener.listen(1)
+
+        captured = {}
+        errors = []
+
+        def serve_once():
+            try:
+                conn, _ = listener.accept()
+
+                with conn:
+                    raw = b""
+
+                    while b"\n" not in raw:
+                        chunk = conn.recv(65536)
+
+                        if not chunk:
+                            break
+
+                        raw += chunk
+
+                    captured["request"] = json.loads(
+                        raw.decode("utf-8").strip()
+                    )
+
+                    conn.sendall(
+                        json.dumps(response).encode("utf-8")
+                        + b"\n"
+                    )
+
+            except Exception as exc:
+                errors.append(exc)
+
+            finally:
+                listener.close()
+
+        thread = threading.Thread(
+            target=serve_once,
+            daemon=True,
+        )
+        thread.start()
+
+        completed = subprocess.run(
+            [
+                sys.executable,
+                str(CLIENT),
+                socket_path,
+                "postgresql-ha",
+                action,
+                "10",
+                "11",
+            ],
+            text=True,
+            capture_output=True,
+            check=False,
+        )
+
+        thread.join(timeout=5)
+
+        require(
+            not thread.is_alive(),
+            "test server did not finish",
+        )
+        require(
+            not errors,
+            f"test server failed: {errors}",
+        )
+
+        return completed, captured["request"]
+
+
+print("===== GENERIC PROXY CLIENT ACTIONS =====")
+
+valid_response = {
+    "protocol_version": 1,
+    "ok": True,
+    "resource": "postgresql-ha",
+    "action": "promote",
+    "result": {
+        "state": "promoted",
+        "changed": True,
+    },
+}
+
+completed, request = run_exchange(
+    valid_response,
+    action="promote",
+)
+
+require(completed.returncode == 0, "generic action failed")
+require(
+    completed.stdout.strip()
+    == '{"changed":true,"state":"promoted"}',
+    "generic result was not emitted as canonical JSON",
+)
+require(
+    request == {
+        "protocol_version": 1,
+        "action": "promote",
+        "resource": "postgresql-ha",
+    },
+    "generic request envelope is incorrect",
+)
+
+print("generic_action         expected=0  actual=0")
+
+wrong_action_response = dict(valid_response)
+wrong_action_response["action"] = "demote"
+
+completed, _ = run_exchange(
+    wrong_action_response,
+    action="promote",
+)
+
+require(
+    completed.returncode == 11,
+    "mismatched action was not a protocol failure",
+)
+
+print("wrong_action_echo      expected=11 actual=11")
+
+with tempfile.TemporaryDirectory() as temp_dir:
+    missing_socket = str(
+        Path(temp_dir) / "missing.sock"
+    )
+
+    completed = subprocess.run(
+        [
+            sys.executable,
+            str(CLIENT),
+            missing_socket,
+            "postgresql-ha",
+            "promote",
+            "10",
+            "11",
+        ],
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+
+require(
+    completed.returncode == 10,
+    "missing socket was not a transport failure",
+)
+
+print("generic_transport      expected=10 actual=10")
+
+print()
+print("PROXY_CLIENT_ACTION_TEST_PASS=1")