453 lines
11 KiB
Python
453 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import json
|
|
import os
|
|
import signal
|
|
import socket
|
|
import struct
|
|
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
|
|
pacemaker_resource_id = None
|
|
pacemaker_peer_uid = None
|
|
pacemaker_peer_gid = None
|
|
pacemaker_node_name = None
|
|
|
|
PROTOCOL_VERSION = 1
|
|
|
|
|
|
def protocol_response(payload):
|
|
"""
|
|
Return a daemon response carrying the wire-protocol version.
|
|
|
|
protocol_version is daemon-owned metadata and cannot be overridden by
|
|
registry or plugin output.
|
|
"""
|
|
response = dict(payload)
|
|
response["protocol_version"] = PROTOCOL_VERSION
|
|
return response
|
|
|
|
|
|
def load_pacemaker_resource_binding(
|
|
current_registry,
|
|
environ=None,
|
|
):
|
|
"""
|
|
Load and validate the Pacemaker resource identity accepted by this
|
|
daemon instance.
|
|
|
|
RESOURCE_ID identifies the registry entry. PACEMAKER_RESOURCE_ID is a
|
|
separate control-plane binding and must resolve to that registry.
|
|
"""
|
|
if environ is None:
|
|
environ = os.environ
|
|
|
|
resource_id = environ.get(
|
|
"PACEMAKER_RESOURCE_ID"
|
|
)
|
|
|
|
if not resource_id:
|
|
raise RegistryError(
|
|
"PACEMAKER_RESOURCE_ID is not configured"
|
|
)
|
|
|
|
if resource_id not in current_registry.resources:
|
|
raise RegistryError(
|
|
"PACEMAKER_RESOURCE_ID does not identify "
|
|
f"a registered resource: {resource_id}"
|
|
)
|
|
|
|
return resource_id
|
|
|
|
|
|
def load_pacemaker_node_name(environ=None):
|
|
"""
|
|
Load the Pacemaker node represented by this daemon instance.
|
|
|
|
The daemon is local to one PostgreSQL instance. This identity is not
|
|
used for routing; it is used to validate promotion and demotion context.
|
|
"""
|
|
if environ is None:
|
|
environ = os.environ
|
|
|
|
raw_node_name = environ.get("PACEMAKER_NODE_NAME")
|
|
|
|
if raw_node_name is None:
|
|
raise RegistryError(
|
|
"PACEMAKER_NODE_NAME is not configured"
|
|
)
|
|
|
|
node_name = raw_node_name.strip().lower()
|
|
|
|
if not node_name:
|
|
raise RegistryError(
|
|
"PACEMAKER_NODE_NAME is empty"
|
|
)
|
|
|
|
if any(char.isspace() for char in node_name):
|
|
raise RegistryError(
|
|
"PACEMAKER_NODE_NAME must not contain whitespace"
|
|
)
|
|
|
|
return node_name
|
|
|
|
|
|
def load_pacemaker_peer_authorization(
|
|
environ=None,
|
|
):
|
|
"""
|
|
Load the peer credentials that authorize Pacemaker resource actions.
|
|
|
|
In the rootless keep-id user namespace, host root is intentionally
|
|
unmapped and therefore appears as the configured overflow UID/GID.
|
|
Socket path permissions remain part of this authorization boundary.
|
|
"""
|
|
if environ is None:
|
|
environ = os.environ
|
|
|
|
raw_uid = environ.get("PACEMAKER_PEER_UID")
|
|
raw_gid = environ.get("PACEMAKER_PEER_GID")
|
|
|
|
if raw_uid is None or raw_gid is None:
|
|
raise RegistryError(
|
|
"PACEMAKER_PEER_UID and "
|
|
"PACEMAKER_PEER_GID must both be configured"
|
|
)
|
|
|
|
try:
|
|
peer_uid = int(raw_uid)
|
|
peer_gid = int(raw_gid)
|
|
|
|
except ValueError as exc:
|
|
raise RegistryError(
|
|
"PACEMAKER_PEER_UID and "
|
|
"PACEMAKER_PEER_GID must be integers"
|
|
) from exc
|
|
|
|
if peer_uid < 0 or peer_gid < 0:
|
|
raise RegistryError(
|
|
"PACEMAKER peer credentials must be non-negative"
|
|
)
|
|
|
|
return peer_uid, peer_gid
|
|
|
|
|
|
def get_peer_credentials(conn):
|
|
raw = conn.getsockopt(
|
|
socket.SOL_SOCKET,
|
|
socket.SO_PEERCRED,
|
|
struct.calcsize("3i"),
|
|
)
|
|
|
|
pid, uid, gid = struct.unpack("3i", raw)
|
|
|
|
return {
|
|
"pid": pid,
|
|
"uid": uid,
|
|
"gid": gid,
|
|
}
|
|
|
|
|
|
def handle_request(
|
|
req,
|
|
peer_credentials=None,
|
|
):
|
|
if not isinstance(req, dict):
|
|
return protocol_response({
|
|
"ok": False,
|
|
"error_code": "invalid_request",
|
|
"error": "request must be a JSON object",
|
|
})
|
|
|
|
requested_version = req.get("protocol_version")
|
|
|
|
# Backward compatibility for the pre-versioned proxy:
|
|
# absence is accepted during the rolling upgrade, but an explicitly
|
|
# incompatible version is rejected.
|
|
if (
|
|
requested_version is not None
|
|
and requested_version != PROTOCOL_VERSION
|
|
):
|
|
return protocol_response({
|
|
"ok": False,
|
|
"error_code": "unsupported_protocol_version",
|
|
"error": (
|
|
"unsupported protocol version: "
|
|
f"{requested_version}; "
|
|
f"supported={PROTOCOL_VERSION}"
|
|
),
|
|
})
|
|
|
|
action = req.get("action")
|
|
|
|
if action == "ping":
|
|
return protocol_response({
|
|
"ok": True,
|
|
"daemon": "managed-containerd",
|
|
"pacemaker_resource": (
|
|
pacemaker_resource_id
|
|
),
|
|
"pacemaker_node": pacemaker_node_name,
|
|
})
|
|
|
|
if action == "registry.list":
|
|
return protocol_response({
|
|
"ok": True,
|
|
"pacemaker_resource": (
|
|
pacemaker_resource_id
|
|
),
|
|
"pacemaker_node": pacemaker_node_name,
|
|
"resources": registry.list_resources(),
|
|
})
|
|
|
|
request_parameters = req.get(
|
|
"parameters",
|
|
{},
|
|
)
|
|
|
|
if not isinstance(request_parameters, dict):
|
|
return protocol_response({
|
|
"ok": False,
|
|
"error_code": "invalid_request",
|
|
"error": "parameters must be a JSON object",
|
|
"resource": req.get("resource"),
|
|
"action": action,
|
|
})
|
|
|
|
resource = req.get("resource")
|
|
|
|
if not resource:
|
|
return protocol_response({
|
|
"ok": False,
|
|
"error_code": "invalid_request",
|
|
"error": "resource is required",
|
|
})
|
|
|
|
if resource != pacemaker_resource_id:
|
|
return protocol_response({
|
|
"ok": False,
|
|
"error_code": "resource_binding_mismatch",
|
|
"error": (
|
|
"requested resource does not match "
|
|
"the configured Pacemaker binding"
|
|
),
|
|
"resource": resource,
|
|
"action": action,
|
|
})
|
|
|
|
if peer_credentials is None:
|
|
return protocol_response({
|
|
"ok": False,
|
|
"error_code": "peer_credentials_unavailable",
|
|
"error": (
|
|
"peer credentials are required for "
|
|
"Pacemaker resource actions"
|
|
),
|
|
"resource": resource,
|
|
"action": action,
|
|
})
|
|
|
|
if (
|
|
peer_credentials.get("uid")
|
|
!= pacemaker_peer_uid
|
|
or peer_credentials.get("gid")
|
|
!= pacemaker_peer_gid
|
|
):
|
|
return protocol_response({
|
|
"ok": False,
|
|
"error_code": "peer_credential_mismatch",
|
|
"error": (
|
|
"peer credentials are not authorized "
|
|
"for Pacemaker resource actions"
|
|
),
|
|
"resource": resource,
|
|
"action": action,
|
|
})
|
|
|
|
if request_parameters:
|
|
return protocol_response(
|
|
registry.execute(
|
|
resource,
|
|
action,
|
|
request_parameters,
|
|
)
|
|
)
|
|
|
|
return protocol_response(
|
|
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
|
|
global pacemaker_resource_id
|
|
global pacemaker_peer_uid
|
|
global pacemaker_peer_gid
|
|
global pacemaker_node_name
|
|
|
|
registry = Registry()
|
|
pacemaker_resource_id = (
|
|
load_pacemaker_resource_binding(registry)
|
|
)
|
|
(
|
|
pacemaker_peer_uid,
|
|
pacemaker_peer_gid,
|
|
) = load_pacemaker_peer_authorization()
|
|
pacemaker_node_name = load_pacemaker_node_name()
|
|
|
|
socket_directory = os.path.dirname(
|
|
SOCKET_PATH
|
|
)
|
|
|
|
os.makedirs(
|
|
socket_directory,
|
|
exist_ok=True,
|
|
)
|
|
os.chmod(socket_directory, 0o700)
|
|
|
|
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, 0o600)
|
|
server_socket.listen(16)
|
|
|
|
print(
|
|
f"managed-containerd listening on {SOCKET_PATH}",
|
|
flush=True,
|
|
)
|
|
|
|
print(
|
|
"registered resources: "
|
|
+ ", ".join(registry.resources),
|
|
flush=True,
|
|
)
|
|
|
|
print(
|
|
"Pacemaker resource binding: "
|
|
+ pacemaker_resource_id,
|
|
flush=True,
|
|
)
|
|
|
|
print(
|
|
"Pacemaker peer authorization: "
|
|
f"uid={pacemaker_peer_uid} "
|
|
f"gid={pacemaker_peer_gid}",
|
|
flush=True,
|
|
)
|
|
|
|
print(
|
|
"Pacemaker node identity: "
|
|
+ pacemaker_node_name,
|
|
flush=True,
|
|
)
|
|
|
|
try:
|
|
while True:
|
|
conn, _ = server_socket.accept()
|
|
|
|
with conn:
|
|
try:
|
|
peer_credentials = (
|
|
get_peer_credentials(conn)
|
|
)
|
|
|
|
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,
|
|
peer_credentials,
|
|
)
|
|
|
|
except (
|
|
RegistryError,
|
|
ValueError,
|
|
KeyError,
|
|
) as exc:
|
|
response = protocol_response({
|
|
"ok": False,
|
|
"error_code": "request_error",
|
|
"error": str(exc),
|
|
})
|
|
|
|
except Exception as exc:
|
|
traceback.print_exc()
|
|
|
|
response = protocol_response({
|
|
"ok": False,
|
|
"error_code": "internal_error",
|
|
"error": str(exc),
|
|
})
|
|
|
|
conn.sendall(
|
|
json.dumps(
|
|
response,
|
|
default=str,
|
|
).encode("utf-8")
|
|
+ b"\n"
|
|
)
|
|
|
|
finally:
|
|
cleanup_socket()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
serve()
|