diff --git a/deploy/telemetry-plane/.env.example b/deploy/telemetry-plane/.env.example new file mode 100644 index 0000000..90017c1 --- /dev/null +++ b/deploy/telemetry-plane/.env.example @@ -0,0 +1,10 @@ +MISSIONCORE_MQTT_BIND_ADDRESS=127.0.0.1 +MISSIONCORE_MQTT_PORT=1883 +MISSIONCORE_TELEMETRY_QUERY_PORT=18030 +MISSIONCORE_DB_NAME=missioncore_telemetry +MISSIONCORE_DB_USER=missioncore_ingest +MISSIONCORE_DB_PASSWORD=replace-with-a-random-local-secret +MISSIONCORE_MQTT_INGEST_USER=missioncore-ingest +MISSIONCORE_MQTT_INGEST_PASSWORD=replace-with-a-random-local-secret +MISSIONCORE_MQTT_WORKER_006_USER=worker-006 +MISSIONCORE_MQTT_WORKER_006_PASSWORD=replace-with-a-different-random-local-secret diff --git a/deploy/telemetry-plane/.gitignore b/deploy/telemetry-plane/.gitignore new file mode 100644 index 0000000..147af14 --- /dev/null +++ b/deploy/telemetry-plane/.gitignore @@ -0,0 +1,2 @@ +.env +runtime/ diff --git a/deploy/telemetry-plane/README.md b/deploy/telemetry-plane/README.md new file mode 100644 index 0000000..4ea6404 --- /dev/null +++ b/deploy/telemetry-plane/README.md @@ -0,0 +1,81 @@ +# NDC Mission Core local telemetry plane + +Portable local-only telemetry infrastructure for compute contours. + +One Docker Compose project, `ndc-mission-core-telemetry`, owns three isolated +containers: + +- `ndc-mission-core-mqtt-broker` — Eclipse Mosquitto `2.1.2-alpine`; +- `ndc-mission-core-telemetry-timescaledb` — TimescaleDB HA OSS + `pg16.14-ts2.28.2-all-oss`; +- `ndc-mission-core-telemetry-normalizer` — Mission Core telemetry normalizer. + +The Telegraf configuration templates cover Windows and Linux, but the agent runs as a +host service rather than inside this Compose project. This preserves access to native +Windows performance counters, Docker Desktop and NVIDIA telemetry. + +The normalizer exposes a read-only normalized telemetry adapter on +`127.0.0.1:18030`. Mission Core reads this adapter; the browser never receives MQTT or +Timescale credentials, and Timescale is not published on a host port. + +The stack is one deployment contour, not one multi-process container. Keeping broker, +normalizer and database in separate containers preserves independent health checks, +least-privilege boundaries and rollback while Compose provides one operator lifecycle: + +```bash +docker compose ps +docker compose up -d +docker compose down +``` + +All NODE.DC-owned Docker objects use the lowercase `ndc-` namespace. Docker names are +case-sensitive identifiers, so the product prefix is normalized to lowercase while the +product name remains NODE.DC in operator-facing copy. + +This directory does not contain credentials. Copy `.env.example` to `.env`, generate +unique passwords and build the ACL/password files before starting the stack. + +```bash +uv run python prepare.py --initialize --mqtt-bind-address +docker compose up -d --build +``` + +Expected Docker object names: + +```text +project: ndc-mission-core-telemetry +network: ndc-mission-core-telemetry +containers: + ndc-mission-core-mqtt-broker + ndc-mission-core-telemetry-normalizer + ndc-mission-core-telemetry-timescaledb +volumes: + ndc-mission-core-mqtt-data + ndc-mission-core-telemetry-timescale-data +``` + +`prepare.py` passes passwords to `mosquitto_passwd` through stdin. Secrets are not placed +in process arguments or committed files. `--initialize` generates `.env` with mode +`0600` and refuses to replace existing credentials. The generated `.env` and `runtime/` +directory are ignored by Git. + +The product architecture and topic contract are defined in +`docs/adr/0031-local-compute-contour-telemetry-plane.md`. + +## Worker agent + +Worker 006 uses the official Windows Telegraf distribution as the host service +`NDC Mission Core Telemetry Agent`. Install or update it with: + +```powershell +.\telegraf\Install-NdcMissionCoreTelegraf.ps1 +.\telegraf\Update-NdcMissionCoreTelegraf.ps1 +``` + +The update path validates the candidate configuration, backs up the active +configuration and rolls back if the service does not return to `Running`. MQTT +credentials are scoped to the service environment and must not be passed on a command +line or stored in the repository. + +The stack and agent are intentionally not started by repository tests. Provisioning a +machine is a separate, explicit operation. diff --git a/deploy/telemetry-plane/compose.yaml b/deploy/telemetry-plane/compose.yaml new file mode 100644 index 0000000..48bfe5c --- /dev/null +++ b/deploy/telemetry-plane/compose.yaml @@ -0,0 +1,112 @@ +name: ndc-mission-core-telemetry + +x-ndc-labels: &ndc-labels + com.nodedc.product: mission-core + com.nodedc.stack: ndc-mission-core-telemetry + com.nodedc.managed-by: docker-compose + +services: + broker: + image: eclipse-mosquitto:2.1.2-alpine + container_name: ndc-mission-core-mqtt-broker + restart: unless-stopped + labels: + <<: *ndc-labels + com.nodedc.role: mqtt-broker + ports: + - "${MISSIONCORE_MQTT_BIND_ADDRESS:-127.0.0.1}:${MISSIONCORE_MQTT_PORT:-1883}:1883" + environment: + MISSIONCORE_MQTT_HEALTH_USER: "${MISSIONCORE_MQTT_INGEST_USER}" + MISSIONCORE_MQTT_HEALTH_PASSWORD: "${MISSIONCORE_MQTT_INGEST_PASSWORD}" + volumes: + - ./mosquitto/mosquitto.conf:/mosquitto/config/mosquitto.conf:ro + - ./runtime/mosquitto/passwords:/mosquitto/config/passwords:ro + - ./runtime/mosquitto/acl:/mosquitto/config/acl:ro + - broker-data:/mosquitto/data + healthcheck: + test: + - CMD-SHELL + - >- + mosquitto_sub -h 127.0.0.1 -t '$$SYS/broker/uptime' -C 1 -W 3 + -u "$$MISSIONCORE_MQTT_HEALTH_USER" + -P "$$MISSIONCORE_MQTT_HEALTH_PASSWORD" + interval: 10s + timeout: 4s + retries: 6 + + timescale: + image: timescale/timescaledb-ha:pg16.14-ts2.28.2-all-oss + container_name: ndc-mission-core-telemetry-timescaledb + restart: unless-stopped + labels: + <<: *ndc-labels + com.nodedc.role: telemetry-database + environment: + POSTGRES_DB: "${MISSIONCORE_DB_NAME}" + POSTGRES_USER: "${MISSIONCORE_DB_USER}" + POSTGRES_PASSWORD: "${MISSIONCORE_DB_PASSWORD}" + volumes: + - timescale-data:/home/postgres/pgdata/data + - ./timescale/001_telemetry.sql:/docker-entrypoint-initdb.d/001_telemetry.sql:ro + healthcheck: + test: ["CMD-SHELL", "pg_isready -U ${MISSIONCORE_DB_USER} -d ${MISSIONCORE_DB_NAME}"] + interval: 10s + timeout: 5s + retries: 10 + + normalizer: + image: nodedc/mission-core-telemetry-normalizer:local + container_name: ndc-mission-core-telemetry-normalizer + build: + context: ../.. + dockerfile: deploy/telemetry-plane/normalizer/Dockerfile + restart: unless-stopped + labels: + <<: *ndc-labels + com.nodedc.role: telemetry-normalizer + depends_on: + broker: + condition: service_healthy + timescale: + condition: service_healthy + ports: + - "127.0.0.1:${MISSIONCORE_TELEMETRY_QUERY_PORT:-18030}:18030" + environment: + MISSIONCORE_MQTT_HOST: broker + MISSIONCORE_MQTT_PORT: "1883" + MISSIONCORE_MQTT_USERNAME: "${MISSIONCORE_MQTT_INGEST_USER}" + MISSIONCORE_MQTT_PASSWORD: "${MISSIONCORE_MQTT_INGEST_PASSWORD}" + MISSIONCORE_DATABASE_DSN: >- + postgresql://${MISSIONCORE_DB_USER}:${MISSIONCORE_DB_PASSWORD}@timescale:5432/${MISSIONCORE_DB_NAME} + MISSIONCORE_QUERY_HOST: "0.0.0.0" + MISSIONCORE_QUERY_PORT: "18030" + healthcheck: + test: + - CMD + - python + - -c + - >- + import urllib.request; + urllib.request.urlopen('http://127.0.0.1:18030/health', timeout=3).read() + interval: 10s + timeout: 4s + retries: 6 + +volumes: + broker-data: + name: ndc-mission-core-mqtt-data + labels: + <<: *ndc-labels + com.nodedc.role: mqtt-storage + timescale-data: + name: ndc-mission-core-telemetry-timescale-data + labels: + <<: *ndc-labels + com.nodedc.role: telemetry-storage + +networks: + default: + name: ndc-mission-core-telemetry + labels: + <<: *ndc-labels + com.nodedc.role: telemetry-network diff --git a/deploy/telemetry-plane/mosquitto/mosquitto.conf b/deploy/telemetry-plane/mosquitto/mosquitto.conf new file mode 100644 index 0000000..bbf63e5 --- /dev/null +++ b/deploy/telemetry-plane/mosquitto/mosquitto.conf @@ -0,0 +1,11 @@ +persistence true +persistence_location /mosquitto/data/ + +listener 1883 0.0.0.0 +allow_anonymous false +password_file /mosquitto/config/passwords +acl_file /mosquitto/config/acl + +log_dest stdout +connection_messages true +log_timestamp true diff --git a/deploy/telemetry-plane/normalizer/Dockerfile b/deploy/telemetry-plane/normalizer/Dockerfile new file mode 100644 index 0000000..4b44ad3 --- /dev/null +++ b/deploy/telemetry-plane/normalizer/Dockerfile @@ -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"] diff --git a/deploy/telemetry-plane/normalizer/normalizer.py b/deploy/telemetry-plane/normalizer/normalizer.py new file mode 100644 index 0000000..e93ba3e --- /dev/null +++ b/deploy/telemetry-plane/normalizer/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[-a-z0-9]+)/" + r"agents/(?P[-a-z0-9]+)/(?Phost|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[-a-z0-9]+)/agents/" + r"(?P[-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() diff --git a/deploy/telemetry-plane/prepare.py b/deploy/telemetry-plane/prepare.py new file mode 100644 index 0000000..f71a63d --- /dev/null +++ b/deploy/telemetry-plane/prepare.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import argparse +import ipaddress +import os +import secrets +import subprocess +from pathlib import Path +from typing import Final + +ROOT: Final = Path(__file__).resolve().parent +ENV_PATH: Final = ROOT / ".env" +RUNTIME: Final = ROOT / "runtime" / "mosquitto" +IMAGE: Final = "eclipse-mosquitto:2.1.2-alpine" +PLACEHOLDER: Final = "replace-with-" + + +def _initialize_environment(bind_address: str, *, overwrite: bool = False) -> None: + if ENV_PATH.exists() and not overwrite: + raise RuntimeError(".env already exists; refusing to overwrite local credentials") + normalized_bind_address = str(ipaddress.ip_address(bind_address)) + values = { + "MISSIONCORE_MQTT_BIND_ADDRESS": normalized_bind_address, + "MISSIONCORE_MQTT_PORT": "1883", + "MISSIONCORE_TELEMETRY_QUERY_PORT": "18030", + "MISSIONCORE_DB_NAME": "missioncore_telemetry", + "MISSIONCORE_DB_USER": "missioncore_ingest", + "MISSIONCORE_DB_PASSWORD": secrets.token_urlsafe(36), + "MISSIONCORE_MQTT_INGEST_USER": "missioncore-ingest", + "MISSIONCORE_MQTT_INGEST_PASSWORD": secrets.token_urlsafe(36), + "MISSIONCORE_MQTT_WORKER_006_USER": "worker-006", + "MISSIONCORE_MQTT_WORKER_006_PASSWORD": secrets.token_urlsafe(36), + } + ENV_PATH.write_text( + "".join(f"{name}={value}\n" for name, value in values.items()), + encoding="utf-8", + ) + os.chmod(ENV_PATH, 0o600) + + +def _environment() -> dict[str, str]: + if not ENV_PATH.is_file(): + raise RuntimeError("copy .env.example to .env and set unique secrets first") + values: dict[str, str] = {} + for raw_line in ENV_PATH.read_text(encoding="utf-8").splitlines(): + line = raw_line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + name, value = line.split("=", 1) + values[name.strip()] = value.strip() + return values + + +def _required(values: dict[str, str], name: str) -> str: + value = values.get(name, "") + if not value or value.startswith(PLACEHOLDER): + raise RuntimeError(f"{name} must contain a non-placeholder value") + return value + + +def _password_entry(path: Path, username: str, password: str, *, create: bool) -> None: + command = [ + "docker", + "run", + "--rm", + "-i", + "-v", + f"{RUNTIME}:/out", + IMAGE, + "mosquitto_passwd", + ] + if create: + command.append("-c") + command.extend([f"/out/{path.name}", username]) + subprocess.run( + command, + check=True, + input=f"{password}\n{password}\n", + text=True, + ) + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument( + "--initialize", + action="store_true", + help="create a private .env with generated local credentials", + ) + parser.add_argument( + "--mqtt-bind-address", + default="127.0.0.1", + help="host IP exposed to telemetry agents", + ) + arguments = parser.parse_args() + if arguments.initialize: + _initialize_environment(arguments.mqtt_bind_address) + values = _environment() + ingest_user = _required(values, "MISSIONCORE_MQTT_INGEST_USER") + ingest_password = _required(values, "MISSIONCORE_MQTT_INGEST_PASSWORD") + worker_user = _required(values, "MISSIONCORE_MQTT_WORKER_006_USER") + worker_password = _required(values, "MISSIONCORE_MQTT_WORKER_006_PASSWORD") + _required(values, "MISSIONCORE_DB_PASSWORD") + + RUNTIME.mkdir(parents=True, exist_ok=True) + password_path = RUNTIME / "passwords" + _password_entry(password_path, ingest_user, ingest_password, create=True) + _password_entry(password_path, worker_user, worker_password, create=False) + acl_path = RUNTIME / "acl" + acl_path.write_text( + "\n".join( + [ + f"user {ingest_user}", + "topic read mission-core/v1/contours/+/agents/+/+", + "topic read $SYS/broker/uptime", + "", + "# Agent username must equal its stable agent id.", + "pattern write mission-core/v1/contours/+/agents/%u/+", + "", + ] + ), + encoding="utf-8", + ) + os.chmod(password_path, 0o600) + os.chmod(acl_path, 0o600) + print("Mosquitto password and ACL files prepared.", flush=True) + + +if __name__ == "__main__": + main() diff --git a/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 b/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 new file mode 100644 index 0000000..f3eb3b3 --- /dev/null +++ b/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 @@ -0,0 +1,120 @@ +[CmdletBinding()] +param( + [string]$Version = "1.38.4", + [string]$ExpectedSha256 = "6c7878ec319471ac85b82443baec2f3fa5dbcf1b6e2da5d5cd2cbb60fff2bb45", + [string]$ConfigurationTemplate = "$PSScriptRoot\mission-core-windows.conf.tmpl" +) + +$ErrorActionPreference = "Stop" +$serviceName = "telegraf" +$installRoot = "C:\Program Files\NDC\Mission Core\Telegraf" +$configurationRoot = "C:\ProgramData\NDC\MissionCore\telemetry-agent" +$configurationPath = Join-Path $configurationRoot "telegraf.conf" +$archiveUrl = "https://dl.influxdata.com/telegraf/releases/telegraf-$($Version)_windows_amd64.zip" +$payload = [Console]::In.ReadToEnd() | ConvertFrom-Json + +foreach ($name in @( + "MISSIONCORE_CONTOUR_ID", + "MISSIONCORE_AGENT_ID", + "MISSIONCORE_NODE_ID", + "MISSIONCORE_MQTT_HOST", + "MISSIONCORE_MQTT_PORT", + "MISSIONCORE_MQTT_USERNAME", + "MISSIONCORE_MQTT_PASSWORD" +)) { + $value = $payload.$name + if (-not $value) { + throw "Provisioning payload is missing $name" + } + Set-Item -Path "Env:$name" -Value ([string]$value) +} + +if (Get-Service -Name $serviceName -ErrorAction SilentlyContinue) { + throw "Service '$serviceName' already exists; refusing an implicit replacement" +} +if (-not (Test-Path -LiteralPath $ConfigurationTemplate -PathType Leaf)) { + throw "Configuration template not found: $ConfigurationTemplate" +} + +$temporaryRoot = Join-Path $env:TEMP "ndc-mission-core-telegraf-$([Guid]::NewGuid().ToString('N'))" +$archivePath = Join-Path $temporaryRoot "telegraf.zip" +$expandedRoot = Join-Path $temporaryRoot "expanded" +try { + New-Item -ItemType Directory -Path $temporaryRoot, $expandedRoot -Force | Out-Null + Invoke-WebRequest -UseBasicParsing -Uri $archiveUrl -OutFile $archivePath + $actualSha256 = (Get-FileHash -LiteralPath $archivePath -Algorithm SHA256).Hash.ToLowerInvariant() + if ($actualSha256 -ne $ExpectedSha256.ToLowerInvariant()) { + throw "Telegraf archive SHA256 mismatch" + } + Expand-Archive -LiteralPath $archivePath -DestinationPath $expandedRoot + $sourceExecutable = Get-ChildItem -Path $expandedRoot -Filter "telegraf.exe" -Recurse | + Select-Object -First 1 + if (-not $sourceExecutable) { + throw "telegraf.exe is missing from the verified archive" + } + + New-Item -ItemType Directory -Path $installRoot, $configurationRoot -Force | Out-Null + Copy-Item -LiteralPath $sourceExecutable.FullName -Destination (Join-Path $installRoot "telegraf.exe") + Copy-Item -LiteralPath $ConfigurationTemplate -Destination $configurationPath + & icacls.exe $configurationRoot /inheritance:r /grant:r ` + "*S-1-5-18:(OI)(CI)F" "*S-1-5-32-544:(OI)(CI)F" | Out-Null + if ($LASTEXITCODE -ne 0) { + throw "Failed to restrict the telemetry agent configuration directory" + } + + $executable = Join-Path $installRoot "telegraf.exe" + $validationOutput = Join-Path $temporaryRoot "validation.out.log" + $validationError = Join-Path $temporaryRoot "validation.error.log" + $validation = Start-Process -FilePath $executable ` + -ArgumentList @("--config", $configurationPath, "--test") ` + -NoNewWindow -Wait -PassThru ` + -RedirectStandardOutput $validationOutput ` + -RedirectStandardError $validationError + if ($validation.ExitCode -ne 0) { + $validationDetail = Get-Content -LiteralPath $validationError -Tail 8 | + Out-String + throw "Telegraf configuration validation failed: $validationDetail" + } + + & $executable --service install --config $configurationPath + if ($LASTEXITCODE -ne 0) { + throw "Telegraf service installation failed" + } + $serviceEnvironment = @( + "MISSIONCORE_CONTOUR_ID=$($payload.MISSIONCORE_CONTOUR_ID)", + "MISSIONCORE_AGENT_ID=$($payload.MISSIONCORE_AGENT_ID)", + "MISSIONCORE_NODE_ID=$($payload.MISSIONCORE_NODE_ID)", + "MISSIONCORE_MQTT_HOST=$($payload.MISSIONCORE_MQTT_HOST)", + "MISSIONCORE_MQTT_PORT=$($payload.MISSIONCORE_MQTT_PORT)", + "MISSIONCORE_MQTT_USERNAME=$($payload.MISSIONCORE_MQTT_USERNAME)", + "MISSIONCORE_MQTT_PASSWORD=$($payload.MISSIONCORE_MQTT_PASSWORD)" + ) + $serviceRegistryPath = "HKLM:\SYSTEM\CurrentControlSet\Services\$serviceName" + Set-ItemProperty -Path $serviceRegistryPath -Name Environment ` + -Type MultiString -Value $serviceEnvironment + Set-ItemProperty -Path $serviceRegistryPath -Name DisplayName ` + -Value "NDC Mission Core Telemetry Agent" + & sc.exe config $serviceName DisplayName= "NDC Mission Core Telemetry Agent" | Out-Null + if ($LASTEXITCODE -ne 0) { + throw "Failed to apply the NDC service display name" + } + Set-Service -Name $serviceName -StartupType Automatic + Start-Service -Name $serviceName + $service = Get-Service -Name $serviceName + $service.WaitForStatus([ServiceProcess.ServiceControllerStatus]::Running, [TimeSpan]::FromSeconds(20)) + + [ordered]@{ + SchemaVersion = "missioncore.telemetry-agent-install-result/v1" + Agent = "Telegraf" + Version = $Version + ServiceName = $service.Name + DisplayName = $service.DisplayName + Status = $service.Status.ToString() + StartType = $service.StartType.ToString() + Configuration = $configurationPath + Sha256 = $actualSha256 + } | ConvertTo-Json -Compress +} +finally { + Remove-Item -LiteralPath $temporaryRoot -Recurse -Force -ErrorAction SilentlyContinue +} diff --git a/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 b/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 new file mode 100644 index 0000000..3ed5a5b --- /dev/null +++ b/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 @@ -0,0 +1,75 @@ +[CmdletBinding()] +param( + [Parameter(Mandatory = $true)] + [string]$ConfigurationTemplate +) + +$ErrorActionPreference = "Stop" +$serviceName = "telegraf" +$installRoot = "C:\Program Files\NDC\Mission Core\Telegraf" +$configurationRoot = "C:\ProgramData\NDC\MissionCore\telemetry-agent" +$configurationPath = Join-Path $configurationRoot "telegraf.conf" +$executable = Join-Path $installRoot "telegraf.exe" +$serviceRegistryPath = "HKLM:\SYSTEM\CurrentControlSet\Services\$serviceName" + +$service = Get-Service -Name $serviceName -ErrorAction Stop +if (-not (Test-Path -LiteralPath $executable -PathType Leaf)) { + throw "NDC Mission Core Telegraf executable is missing" +} +if (-not (Test-Path -LiteralPath $ConfigurationTemplate -PathType Leaf)) { + throw "Configuration template not found: $ConfigurationTemplate" +} + +foreach ($entry in @((Get-ItemProperty -Path $serviceRegistryPath).Environment)) { + $name, $value = $entry -split "=", 2 + if ($name -and $value) { + Set-Item -Path "Env:$name" -Value $value + } +} + +$temporaryRoot = Join-Path $env:TEMP "ndc-mission-core-telegraf-update-$([Guid]::NewGuid().ToString('N'))" +$validationOutput = Join-Path $temporaryRoot "validation.out.log" +$validationError = Join-Path $temporaryRoot "validation.error.log" +$backupRoot = Join-Path $configurationRoot "backups" +$backupPath = Join-Path $backupRoot "telegraf-$([DateTime]::UtcNow.ToString('yyyyMMddTHHmmssZ')).conf" +try { + New-Item -ItemType Directory -Path $temporaryRoot, $backupRoot -Force | Out-Null + $validation = Start-Process -FilePath $executable ` + -ArgumentList @("--config", $ConfigurationTemplate, "--test") ` + -NoNewWindow -Wait -PassThru ` + -RedirectStandardOutput $validationOutput ` + -RedirectStandardError $validationError + if ($validation.ExitCode -ne 0) { + $validationDetail = Get-Content -LiteralPath $validationError -Tail 8 | + Out-String + throw "Telegraf configuration validation failed: $validationDetail" + } + + Copy-Item -LiteralPath $configurationPath -Destination $backupPath + Stop-Service -Name $serviceName + try { + Copy-Item -LiteralPath $ConfigurationTemplate -Destination $configurationPath + Start-Service -Name $serviceName + $service = Get-Service -Name $serviceName + $service.WaitForStatus( + [ServiceProcess.ServiceControllerStatus]::Running, + [TimeSpan]::FromSeconds(20) + ) + } + catch { + Copy-Item -LiteralPath $backupPath -Destination $configurationPath + Start-Service -Name $serviceName + throw + } + + [ordered]@{ + SchemaVersion = "missioncore.telemetry-agent-config-result/v1" + ServiceName = $serviceName + Status = (Get-Service -Name $serviceName).Status.ToString() + Configuration = $configurationPath + Backup = $backupPath + } | ConvertTo-Json -Compress +} +finally { + Remove-Item -LiteralPath $temporaryRoot -Recurse -Force -ErrorAction SilentlyContinue +} diff --git a/deploy/telemetry-plane/telegraf/mission-core-linux.conf.tmpl b/deploy/telemetry-plane/telegraf/mission-core-linux.conf.tmpl new file mode 100644 index 0000000..8001796 --- /dev/null +++ b/deploy/telemetry-plane/telegraf/mission-core-linux.conf.tmpl @@ -0,0 +1,35 @@ +[agent] + interval = "2s" + round_interval = true + omit_hostname = false + +[global_tags] + node_id = "${MISSIONCORE_NODE_ID}" + contour_id = "${MISSIONCORE_CONTOUR_ID}" + agent_id = "${MISSIONCORE_AGENT_ID}" + +[[inputs.cpu]] + percpu = false + totalcpu = true + report_active = true + +[[inputs.mem]] +[[inputs.disk]] + ignore_fs = ["tmpfs", "devtmpfs", "devfs"] +[[inputs.net]] +[[inputs.system]] +[[inputs.nvidia_smi]] + +[[inputs.docker]] + endpoint = "unix:///var/run/docker.sock" + container_name_include = [] + total = true + +[[outputs.mqtt]] + servers = ["tcp://${MISSIONCORE_MQTT_HOST}:${MISSIONCORE_MQTT_PORT}"] + topic = "mission-core/v1/contours/${MISSIONCORE_CONTOUR_ID}/agents/${MISSIONCORE_AGENT_ID}/host" + username = "${MISSIONCORE_MQTT_USERNAME}" + password = "${MISSIONCORE_MQTT_PASSWORD}" + qos = 1 + data_format = "json" + json_timestamp_units = "1s" diff --git a/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl b/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl new file mode 100644 index 0000000..6155f17 --- /dev/null +++ b/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl @@ -0,0 +1,57 @@ +[agent] + interval = "2s" + round_interval = true + omit_hostname = false + +[global_tags] + node_id = "${MISSIONCORE_NODE_ID}" + contour_id = "${MISSIONCORE_CONTOUR_ID}" + agent_id = "${MISSIONCORE_AGENT_ID}" + +[[inputs.cpu]] + percpu = false + totalcpu = true + report_active = true + +[[inputs.mem]] +[[inputs.system]] +[[inputs.disk]] + ignore_fs = ["tmpfs", "devtmpfs", "devfs"] +[[inputs.net]] +[[inputs.win_perf_counters]] + [[inputs.win_perf_counters.object]] + ObjectName = "System" + Counters = ["System Up Time"] + Instances = ["------"] + Measurement = "win_system" + +[[inputs.nvidia_smi]] + +[[inputs.docker]] + endpoint = "npipe:////./pipe/docker_engine" + container_name_include = [] + +[[inputs.http_response]] + urls = ["http://127.0.0.1:8000/v2/health/ready"] + response_timeout = "2s" + response_status_code = 200 + name_override = "missioncore_triton_health" + +[[inputs.prometheus]] + urls = ["http://127.0.0.1:8002/metrics"] + metric_version = 1 + response_timeout = "2s" + namepass = [ + "nv_inference_request_success", + "nv_inference_request_failure", + "nv_inference_count", + ] + +[[outputs.mqtt]] + servers = ["tcp://${MISSIONCORE_MQTT_HOST}:${MISSIONCORE_MQTT_PORT}"] + topic = "mission-core/v1/contours/${MISSIONCORE_CONTOUR_ID}/agents/${MISSIONCORE_AGENT_ID}/host" + username = "${MISSIONCORE_MQTT_USERNAME}" + password = "${MISSIONCORE_MQTT_PASSWORD}" + qos = 1 + data_format = "json" + json_timestamp_units = "1s" diff --git a/deploy/telemetry-plane/timescale/001_telemetry.sql b/deploy/telemetry-plane/timescale/001_telemetry.sql new file mode 100644 index 0000000..a36082d --- /dev/null +++ b/deploy/telemetry-plane/timescale/001_telemetry.sql @@ -0,0 +1,43 @@ +CREATE EXTENSION IF NOT EXISTS timescaledb; + +CREATE TABLE IF NOT EXISTS contour_telemetry_samples ( + observed_at TIMESTAMPTZ NOT NULL, + ingested_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + contour_id TEXT NOT NULL, + agent_id TEXT NOT NULL, + node_id TEXT NOT NULL, + kind TEXT NOT NULL, + measurement TEXT NOT NULL, + series_key TEXT NOT NULL DEFAULT '', + source_schema TEXT NOT NULL, + source_topic TEXT NOT NULL, + lab_id TEXT, + run_id TEXT, + request_id TEXT, + frame_index BIGINT, + payload JSONB NOT NULL, + PRIMARY KEY ( + observed_at, + contour_id, + agent_id, + kind, + measurement, + series_key + ) +); + +SELECT create_hypertable( + 'contour_telemetry_samples', + by_range('observed_at'), + if_not_exists => TRUE +); + +CREATE INDEX IF NOT EXISTS contour_telemetry_latest_idx + ON contour_telemetry_samples ( + contour_id, + agent_id, + kind, + measurement, + series_key, + observed_at DESC + ); diff --git a/docs/adr/0031-local-compute-contour-telemetry-plane.md b/docs/adr/0031-local-compute-contour-telemetry-plane.md new file mode 100644 index 0000000..8c1d5a7 --- /dev/null +++ b/docs/adr/0031-local-compute-contour-telemetry-plane.md @@ -0,0 +1,193 @@ +# ADR 0031 — Local compute contour telemetry plane + +Status: accepted for implementation +Date: 2026-07-27 + +## Implementation status + +The first local contour is active: + +- Worker 006 publishes through the native Windows service + `NDC Mission Core Telemetry Agent` (Telegraf); +- MQTT, normalization and storage run as the single Compose project + `ndc-mission-core-telemetry`; +- the normalizer exposes the read-only product adapter only on + `127.0.0.1:18030`; +- Mission Core on canonical port `8000` uses this adapter for live System + telemetry and reports the source as `agent-mqtt`; +- SSH is retained only for explicit profile diagnostics and bootstrap checks. + +The existing Worker 006 containers were renamed in place to +`ndc-mission-core-triton` and `ndc-mission-core-perception-worker`. Their +container identities were preserved, so this namespace migration did not restart the +inference or perception runtimes. + +## Decision + +Mission Core treats a compute worker as a configurable **local compute contour**, not as +one hard-coded workstation. + +The portable telemetry path is: + +```text +Telegraf agent + -> authenticated MQTT + -> Eclipse Mosquitto + -> Mission Core telemetry normalizer + -> TimescaleDB OSS + -> Mission Core System API + -> product workspaces +``` + +The broker, normalizer and telemetry database form one Docker Compose contour named +`ndc-mission-core-telemetry`. They remain separate containers so each process has an +independent healthcheck, least-privilege boundary, lifecycle and rollback path. This is +one operator-managed stack, not one multi-process container. + +The Telegraf agent runs as a native host service. On Worker 006 this is required for +portable access to Windows performance counters, Docker Desktop and NVIDIA telemetry. +Putting it into the Linux Docker Desktop VM would make the agent less universal and +would hide part of the host it is meant to observe. + +The direct SSH/PowerShell probe remains a bounded bootstrap and diagnostic fallback for +the existing Worker 006 profile. It is not the target live telemetry transport. + +## Why these components + +- **Telegraf** is the universal host agent. It already supports Windows and Linux host + counters, Docker, NVIDIA SMI, network counters and custom HTTP/Prometheus inputs. + Hardware discovery does not belong in a custom Mission Core agent. +- **Eclipse Mosquitto** is the local, fully open-source broker. It separates worker + lifecycle and network location from the Mission Core UI. +- **Mission Core telemetry normalizer** owns only product semantics: contour identity, + runtime ownership, LAB/run identity and processing-stage names. These concepts cannot + be delegated to a generic hardware agent. +- **TimescaleDB OSS** stores normalized samples and permits comparison between laboratory + runs without coupling agents to database credentials or schema. + +Only the broker and normalizer know the MQTT credentials. Agents receive a scoped +publisher identity. The UI never receives broker or database credentials. + +## Product surface brief + +### User job + +An operator must be able to carry Mission Core and a worker to another local network, +register one or more compute contours, install the standard telemetry agent, select the +active contour and inspect its hardware, processing and network state. + +### Placement + +- The `Система` left panel is a contour selector. +- A canonical plus action in its header creates another contour configuration. +- System workspace modes (`Вычислительные модули`, `Интеграции`, `Сеть`, + `Журнал и аудит`) live in the content header because they describe the selected + contour. +- A canonical settings utility action opens settings for the selected contour. +- Global Mission Core settings remain in the profile menu and do not mix with contour + settings. + +Rejected placements: + +- keeping the single `Worker 006` as a permanent page identity; +- placing contour settings in a global `Система / Настройки` workspace; +- creating one bespoke page layout per worker; +- placing installation instructions directly in the telemetry dashboard. + +### State grammar + +- `unconfigured`: contour exists but no agent transport is configured; +- `provisioning`: bootstrap material was issued but the agent has not published a sample; +- `online`: a fresh normalized sample matches the expected node identity; +- `stale`: the last valid sample is older than the freshness window; +- `identity-mismatch`: an agent published with a different node identity; +- `offline`: the contour was configured but no current sample is available; +- `legacy-diagnostic`: the UI is temporarily backed by the direct SSH probe. + +The UI must name the evidence source. It must not call an SSH snapshot “live agent +telemetry”. + +### Evidence contract + +Each normalized sample carries: + +- `contour_id`, `node_id`, `agent_id`, `observed_at_utc`; +- source and normalizer schema versions; +- hardware, runtime, network and pipeline measurements; +- optional `lab_id`, `run_id`, `request_id` and frame index; +- ingestion timestamp and source topic. + +Unknown measurements remain `null`; they are never synthesized for presentation. + +Several Telegraf measurements can have the same name and timestamp but belong to +different disks, interfaces, containers or GPU processes. Storage therefore keys a +sample by a stable `series_key` derived from its distinguishing source tags. When a +legacy Telegraf input emits one logical series as several field fragments, the +normalizer merges those fields on conflict instead of replacing an earlier fragment. +This preserves series identity without leaking source-specific tag grammar into the +product API. + +## Deployment boundary + +The first stack runs inside one trusted local network and binds only to explicitly +configured LAN interfaces. Mosquitto uses password authentication and ACLs; anonymous +access is forbidden. TimescaleDB is not exposed outside the compose network. + +All NODE.DC-owned Docker containers, networks and named volumes use a lowercase `ndc-` +prefix. The concrete telemetry objects are: + +- `ndc-mission-core-mqtt-broker`; +- `ndc-mission-core-telemetry-normalizer`; +- `ndc-mission-core-telemetry-timescaledb`; +- network `ndc-mission-core-telemetry`; +- volumes `ndc-mission-core-mqtt-data` and + `ndc-mission-core-telemetry-timescale-data`. + +Vendor image names remain upstream-pinned; the NODE.DC ownership boundary is expressed +by container names and `com.nodedc.*` labels rather than by retagging third-party +images. + +The durable Worker 006 compute processes follow the same namespace: + +- `ndc-mission-core-triton`; +- `ndc-mission-core-perception-worker`. + +Their existing `mission-core-compute` Compose project and network are a documented +legacy predecessor. Renaming that project would recreate the network shared by the +persistent perception worker, so it is intentionally deferred to a separate +availability-reviewed migration rather than hidden inside this container-name change. + +The telemetry probe and laboratory PowerShell launchers accept the previous +`mission-core-*` names only as a bounded migration fallback. They prefer and create the +`ndc-*` names. Historical report text and explicitly named debug/backup containers are +evidence, not durable product runtime, and are not rewritten. + +The same compose bundle can later move from the operator machine to a NODE.DC server +without changing agent topics or UI contracts. Internet relay, multi-tenant access and +remote command authority are out of scope for this decision. + +## MQTT topic contract + +```text +mission-core/v1/contours//agents//host +mission-core/v1/contours//agents//runtime +mission-core/v1/contours//agents//pipeline +mission-core/v1/contours//agents//heartbeat +``` + +Agents may publish only below their own contour and agent prefix. The normalizer may +subscribe to `mission-core/v1/contours/+/agents/+/+`. + +## Consequences + +- Worker 006 becomes the first contour configuration, not an architectural singleton. +- Live System telemetry no longer depends on SSH polling. The existing SSH probe is a + bounded diagnostic path and can be removed after contour provisioning no longer + needs it. +- Host portability is delegated to Telegraf plugins and configuration templates. +- Mission Core code stays responsible for product semantics and stable API contracts, + not OS-specific metric collection. +- Host, container, network and inference telemetry is stored in the normalized path. + Per-stage LAB processing telemetry remains unavailable until a laboratory worker + publishes the native `runtime` and `pipeline` topic contracts; the UI must keep those + stages explicitly unavailable rather than infer them from aggregate hardware load. diff --git a/tests/test_telemetry_normalizer_contract.py b/tests/test_telemetry_normalizer_contract.py new file mode 100644 index 0000000..1ea7c25 --- /dev/null +++ b/tests/test_telemetry_normalizer_contract.py @@ -0,0 +1,164 @@ +from __future__ import annotations + +import importlib.util +import json +from datetime import UTC, datetime +from pathlib import Path +from types import ModuleType + +import pytest +import yaml + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +NORMALIZER_PATH = ( + REPOSITORY_ROOT + / "deploy" + / "telemetry-plane" + / "normalizer" + / "normalizer.py" +) +COMPOSE_PATH = REPOSITORY_ROOT / "deploy" / "telemetry-plane" / "compose.yaml" +NORMALIZER_SOURCE = NORMALIZER_PATH.read_text(encoding="utf-8") + + +def _normalizer() -> ModuleType: + spec = importlib.util.spec_from_file_location( + "missioncore_telemetry_normalizer", + NORMALIZER_PATH, + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_normalizer_accepts_native_telegraf_host_metric() -> None: + normalizer = _normalizer() + payload = json.dumps( + { + "name": "cpu", + "tags": { + "node_id": "DESKTOP-OPJ8J04", + "host": "DESKTOP-OPJ8J04", + "contour_id": "worker-006", + "agent_id": "worker-006", + }, + "fields": {"usage_active": 12.5}, + "timestamp": 1_785_179_600, + } + ).encode() + + row = normalizer._normalize( + "mission-core/v1/contours/worker-006/agents/worker-006/host", + payload, + ) + + assert row[0] == datetime.fromtimestamp(1_785_179_600, tz=UTC) + assert row[1:5] == ("worker-006", "worker-006", "DESKTOP-OPJ8J04", "host") + assert row[5] == "cpu" + assert row[6] == "{}" + assert row[7] == "telegraf.metric-json/v1" + + +def test_normalizer_keeps_same_timestamp_host_measurements_distinct() -> None: + normalizer = _normalizer() + rows = [ + normalizer._normalize( + "mission-core/v1/contours/worker-006/agents/worker-006/host", + json.dumps( + { + "name": name, + "tags": { + "node_id": "DESKTOP-OPJ8J04", + "contour_id": "worker-006", + "agent_id": "worker-006", + }, + "fields": fields, + "timestamp": 1_785_179_600, + } + ).encode(), + ) + for name, fields in ( + ("cpu", {"usage_active": 12.5}), + ("mem", {"used_percent": 43.2}), + ) + ] + + assert rows[0][0] == rows[1][0] + assert rows[0][5] == "cpu" + assert rows[1][5] == "mem" + assert rows[0][:5] == rows[1][:5] + + +def test_normalizer_keeps_same_measurement_series_distinct() -> None: + normalizer = _normalizer() + rows = [ + normalizer._normalize( + "mission-core/v1/contours/worker-006/agents/worker-006/host", + json.dumps( + { + "name": "net", + "tags": { + "node_id": "DESKTOP-OPJ8J04", + "contour_id": "worker-006", + "agent_id": "worker-006", + "interface": interface, + }, + "fields": {"bytes_recv": received}, + "timestamp": 1_785_179_600, + } + ).encode(), + ) + for interface, received in (("Ethernet", 10), ("Wi-Fi", 20)) + ] + + assert rows[0][0:6] == rows[1][0:6] + assert rows[0][6] == '{"interface":"Ethernet"}' + assert rows[1][6] == '{"interface":"Wi-Fi"}' + + +def test_telegraf_upsert_merges_split_fields_in_one_series() -> None: + assert "contour_telemetry_samples.payload -> 'fields'" in NORMALIZER_SOURCE + assert "EXCLUDED.payload -> 'fields'" in NORMALIZER_SOURCE + + +def test_normalizer_rejects_unschematized_pipeline_payload() -> None: + normalizer = _normalizer() + with pytest.raises(ValueError, match="schema"): + normalizer._normalize( + "mission-core/v1/contours/worker-006/agents/worker-006/pipeline", + json.dumps( + { + "node_id": "DESKTOP-OPJ8J04", + "observed_at_utc": "2026-07-27T12:00:00Z", + } + ).encode(), + ) + + +def test_telemetry_plane_uses_the_ndc_docker_namespace() -> None: + document = yaml.safe_load(COMPOSE_PATH.read_text(encoding="utf-8")) + + assert document["name"] == "ndc-mission-core-telemetry" + services = document["services"] + assert { + service["container_name"] + for service in services.values() + } == { + "ndc-mission-core-mqtt-broker", + "ndc-mission-core-telemetry-normalizer", + "ndc-mission-core-telemetry-timescaledb", + } + for service in services.values(): + assert service["container_name"].startswith("ndc-") + assert service["labels"]["com.nodedc.product"] == "mission-core" + assert service["labels"]["com.nodedc.stack"] == "ndc-mission-core-telemetry" + + assert document["networks"]["default"]["name"] == "ndc-mission-core-telemetry" + assert { + volume["name"] + for volume in document["volumes"].values() + } == { + "ndc-mission-core-mqtt-data", + "ndc-mission-core-telemetry-timescale-data", + } diff --git a/tests/test_telemetry_plane_prepare.py b/tests/test_telemetry_plane_prepare.py new file mode 100644 index 0000000..018f88a --- /dev/null +++ b/tests/test_telemetry_plane_prepare.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import importlib.util +import stat +from pathlib import Path +from types import ModuleType + +import pytest + + +def _prepare_module() -> ModuleType: + path = ( + Path(__file__).resolve().parents[1] + / "deploy" + / "telemetry-plane" + / "prepare.py" + ) + specification = importlib.util.spec_from_file_location( + "mission_core_telemetry_prepare", + path, + ) + assert specification is not None + assert specification.loader is not None + module = importlib.util.module_from_spec(specification) + specification.loader.exec_module(module) + return module + + +prepare = _prepare_module() + + +def test_initialize_environment_generates_private_unique_secrets( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + env_path = tmp_path / ".env" + monkeypatch.setattr(prepare, "ENV_PATH", env_path) + + prepare._initialize_environment("192.0.2.15") + + values = dict( + line.split("=", 1) + for line in env_path.read_text(encoding="utf-8").splitlines() + ) + assert values["MISSIONCORE_MQTT_BIND_ADDRESS"] == "192.0.2.15" + secrets = { + values["MISSIONCORE_DB_PASSWORD"], + values["MISSIONCORE_MQTT_INGEST_PASSWORD"], + values["MISSIONCORE_MQTT_WORKER_006_PASSWORD"], + } + assert len(secrets) == 3 + assert all(len(secret) >= 40 for secret in secrets) + assert stat.S_IMODE(env_path.stat().st_mode) == 0o600 + + +def test_initialize_environment_refuses_to_replace_credentials( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + env_path = tmp_path / ".env" + env_path.write_text("existing=true\n", encoding="utf-8") + monkeypatch.setattr(prepare, "ENV_PATH", env_path) + + with pytest.raises(RuntimeError, match="refusing to overwrite"): + prepare._initialize_environment("127.0.0.1") + + assert env_path.read_text(encoding="utf-8") == "existing=true\n"