pigway-console-adapter-frame-flow/plugin/worker.py

#!/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()