feat(telemetry): add portable local telemetry plane

2026-07-28 00:54:11 +03:00
parent 9fc002d86f
commit b8a008b50e
16 changed files with 1486 additions and 0 deletions
@@ -0,0 +1,8 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+RUN pip install --no-cache-dir "paho-mqtt>=2.1,<3" "psycopg[binary]>=3.2,<4"
+COPY deploy/telemetry-plane/normalizer/normalizer.py /app/normalizer.py
+
+USER 65532:65532
+ENTRYPOINT ["python", "/app/normalizer.py"]
@@ -0,0 +1,378 @@
+from __future__ import annotations
+
+import json
+import os
+import re
+import threading
+import time
+from datetime import UTC, datetime
+from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any, Final
+from urllib.parse import parse_qs, urlparse
+
+TOPIC: Final = "mission-core/v1/contours/+/agents/+/+"
+TOPIC_RE: Final = re.compile(
+    r"^mission-core/v1/contours/(?P<contour>[-a-z0-9]+)/"
+    r"agents/(?P<agent>[-a-z0-9]+)/(?P<kind>host|runtime|pipeline|heartbeat)$"
+)
+SOURCE_SCHEMAS: Final = {
+    "host": "missioncore.agent-host-telemetry/v1",
+    "runtime": "missioncore.agent-runtime-telemetry/v1",
+    "pipeline": "missioncore.agent-pipeline-telemetry/v1",
+    "heartbeat": "missioncore.agent-heartbeat/v1",
+}
+TELEGRAF_SCHEMA: Final = "telegraf.metric-json/v1"
+QUERY_SCHEMA: Final = "missioncore.telemetry-query/v1"
+SAFE_IDENTIFIER: Final = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,62}[a-z0-9])?$")
+
+
+def _required(name: str) -> str:
+    value = os.environ.get(name, "").strip()
+    if not value:
+        raise RuntimeError(f"{name} is required")
+    return value
+
+
+def _timestamp(value: object) -> datetime:
+    if isinstance(value, int | float) and not isinstance(value, bool):
+        return datetime.fromtimestamp(float(value), tz=UTC)
+    if not isinstance(value, str):
+        raise ValueError("sample timestamp is required")
+    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    if parsed.tzinfo is None:
+        raise ValueError("observed_at_utc must include a timezone")
+    return parsed.astimezone(UTC)
+
+
+def _optional_text(document: dict[str, Any], name: str) -> str | None:
+    value = document.get(name)
+    return value if isinstance(value, str) and value else None
+
+
+def _optional_integer(document: dict[str, Any], name: str) -> int | None:
+    value = document.get(name)
+    return value if isinstance(value, int) and not isinstance(value, bool) else None
+
+
+def _tag_text(document: dict[str, Any], name: str) -> str | None:
+    tags = document.get("tags")
+    if not isinstance(tags, dict):
+        return None
+    value = tags.get(name)
+    return value if isinstance(value, str) and value else None
+
+
+def _series_key(document: dict[str, Any]) -> str:
+    tags = document.get("tags")
+    if not isinstance(tags, dict):
+        value = document.get("series_key")
+        return value if isinstance(value, str) else ""
+    stable_tags = {
+        str(name): str(value)
+        for name, value in tags.items()
+        if name not in {"agent_id", "contour_id", "host", "node_id"}
+    }
+    return json.dumps(
+        stable_tags,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    )
+
+
+def _normalize(topic: str, payload: bytes) -> tuple[object, ...]:
+    match = TOPIC_RE.fullmatch(topic)
+    if match is None:
+        raise ValueError("topic is outside the telemetry contract")
+    document = json.loads(payload.decode("utf-8"))
+    if not isinstance(document, dict):
+        raise ValueError("payload must be an object")
+    kind = match.group("kind")
+    source_schema = document.get("schema_version")
+    if source_schema == SOURCE_SCHEMAS[kind]:
+        observed_value = document.get("observed_at_utc")
+        measurement = kind
+    elif (
+        kind == "host"
+        and isinstance(document.get("name"), str)
+        and isinstance(document.get("fields"), dict)
+    ):
+        source_schema = TELEGRAF_SCHEMA
+        observed_value = document.get("timestamp")
+        measurement = document["name"].strip()
+        if not measurement:
+            raise ValueError("Telegraf measurement name is required")
+    else:
+        raise ValueError("payload schema does not match topic kind")
+    node_id = document.get("node_id") or _tag_text(document, "node_id")
+    if not node_id:
+        node_id = _tag_text(document, "host")
+    if not isinstance(node_id, str) or not node_id.strip():
+        raise ValueError("node_id is required")
+    observed_at = _timestamp(observed_value)
+    return (
+        observed_at,
+        match.group("contour"),
+        match.group("agent"),
+        node_id.strip(),
+        kind,
+        measurement,
+        _series_key(document),
+        source_schema,
+        topic,
+        _optional_text(document, "lab_id") or _tag_text(document, "lab_id"),
+        _optional_text(document, "run_id") or _tag_text(document, "run_id"),
+        _optional_text(document, "request_id") or _tag_text(document, "request_id"),
+        _optional_integer(document, "frame_index"),
+        json.dumps(document, separators=(",", ":"), ensure_ascii=False),
+    )
+
+
+class TelemetryQueryServer(ThreadingHTTPServer):
+    daemon_threads = True
+
+    def __init__(self, server_address: tuple[str, int], dsn: str) -> None:
+        super().__init__(server_address, TelemetryQueryHandler)
+        self.dsn = dsn
+
+
+class TelemetryQueryHandler(BaseHTTPRequestHandler):
+    server: TelemetryQueryServer
+
+    def do_GET(self) -> None:  # noqa: N802 - stdlib HTTP handler contract
+        parsed = urlparse(self.path)
+        if parsed.path == "/health":
+            self._health()
+            return
+        match = re.fullmatch(
+            r"/v1/contours/(?P<contour>[-a-z0-9]+)/agents/"
+            r"(?P<agent>[-a-z0-9]+)/latest",
+            parsed.path,
+        )
+        if match is None:
+            self._json(HTTPStatus.NOT_FOUND, {"detail": "not found"})
+            return
+        query = parse_qs(parsed.query)
+        try:
+            max_age_seconds = int(query.get("max_age_seconds", ["120"])[0])
+        except ValueError:
+            self._json(HTTPStatus.BAD_REQUEST, {"detail": "invalid max_age_seconds"})
+            return
+        if not 1 <= max_age_seconds <= 86_400:
+            self._json(HTTPStatus.BAD_REQUEST, {"detail": "invalid max_age_seconds"})
+            return
+        self._latest(
+            match.group("contour"),
+            match.group("agent"),
+            max_age_seconds=max_age_seconds,
+        )
+
+    def log_message(self, format: str, *args: object) -> None:
+        print(f"telemetry query: {format % args}", flush=True)
+
+    def _health(self) -> None:
+        import psycopg  # type: ignore[import-not-found]
+
+        try:
+            with (
+                psycopg.connect(self.server.dsn) as connection,
+                connection.cursor() as cursor,
+            ):
+                cursor.execute("SELECT 1")
+                cursor.fetchone()
+        except psycopg.Error:
+            self._json(
+                HTTPStatus.SERVICE_UNAVAILABLE,
+                {"ok": False, "schema_version": QUERY_SCHEMA},
+            )
+            return
+        self._json(HTTPStatus.OK, {"ok": True, "schema_version": QUERY_SCHEMA})
+
+    def _latest(
+        self,
+        contour_id: str,
+        agent_id: str,
+        *,
+        max_age_seconds: int,
+    ) -> None:
+        import psycopg  # type: ignore[import-not-found]
+
+        if (
+            SAFE_IDENTIFIER.fullmatch(contour_id) is None
+            or SAFE_IDENTIFIER.fullmatch(agent_id) is None
+        ):
+            self._json(HTTPStatus.BAD_REQUEST, {"detail": "invalid identifier"})
+            return
+        try:
+            with (
+                psycopg.connect(self.server.dsn) as connection,
+                connection.cursor() as cursor,
+            ):
+                cursor.execute(
+                    """
+                    SELECT DISTINCT ON (kind, measurement, series_key)
+                        observed_at, ingested_at, node_id, kind, measurement,
+                        series_key, source_schema, lab_id, run_id, request_id,
+                        frame_index, payload
+                    FROM contour_telemetry_samples
+                    WHERE contour_id = %s
+                      AND agent_id = %s
+                      AND observed_at >= NOW() - (%s * INTERVAL '1 second')
+                    ORDER BY kind, measurement, series_key, observed_at DESC
+                    """,
+                    (contour_id, agent_id, max_age_seconds),
+                )
+                rows = cursor.fetchall()
+        except psycopg.Error:
+            self._json(
+                HTTPStatus.SERVICE_UNAVAILABLE,
+                {"detail": "telemetry database unavailable"},
+            )
+            return
+        samples = [
+            {
+                "observed_at_utc": row[0].astimezone(UTC).isoformat().replace("+00:00", "Z"),
+                "ingested_at_utc": row[1].astimezone(UTC).isoformat().replace("+00:00", "Z"),
+                "node_id": row[2],
+                "kind": row[3],
+                "measurement": row[4],
+                "series_key": row[5],
+                "source_schema": row[6],
+                "lab_id": row[7],
+                "run_id": row[8],
+                "request_id": row[9],
+                "frame_index": row[10],
+                "payload": row[11],
+            }
+            for row in rows
+        ]
+        self._json(
+            HTTPStatus.OK,
+            {
+                "schema_version": QUERY_SCHEMA,
+                "contour_id": contour_id,
+                "agent_id": agent_id,
+                "samples": samples,
+            },
+        )
+
+    def _json(self, status: HTTPStatus, document: dict[str, object]) -> None:
+        body = json.dumps(
+            document,
+            ensure_ascii=False,
+            separators=(",", ":"),
+        ).encode("utf-8")
+        self.send_response(status.value)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+
+def _start_query_server(dsn: str) -> ThreadingHTTPServer:
+    host = os.environ.get("MISSIONCORE_QUERY_HOST", "0.0.0.0")
+    port = int(os.environ.get("MISSIONCORE_QUERY_PORT", "18030"))
+    server = TelemetryQueryServer((host, port), dsn)
+    thread = threading.Thread(
+        target=server.serve_forever,
+        name="telemetry-query",
+        daemon=True,
+    )
+    thread.start()
+    return server
+
+
+def main() -> None:
+    import paho.mqtt.client as mqtt
+    import psycopg  # type: ignore[import-not-found]
+    from paho.mqtt import MQTTException
+    from paho.mqtt.enums import CallbackAPIVersion
+
+    dsn = _required("MISSIONCORE_DATABASE_DSN")
+    host = _required("MISSIONCORE_MQTT_HOST")
+    port = int(os.environ.get("MISSIONCORE_MQTT_PORT", "1883"))
+    username = _required("MISSIONCORE_MQTT_USERNAME")
+    password = _required("MISSIONCORE_MQTT_PASSWORD")
+    _start_query_server(dsn)
+    connection = psycopg.connect(dsn, autocommit=True)
+    client = mqtt.Client(
+        callback_api_version=CallbackAPIVersion.VERSION2,
+        client_id="ndc-mission-core-telemetry-normalizer",
+        clean_session=False,
+    )
+    client.username_pw_set(username, password)
+
+    def on_connect(
+        connected_client: Any,
+        _userdata: object,
+        _flags: Any,
+        reason_code: Any,
+        _properties: Any,
+    ) -> None:
+        if reason_code.is_failure:
+            raise RuntimeError(f"MQTT connection rejected: {reason_code}")
+        connected_client.subscribe(TOPIC, qos=1)
+
+    def on_message(_client: Any, _userdata: object, message: Any) -> None:
+        try:
+            row = _normalize(message.topic, message.payload)
+            with connection.cursor() as cursor:
+                cursor.execute(
+                    """
+                    INSERT INTO contour_telemetry_samples (
+                        observed_at, contour_id, agent_id, node_id, kind, measurement,
+                        series_key, source_schema, source_topic, lab_id, run_id,
+                        request_id, frame_index, payload
+                    )
+                    VALUES (
+                        %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
+                        %s::jsonb
+                    )
+                    ON CONFLICT (
+                        observed_at, contour_id, agent_id, kind, measurement, series_key
+                    )
+                    DO UPDATE SET
+                        ingested_at = NOW(),
+                        node_id = EXCLUDED.node_id,
+                        series_key = EXCLUDED.series_key,
+                        source_schema = EXCLUDED.source_schema,
+                        source_topic = EXCLUDED.source_topic,
+                        lab_id = EXCLUDED.lab_id,
+                        run_id = EXCLUDED.run_id,
+                        request_id = EXCLUDED.request_id,
+                        frame_index = EXCLUDED.frame_index,
+                        payload = CASE
+                            WHEN contour_telemetry_samples.source_schema = %s
+                             AND EXCLUDED.source_schema = %s
+                            THEN EXCLUDED.payload || jsonb_build_object(
+                                'fields',
+                                COALESCE(
+                                    contour_telemetry_samples.payload -> 'fields',
+                                    '{}'::jsonb
+                                ) || COALESCE(
+                                    EXCLUDED.payload -> 'fields',
+                                    '{}'::jsonb
+                                )
+                            )
+                            ELSE EXCLUDED.payload
+                        END
+                    """,
+                    (*row, TELEGRAF_SCHEMA, TELEGRAF_SCHEMA),
+                )
+        except (ValueError, UnicodeDecodeError, json.JSONDecodeError, psycopg.Error) as exc:
+            print(f"telemetry sample rejected: {exc}", flush=True)
+
+    client.on_connect = on_connect
+    client.on_message = on_message
+    while True:
+        try:
+            client.connect(host, port, keepalive=30)
+            client.loop_forever(retry_first_connection=True)
+        except (OSError, MQTTException, psycopg.Error) as exc:
+            print(f"telemetry normalizer reconnecting: {exc}", flush=True)
+            time.sleep(3)
+
+
+if __name__ == "__main__":
+    main()