235 lines
8.5 KiB
Python
235 lines
8.5 KiB
Python
#!/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
|
|
METRIC_LABELS = {
|
|
"service_ready": {"zh": "服务就绪", "en": "Service ready"},
|
|
"device_online": {"zh": "画框在线", "en": "Frame online"},
|
|
"battery_percent": {"zh": "电量", "en": "Battery"},
|
|
"job_enabled": {"zh": "任务启用", "en": "Job enabled"},
|
|
"last_run_success": {"zh": "最近运行成功", "en": "Last run succeeded"},
|
|
"device_count": {"zh": "画框数量", "en": "Frame count"},
|
|
"job_count": {"zh": "任务数量", "en": "Job count"},
|
|
}
|
|
|
|
|
|
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,
|
|
"label": METRIC_LABELS[key],
|
|
}
|
|
)
|
|
|
|
def alert(identifier):
|
|
kind = identifier.split(":", 1)[0]
|
|
labels = {
|
|
"SERVICE_NOT_READY": {"zh": "服务未就绪", "en": "Service not ready"},
|
|
"DEVICE_OFFLINE": {"zh": "画框离线", "en": "Frame offline"},
|
|
"JOB_FAILED": {"zh": "任务最近运行失败", "en": "Last job run failed"},
|
|
"JOB_UNKNOWN": {
|
|
"zh": "任务最近结果待确认",
|
|
"en": "Last job result unconfirmed",
|
|
},
|
|
}
|
|
alerts.append({"id": identifier, "severity": 1, "label": labels[kind]})
|
|
|
|
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()
|