73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Container health probe for web, database, OpenRGB process, and SDK."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import socket
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
|
|
def main() -> int:
|
|
port = int(os.environ.get("APP_PORT", "8080"))
|
|
try:
|
|
with urllib.request.urlopen(f"http://127.0.0.1:{port}/health/live", timeout=2) as response:
|
|
if response.status != 200:
|
|
return fail("web liveness failed")
|
|
with urllib.request.urlopen(f"http://127.0.0.1:{port}/health/ready", timeout=2) as response:
|
|
if response.status != 200:
|
|
return fail("backend readiness failed")
|
|
with urllib.request.urlopen(f"http://127.0.0.1:{port}/health/container", timeout=3) as response:
|
|
report = json.load(response)
|
|
except (urllib.error.URLError, TimeoutError, ValueError, json.JSONDecodeError) as exc:
|
|
return fail(f"health API failed: {exc}")
|
|
|
|
pid_file = Path("/run/lumaops/openrgb.pid")
|
|
if not pid_file.is_file():
|
|
return fail("OpenRGB PID file missing")
|
|
try:
|
|
pid = int(pid_file.read_text(encoding="ascii"))
|
|
process_status = Path(f"/proc/{pid}/status").read_text(encoding="ascii")
|
|
process_name = next(
|
|
line.split(":", 1)[1].strip()
|
|
for line in process_status.splitlines()
|
|
if line.startswith("Name:")
|
|
)
|
|
process_state = next(
|
|
line.split(":", 1)[1].strip()
|
|
for line in process_status.splitlines()
|
|
if line.startswith("State:")
|
|
)
|
|
if process_name != "openrgb" or process_state.startswith("Z"):
|
|
raise ValueError("stale OpenRGB PID")
|
|
except (OSError, StopIteration, ValueError):
|
|
return fail("OpenRGB process is not running")
|
|
|
|
try:
|
|
with socket.create_connection(
|
|
("127.0.0.1", int(os.environ.get("OPENRGB_PORT", "6742"))), timeout=2
|
|
):
|
|
pass
|
|
except OSError as exc:
|
|
return fail(f"OpenRGB loopback socket failed: {exc}")
|
|
|
|
connector = report.get("components", {}).get("openrgb-local", {})
|
|
if os.environ.get("HEALTHCHECK_REQUIRE_OPENRGB", "true").lower() == "true":
|
|
if not connector.get("connected"):
|
|
return fail("backend SDK connection is degraded")
|
|
print(json.dumps({"status": report.get("status"), "openrgb": "connected"}))
|
|
return 0
|
|
|
|
|
|
def fail(message: str) -> int:
|
|
print(message, file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|