#!/usr/bin/env python3 """Read-only Frame Flow monitoring via Console adapter protocol v1.""" import concurrent.futures import json import sys import threading import time from pathlib import Path from urllib.error import HTTPError from urllib.parse import urlparse from urllib.request import HTTPRedirectHandler, Request, build_opener MANIFEST = json.loads(Path(__file__).with_name("manifest.json").read_text()) OUTPUT_LOCK = threading.Lock() DEVICE_LOCKS = {} LOCK = threading.Lock() MAX_MESSAGE = 16 * 1024 * 1024 class AdapterError(Exception): def __init__(self, code): self.code = code class NoRedirect(HTTPRedirectHandler): def redirect_request(self, *args, **kwargs): return None def request(device, endpoint): if endpoint not in {"/health", "/ready", "/api/v1/devices", "/api/v1/jobs"}: raise AdapterError("UNSUPPORTED") origin = device.get("connection", {}).get("url", "").rstrip("/") parsed = urlparse(origin) if ( parsed.scheme not in ("http", "https") or not parsed.hostname or parsed.username or parsed.password or parsed.path or parsed.query or parsed.fragment ): raise AdapterError("INVALID_CONFIG") token = device.get("credentials", {}).get("token") if not isinstance(token, str) or not token or any(c in token for c in "\r\n"): raise AdapterError("INVALID_CONFIG") headers = {"Accept": "application/json"} if endpoint.startswith("/api/"): headers["Authorization"] = "Bearer " + token try: req = Request(origin + endpoint, headers=headers, method="GET") try: response = build_opener(NoRedirect).open(req, timeout=3) except HTTPError as exc: if endpoint == "/ready" and exc.code == 503: response = exc else: raise AdapterError( "AUTH_FAILED" if exc.code in (401, 403) else "DEVICE_API_ERROR" ) from None with response: raw = response.read(2 * 1024 * 1024 + 1) if len(raw) > 2 * 1024 * 1024: raise AdapterError("INVALID_RESPONSE") try: return json.loads(raw) except ValueError: raise AdapterError("INVALID_RESPONSE") from None except (OSError, ValueError): raise AdapterError("DEVICE_UNAVAILABLE") from None def collect(device): health = request(device, "/health") ready = request(device, "/ready") devices = request(device, "/api/v1/devices") jobs = request(device, "/api/v1/jobs") if ( not isinstance(health, dict) or health.get("status") != "ok" or not isinstance(ready, dict) or ready.get("status") not in ("ready", "not_ready") ): raise AdapterError("INVALID_RESPONSE") resources = [{"id": "service", "type": "service", "name": "Frame Flow"}] metrics = [] alerts = [] def metric(rid, key, value, unit="boolean"): metrics.append({"resource_id": rid, "key": key, "value": value, "unit": unit}) def alert(identifier): alerts.append({"id": identifier, "severity": 1}) metric("service", "service_ready", int(ready["status"] == "ready")) if ready["status"] != "ready": alert("SERVICE_NOT_READY") for items, kind in ((devices, "device"), (jobs, "job")): if not isinstance(items, list) or len(items) > 2000: raise AdapterError("INVALID_RESPONSE") seen = set() for item in items: if ( not isinstance(item, dict) or not isinstance(item.get("id"), str) or not item["id"] or item["id"] in seen ): raise AdapterError("INVALID_RESPONSE") seen.add(item["id"]) rid = kind + ":" + item["id"] resources.append( { "id": rid, "parent_id": "service", "type": kind, "name": item.get("name") or item["id"], } ) if kind == "device": if item.get("status") not in ("online", "offline"): raise AdapterError("INVALID_RESPONSE") metric(rid, "device_online", int(item["status"] == "online")) if item["status"] == "offline": alert("DEVICE_OFFLINE:" + item["id"]) power = item.get("power") or {} battery = power.get("battery_percent") if type(battery) is int and 0 <= battery <= 100: metric(rid, "battery_percent", battery, "percent") else: if type(item.get("enabled")) is not bool: raise AdapterError("INVALID_RESPONSE") metric(rid, "job_enabled", int(item["enabled"])) result = item.get("last_result") if result in ("COMPLETED", "FAILED", "UNKNOWN"): metric(rid, "last_run_success", int(result == "COMPLETED")) if result in ("FAILED", "UNKNOWN"): alert("JOB_" + result + ":" + item["id"]) metric("service", kind + "_count", len(items), "count") return { "schema_version": 1, "observed_at": time.time(), "identity": {"reported_name": "Frame Flow", "category": "host"}, "collection": MANIFEST["collection"], "capabilities": ["status"], "resources": resources, "metrics": metrics, "alerts": alerts, "alerts_complete": True, } def dispatch(method, params): if method in ("adapter.initialize", "adapter.describe", "adapter.health"): return MANIFEST if method not in ("device.validate", "device.collect"): raise AdapterError("UNSUPPORTED") device = params["device"] with LOCK: device_lock = DEVICE_LOCKS.setdefault(device["id"], threading.Lock()) with device_lock: snapshot = collect(device) return {"valid": True} if method == "device.validate" else snapshot def handle(message): try: result = dispatch(message["method"], message.get("params", {})) reply = {"jsonrpc": "2.0", "id": message["id"], "result": result} except AdapterError as exc: reply = { "jsonrpc": "2.0", "id": message["id"], "error": {"code": -32000, "data": exc.code}, } except Exception: # Do not copy vendor exception messages: they may contain credentials or URLs. reply = { "jsonrpc": "2.0", "id": message["id"], "error": {"code": -32000, "data": "DEVICE_UNAVAILABLE"}, } raw = json.dumps(reply, allow_nan=False, separators=(",", ":")) with OUTPUT_LOCK: print(raw, flush=True) def main(): # Bound both running and queued calls even if the parent violates its limits. slots = threading.BoundedSemaphore(8) with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: while True: line = sys.stdin.buffer.readline(MAX_MESSAGE + 1) if not line: break if len(line) > MAX_MESSAGE or not line.endswith(b"\n"): break message = json.loads(line) slots.acquire() task = pool.submit(handle, message) task.add_done_callback(lambda _: slots.release()) if __name__ == "__main__": main()