feat(awooop): show controlled cd live proof from prometheus

apps/api/src/services/awoooi_priority_work_order_readback.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 import json
 import re
 import urllib.error
+import urllib.parse
 import urllib.request
 from datetime import datetime
 from pathlib import Path
@@ -25,6 +26,19 @@ _CONTROLLED_CD_LANE_LIVE_METRIC_SCHEMA_VERSION = (
     "controlled_cd_lane_live_metric_readback_v1"
 )
 _CONTROLLED_CD_LANE_NODE_EXPORTER_URL = "http://192.168.0.110:9100/metrics"
+_CONTROLLED_CD_LANE_PROMETHEUS_QUERY_URL = "http://192.168.0.110:9090/api/v1/query"
+_CONTROLLED_CD_LANE_ALLOWED_METRIC_NAMES = {
+    "awoooi_host_gitea_actions_active_container_count",
+    "awoooi_host_gitea_actions_active_process_count",
+    "awoooi_host_load5_per_core",
+    "awoooi_runner_failclosed_enforcer_active_job_containers",
+    "awoooi_runner_failclosed_enforcer_apply_performed",
+    "awoooi_runner_failclosed_enforcer_controlled_drain_active_allowed",
+    "awoooi_runner_failclosed_enforcer_controlled_drain_staging_allowed",
+    "awoooi_runner_failclosed_enforcer_lane_process_count",
+    "awoooi_runner_failclosed_enforcer_last_run_timestamp",
+    "awoooi_runner_failclosed_enforcer_root_restore_sources_left",
+}
 _SHA_RE = re.compile(r"^[0-9a-f]{40}$")
 _METRIC_LINE_RE = re.compile(
     r"^(?P<name>[a-zA-Z_:][a-zA-Z0-9_:]*)(?:\{[^}]*\})?\s+"
@@ -53,6 +67,7 @@ def load_latest_awoooi_priority_work_order_readback(
 
 def load_controlled_cd_lane_live_metric_readback(
     metrics_url: str = _CONTROLLED_CD_LANE_NODE_EXPORTER_URL,
+    prometheus_query_url: str = _CONTROLLED_CD_LANE_PROMETHEUS_QUERY_URL,
     timeout_seconds: float = 3.0,
 ) -> dict[str, Any]:
     """Read 110 node_exporter text metrics for the controlled CD lane.
@@ -60,28 +75,38 @@ def load_controlled_cd_lane_live_metric_readback(
     This is a read-only HTTP metrics probe. It does not SSH, read runner
     registration contents, touch Docker/systemd, or perform runtime writes.
     """
+    source = "node_exporter_text_metrics"
+    error_class = ""
     try:
         request = urllib.request.Request(metrics_url, method="GET")
         with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
             raw = response.read(2_000_000).decode("utf-8", errors="replace")
+        metrics = _parse_prometheus_text_metrics(raw)
     except (OSError, TimeoutError, urllib.error.URLError) as exc:
-        return {
-            "schema_version": _CONTROLLED_CD_LANE_LIVE_METRIC_SCHEMA_VERSION,
-            "status": "controlled_cd_lane_live_metric_unavailable",
-            "source": "node_exporter_text_metrics",
-            "source_url": metrics_url,
-            "metadata_only": True,
-            "active_blockers": ["controlled_cd_lane_live_metric_unavailable"],
-            "error_class": exc.__class__.__name__,
-            "readback": {
-                "enforcer_metric_present": False,
-                "controlled_drain_active_allowed": False,
-                "controlled_drain_staging_allowed": False,
-                "enforcer_last_run_fresh": False,
-            },
-        }
+        metrics, fallback_error_class = _load_prometheus_query_metrics(
+            prometheus_query_url,
+            timeout_seconds=timeout_seconds,
+        )
+        if fallback_error_class:
+            return {
+                "schema_version": _CONTROLLED_CD_LANE_LIVE_METRIC_SCHEMA_VERSION,
+                "status": "controlled_cd_lane_live_metric_unavailable",
+                "source": "node_exporter_text_metrics",
+                "fallback_source": "prometheus_query_api",
+                "metadata_only": True,
+                "active_blockers": ["controlled_cd_lane_live_metric_unavailable"],
+                "error_class": exc.__class__.__name__,
+                "fallback_error_class": fallback_error_class,
+                "readback": {
+                    "enforcer_metric_present": False,
+                    "controlled_drain_active_allowed": False,
+                    "controlled_drain_staging_allowed": False,
+                    "enforcer_last_run_fresh": False,
+                },
+            }
+        source = "prometheus_query_api"
+        error_class = exc.__class__.__name__
 
-    metrics = _parse_prometheus_text_metrics(raw)
     readback = _build_controlled_cd_lane_metric_readback(metrics)
     active_blockers: list[str] = []
     if not readback["enforcer_metric_present"]:
@@ -100,9 +125,9 @@ def load_controlled_cd_lane_live_metric_readback(
             if readback["enforcer_metric_present"]
             else "controlled_cd_lane_live_metric_missing"
         ),
-        "source": "node_exporter_text_metrics",
-        "source_url": metrics_url,
+        "source": source,
         "metadata_only": True,
+        "primary_error_class": error_class or None,
         "active_blockers": active_blockers,
         "readback": readback,
     }
@@ -2290,18 +2315,6 @@ def _require_mainline_consistency(payload: dict[str, Any], label: str) -> None:
 
 def _parse_prometheus_text_metrics(raw: str) -> dict[str, float]:
     metrics: dict[str, float] = {}
-    allowed_names = {
-        "awoooi_host_gitea_actions_active_container_count",
-        "awoooi_host_gitea_actions_active_process_count",
-        "awoooi_host_load5_per_core",
-        "awoooi_runner_failclosed_enforcer_active_job_containers",
-        "awoooi_runner_failclosed_enforcer_apply_performed",
-        "awoooi_runner_failclosed_enforcer_controlled_drain_active_allowed",
-        "awoooi_runner_failclosed_enforcer_controlled_drain_staging_allowed",
-        "awoooi_runner_failclosed_enforcer_lane_process_count",
-        "awoooi_runner_failclosed_enforcer_last_run_timestamp",
-        "awoooi_runner_failclosed_enforcer_root_restore_sources_left",
-    }
     for line in raw.splitlines():
         if not line or line.startswith("#"):
             continue
@@ -2309,7 +2322,7 @@ def _parse_prometheus_text_metrics(raw: str) -> dict[str, float]:
         if not match:
             continue
         name = match.group("name")
-        if name not in allowed_names:
+        if name not in _CONTROLLED_CD_LANE_ALLOWED_METRIC_NAMES:
             continue
         try:
             metrics[name] = float(match.group("value"))
@@ -2318,6 +2331,46 @@ def _parse_prometheus_text_metrics(raw: str) -> dict[str, float]:
     return metrics
 
 
+def _load_prometheus_query_metrics(
+    prometheus_query_url: str,
+    *,
+    timeout_seconds: float,
+) -> tuple[dict[str, float], str]:
+    metrics: dict[str, float] = {}
+    try:
+        for name in sorted(_CONTROLLED_CD_LANE_ALLOWED_METRIC_NAMES):
+            query_url = (
+                f"{prometheus_query_url}?"
+                f"{urllib.parse.urlencode({'query': name})}"
+            )
+            request = urllib.request.Request(query_url, method="GET")
+            with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
+                payload = json.loads(
+                    response.read(1_000_000).decode("utf-8", errors="replace")
+                )
+            if not isinstance(payload, dict) or payload.get("status") != "success":
+                return {}, "PrometheusQueryNotSuccessful"
+            results = _list(_dict(payload.get("data")).get("result"))
+            if not results:
+                continue
+            value_pair = _list(_dict(results[0]).get("value"))
+            if len(value_pair) < 2:
+                continue
+            try:
+                metrics[name] = float(value_pair[1])
+            except (TypeError, ValueError):
+                continue
+    except (
+        OSError,
+        TimeoutError,
+        urllib.error.URLError,
+        json.JSONDecodeError,
+        ValueError,
+    ) as exc:
+        return {}, exc.__class__.__name__
+    return metrics, ""
+
+
 def _build_controlled_cd_lane_metric_readback(
     metrics: dict[str, float],
 ) -> dict[str, Any]:

apps/api/tests/test_awoooi_priority_work_order_readback_api.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import urllib.error
 from pathlib import Path
 
 import pytest
@@ -9,6 +10,7 @@ from fastapi.testclient import TestClient
 
 from src.api.v1 import agents
 from src.api.v1.agents import router
+from src.services import awoooi_priority_work_order_readback as priority_service
 from src.services.ai_agent_log_controlled_writeback_executor_readback import (
     load_latest_ai_agent_log_controlled_writeback_executor_readback,
 )
@@ -18,6 +20,7 @@ from src.services.awoooi_priority_work_order_readback import (
     apply_harbor_registry_controlled_recovery_preflight,
     apply_stockplatform_public_api_controlled_recovery_preflight,
     apply_stockplatform_public_api_runtime_readback,
+    load_controlled_cd_lane_live_metric_readback,
     load_latest_awoooi_priority_work_order_readback,
 )
 from src.services.harbor_registry_controlled_recovery_preflight import (
@@ -164,6 +167,72 @@ def test_awoooi_priority_work_order_readback_endpoint_returns_snapshot(
     assert "do not reopen Harbor 502" in data["next_execution_order"][0]
 
 
+def test_controlled_cd_lane_live_metric_readback_falls_back_to_prometheus(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    values = {
+        "awoooi_runner_failclosed_enforcer_last_run_timestamp": "1782968690",
+        "awoooi_runner_failclosed_enforcer_apply_performed": "0",
+        "awoooi_runner_failclosed_enforcer_active_job_containers": "0",
+        "awoooi_runner_failclosed_enforcer_lane_process_count": "0",
+        "awoooi_runner_failclosed_enforcer_root_restore_sources_left": "0",
+        "awoooi_runner_failclosed_enforcer_controlled_drain_active_allowed": "0",
+        "awoooi_runner_failclosed_enforcer_controlled_drain_staging_allowed": "0",
+        "awoooi_host_load5_per_core": "0.64",
+        "awoooi_host_gitea_actions_active_container_count": "0",
+        "awoooi_host_gitea_actions_active_process_count": "0",
+    }
+
+    class _Response:
+        def __init__(self, body: bytes):
+            self.body = body
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_args):
+            return False
+
+        def read(self, _limit: int = -1) -> bytes:
+            return self.body
+
+    def _urlopen(request, timeout):  # noqa: ARG001
+        url = getattr(request, "full_url", str(request))
+        if url.endswith("/metrics"):
+            raise urllib.error.URLError("node exporter unavailable")
+        metric_name = next((name for name in values if name in url), "")
+        result = []
+        if metric_name:
+            result = [
+                {
+                    "metric": {"__name__": metric_name, "host": "110"},
+                    "value": [1782968699.0, values[metric_name]],
+                }
+            ]
+        return _Response(
+            json.dumps(
+                {"status": "success", "data": {"result": result}},
+                separators=(",", ":"),
+            ).encode("utf-8")
+        )
+
+    monkeypatch.setattr(priority_service.urllib.request, "urlopen", _urlopen)
+
+    readback = load_controlled_cd_lane_live_metric_readback(
+        metrics_url="http://node-exporter.invalid/metrics",
+        prometheus_query_url="http://prometheus.invalid/api/v1/query",
+    )
+
+    assert readback["status"] == "controlled_cd_lane_live_metric_readback_ready"
+    assert readback["source"] == "prometheus_query_api"
+    assert readback["metadata_only"] is True
+    assert readback["readback"]["enforcer_metric_present"] is True
+    assert readback["readback"]["host_load5_per_core"] == 0.64
+    serialized = json.dumps(readback)
+    assert "node-exporter.invalid" not in serialized
+    assert "prometheus.invalid" not in serialized
+
+
 def test_awoooi_priority_work_order_readback_endpoint_redacts_ai_loop_queue(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,