diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index 6f36e2a..2d865ca 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -33,6 +33,7 @@ from k1link.sessions import ( SessionStore, ) from k1link.web.advanced_laboratory_api import build_advanced_laboratory_router +from k1link.web.compute_contour_api import build_compute_contour_router from k1link.web.device_plugin_composition import load_installed_device_plugins from k1link.web.e30_engineering_api import build_e30_engineering_router from k1link.web.e30_human_review_api import build_e30_human_review_router @@ -624,6 +625,11 @@ app.include_router( root_provider=lambda: REPOSITORY_ROOT / ".runtime" / "system", ) ) +app.include_router( + build_compute_contour_router( + root_provider=lambda: REPOSITORY_ROOT / ".runtime" / "system", + ) +) frontend_dist = REPOSITORY_ROOT / "apps" / "control-station" / "dist" diff --git a/src/k1link/web/compute_contour_api.py b/src/k1link/web/compute_contour_api.py new file mode 100644 index 0000000..1238fc3 --- /dev/null +++ b/src/k1link/web/compute_contour_api.py @@ -0,0 +1,321 @@ +from __future__ import annotations + +import json +import os +import re +import tempfile +import threading +import uuid +from collections.abc import Callable +from datetime import UTC, datetime +from pathlib import Path +from typing import Final, Literal + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel, ConfigDict, Field, field_validator + +CONTOUR_SCHEMA: Final = "missioncore.compute-contour/v1" +CATALOG_SCHEMA: Final = "missioncore.compute-contour-catalog/v1" +INSTALL_SCHEMA: Final = "missioncore.compute-contour-agent-install/v1" +CATALOG_FILE_NAME: Final = "compute-contours.json" +SAFE_IDENTIFIER: Final = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,62}[a-z0-9])?$") +SAFE_NODE_ID: Final = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9_.-]{0,126}[A-Za-z0-9])?$") + +RootProvider = Callable[[], Path] +TelemetryMode = Literal["agent-mqtt", "legacy-ssh"] +ContourPlatform = Literal["windows", "linux", "unknown"] + + +def _utc_now() -> str: + return datetime.now(UTC).isoformat().replace("+00:00", "Z") + + +class StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + +class ComputeContour(StrictModel): + schema_version: Literal["missioncore.compute-contour/v1"] = CONTOUR_SCHEMA + contour_id: str + display_name: str = Field(min_length=1, max_length=80) + expected_node_id: str = Field(min_length=1, max_length=128) + agent_id: str + platform: ContourPlatform = "unknown" + telemetry_mode: TelemetryMode = "agent-mqtt" + address: str = Field(default="", max_length=253) + ssh_port: int = Field(default=22, ge=1, le=65535) + mqtt_host: str = Field(default="127.0.0.1", min_length=1, max_length=253) + mqtt_port: int = Field(default=1883, ge=1, le=65535) + revision: int = Field(default=0, ge=0) + updated_at_utc: str | None = None + + @field_validator("contour_id", "agent_id") + @classmethod + def validate_identifier(cls, value: str) -> str: + normalized = value.strip().lower() + if SAFE_IDENTIFIER.fullmatch(normalized) is None: + raise ValueError("identifier must be a lowercase DNS-safe token") + return normalized + + @field_validator("display_name") + @classmethod + def normalize_display_name(cls, value: str) -> str: + normalized = " ".join(value.split()) + if not normalized: + raise ValueError("display name is required") + return normalized + + @field_validator("expected_node_id") + @classmethod + def validate_node_id(cls, value: str) -> str: + normalized = value.strip() + if SAFE_NODE_ID.fullmatch(normalized) is None: + raise ValueError("node id contains unsupported characters") + return normalized + + @field_validator("address", "mqtt_host") + @classmethod + def validate_host(cls, value: str) -> str: + normalized = value.strip() + if any(character.isspace() or ord(character) < 32 for character in normalized): + raise ValueError("host contains whitespace or control characters") + return normalized + + +class ComputeContourCreate(StrictModel): + display_name: str = Field(min_length=1, max_length=80) + expected_node_id: str = Field(min_length=1, max_length=128) + platform: ContourPlatform = "unknown" + address: str = Field(default="", max_length=253) + ssh_port: int = Field(default=22, ge=1, le=65535) + mqtt_host: str = Field(default="127.0.0.1", min_length=1, max_length=253) + mqtt_port: int = Field(default=1883, ge=1, le=65535) + + @field_validator("display_name") + @classmethod + def normalize_display_name(cls, value: str) -> str: + return ComputeContour.normalize_display_name(value) + + @field_validator("expected_node_id") + @classmethod + def validate_node_id(cls, value: str) -> str: + return ComputeContour.validate_node_id(value) + + @field_validator("address", "mqtt_host") + @classmethod + def validate_host(cls, value: str) -> str: + return ComputeContour.validate_host(value) + + +class ComputeContourPut(ComputeContourCreate): + revision: int = Field(ge=0) + telemetry_mode: TelemetryMode = "agent-mqtt" + + +def default_compute_contour() -> ComputeContour: + return ComputeContour( + contour_id="worker-006", + display_name="Worker 006", + expected_node_id="DESKTOP-OPJ8J04", + agent_id="worker-006", + platform="windows", + telemetry_mode="agent-mqtt", + address="", + ssh_port=22, + mqtt_host="127.0.0.1", + mqtt_port=1883, + ) + + +class ComputeContourStore: + def __init__(self, root: Path) -> None: + self.root = root + self.path = root / CATALOG_FILE_NAME + self._lock = threading.Lock() + + def list_contours(self) -> list[ComputeContour]: + with self._lock: + return self._read_unlocked() + + def create(self, request: ComputeContourCreate) -> ComputeContour: + with self._lock: + contours = self._read_unlocked() + contour_id = f"worker-{uuid.uuid4().hex[:8]}" + contour = ComputeContour( + contour_id=contour_id, + display_name=request.display_name, + expected_node_id=request.expected_node_id, + agent_id=contour_id, + platform=request.platform, + telemetry_mode="agent-mqtt", + address=request.address, + ssh_port=request.ssh_port, + mqtt_host=request.mqtt_host, + mqtt_port=request.mqtt_port, + revision=0, + updated_at_utc=_utc_now(), + ) + contours.append(contour) + self._write_unlocked(contours) + return contour + + def update(self, contour_id: str, request: ComputeContourPut) -> ComputeContour: + with self._lock: + contours = self._read_unlocked() + for index, current in enumerate(contours): + if current.contour_id != contour_id: + continue + if current.revision != request.revision: + raise RuntimeError("compute contour revision changed") + updated = current.model_copy( + update={ + "display_name": request.display_name, + "expected_node_id": request.expected_node_id, + "platform": request.platform, + "telemetry_mode": request.telemetry_mode, + "address": request.address, + "ssh_port": request.ssh_port, + "mqtt_host": request.mqtt_host, + "mqtt_port": request.mqtt_port, + "revision": current.revision + 1, + "updated_at_utc": _utc_now(), + } + ) + contours[index] = updated + self._write_unlocked(contours) + return updated + raise KeyError(contour_id) + + def get(self, contour_id: str) -> ComputeContour: + for contour in self.list_contours(): + if contour.contour_id == contour_id: + return contour + raise KeyError(contour_id) + + def _read_unlocked(self) -> list[ComputeContour]: + if not self.path.is_file(): + return [default_compute_contour()] + try: + document = json.loads(self.path.read_text(encoding="utf-8")) + rows = document.get("contours") if isinstance(document, dict) else None + if not isinstance(rows, list): + return [default_compute_contour()] + contours = [ComputeContour.model_validate(row) for row in rows] + return contours or [default_compute_contour()] + except (OSError, ValueError, TypeError, json.JSONDecodeError): + return [default_compute_contour()] + + def _write_unlocked(self, contours: list[ComputeContour]) -> None: + self.root.mkdir(mode=0o700, parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp( + prefix=".compute-contours.", + suffix=".tmp", + dir=self.root, + ) + temporary = Path(temporary_name) + try: + with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as stream: + json.dump( + { + "schema_version": CATALOG_SCHEMA, + "contours": [ + contour.model_dump(mode="json") for contour in contours + ], + }, + stream, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + os.chmod(temporary, 0o600) + os.replace(temporary, self.path) + finally: + temporary.unlink(missing_ok=True) + + +def _catalog_document(contours: list[ComputeContour]) -> dict[str, object]: + return { + "schema_version": CATALOG_SCHEMA, + "contours": [contour.model_dump(mode="json") for contour in contours], + } + + +def _agent_install_document(contour: ComputeContour) -> dict[str, object]: + environment = { + "MISSIONCORE_CONTOUR_ID": contour.contour_id, + "MISSIONCORE_AGENT_ID": contour.agent_id, + "MISSIONCORE_NODE_ID": contour.expected_node_id, + "MISSIONCORE_MQTT_HOST": contour.mqtt_host, + "MISSIONCORE_MQTT_PORT": str(contour.mqtt_port), + "MISSIONCORE_MQTT_USERNAME": contour.agent_id, + } + if contour.platform == "windows": + command = ( + "$env:MISSIONCORE_MQTT_PASSWORD=Read-Host 'MQTT password'; " + "telegraf.exe --service install " + '--config "C:\\ProgramData\\MissionCore\\telegraf.conf"' + ) + template = "deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl" + else: + command = ( + "read -s MISSIONCORE_MQTT_PASSWORD && export MISSIONCORE_MQTT_PASSWORD && " + "sudo telegraf --config /etc/telegraf/telegraf.d/mission-core.conf --test" + ) + template = "deploy/telemetry-plane/telegraf/mission-core-linux.conf.tmpl" + return { + "schema_version": INSTALL_SCHEMA, + "contour_id": contour.contour_id, + "platform": contour.platform, + "agent": { + "distribution": "Telegraf", + "configuration_template": template, + "environment": environment, + "secret_delivery": "interactive-prompt", + }, + "command": command, + "ready": False, + "blocked_reason": ( + "Сначала выпустите scoped MQTT credential в локальном telemetry plane. " + "Пароль не передаётся через Mission Core UI." + ), + } + + +def build_compute_contour_router(*, root_provider: RootProvider) -> APIRouter: + store = ComputeContourStore(root_provider()) + router = APIRouter(prefix="/api/v1/system", tags=["system"]) + + @router.get("/contours") + def list_contours() -> dict[str, object]: + return _catalog_document(store.list_contours()) + + @router.post("/contours", status_code=201) + def create_contour(request: ComputeContourCreate) -> dict[str, object]: + return store.create(request).model_dump(mode="json") + + @router.put("/contours/{contour_id}") + def update_contour( + contour_id: str, + request: ComputeContourPut, + ) -> dict[str, object]: + try: + return store.update(contour_id, request).model_dump(mode="json") + except KeyError as exc: + raise HTTPException(status_code=404, detail="Контур не найден.") from exc + except RuntimeError as exc: + raise HTTPException( + status_code=409, + detail="Конфигурация контура была изменена в другой сессии.", + ) from exc + + @router.get("/contours/{contour_id}/agent-install") + def get_agent_install(contour_id: str) -> dict[str, object]: + try: + return _agent_install_document(store.get(contour_id)) + except KeyError as exc: + raise HTTPException(status_code=404, detail="Контур не найден.") from exc + + return router diff --git a/src/k1link/web/system_telemetry_api.py b/src/k1link/web/system_telemetry_api.py index d34c32c..d0ee7bb 100644 --- a/src/k1link/web/system_telemetry_api.py +++ b/src/k1link/web/system_telemetry_api.py @@ -10,6 +10,8 @@ import subprocess import tempfile import threading import time +import urllib.error +import urllib.request from collections import deque from collections.abc import Callable from datetime import UTC, datetime @@ -27,11 +29,40 @@ DEFAULT_DISPLAY_NAME: Final = "Worker 006" EXPECTED_NODE_ID: Final = "DESKTOP-OPJ8J04" SSH_HOST_ALIAS: Final = "mission-gpu" PROFILE_FILE_NAME: Final = "worker-006.json" -CONTAINER_NAMES: Final = ( - "mission-core-triton", - "mission-core-perception-worker", - "sentinel-frigate", - "sentinel-ollama", +CONTAINER_GROUPS: Final = ( + ( + "ndc-mission-core-triton", + ("ndc-mission-core-triton", "mission-core-triton"), + "Inference Runtime", + "mission-core", + False, + ), + ( + "ndc-mission-core-perception-worker", + ("ndc-mission-core-perception-worker", "mission-core-perception-worker"), + "Perception Pipeline", + "mission-core", + False, + ), + ( + "sentinel-frigate", + ("sentinel-frigate",), + "Sentinel Frigate", + "external", + True, + ), + ( + "sentinel-ollama", + ("sentinel-ollama",), + "Sentinel Ollama", + "external", + True, + ), +) +CONTAINER_NAMES: Final = tuple( + alias + for _canonical, aliases, _role, _owner, _external in CONTAINER_GROUPS + for alias in aliases ) SAFE_HOSTNAME = re.compile( r"^(?=.{1,253}\.?$)(?:[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?\.)*" @@ -41,6 +72,8 @@ PROMETHEUS_SAMPLE = re.compile( r"^(?Pnv_inference_(?:request_success|request_failure|count))" r"(?:\{[^}]*\})?\s+(?P[0-9.eE+-]+)$" ) +TELEMETRY_QUERY_SCHEMA: Final = "missioncore.telemetry-query/v1" +DEFAULT_TELEMETRY_QUERY_URL: Final = "http://127.0.0.1:18030" RootProvider = Callable[[], Path] ProbeRunner = Callable[["WorkerConnectionProfile"], dict[str, Any]] @@ -160,7 +193,9 @@ class WorkerProfileStore: WORKER_PROBE_POWERSHELL: Final = r""" $ErrorActionPreference = "Stop" $containers = @( + "ndc-mission-core-triton", "mission-core-triton", + "ndc-mission-core-perception-worker", "mission-core-perception-worker", "sentinel-frigate", "sentinel-ollama" @@ -170,9 +205,9 @@ $cpu = Get-CimInstance Win32_Processor $computer = Get-CimInstance Win32_ComputerSystem $dockerStats = @{} try { - $statsLines = @(docker stats --no-stream --format "{{json .}}" $containers) - foreach ($line in $statsLines) { - if ($line) { + foreach ($name in $containers) { + $line = docker stats --no-stream --format "{{json .}}" $name 2>$null + if ($LASTEXITCODE -eq 0 -and $line) { $row = $line | ConvertFrom-Json $dockerStats[$row.Name] = $row } @@ -243,10 +278,23 @@ try { ) } catch {} $perceptionHealth = $null -try { - $healthJson = docker exec mission-core-perception-worker python3 -c "import urllib.request;print(urllib.request.urlopen('http://127.0.0.1:18020/health',timeout=2).read().decode())" - $perceptionHealth = $healthJson | ConvertFrom-Json -} catch {} +$perceptionContainer = $null +foreach ($candidate in @( + "ndc-mission-core-perception-worker", + "mission-core-perception-worker" +)) { + docker inspect $candidate *> $null + if ($LASTEXITCODE -eq 0) { + $perceptionContainer = $candidate + break + } +} +if ($perceptionContainer) { + try { + $healthJson = docker exec $perceptionContainer python3 -c "import urllib.request;print(urllib.request.urlopen('http://127.0.0.1:18020/health',timeout=2).read().decode())" + $perceptionHealth = $healthJson | ConvertFrom-Json + } catch {} +} $disks = @( Get-CimInstance Win32_LogicalDisk -Filter "DriveType=3" | ForEach-Object { @@ -355,6 +403,7 @@ def run_worker_probe(profile: WorkerConnectionProfile) -> dict[str, Any]: } return { "schema_version": PROBE_SCHEMA, + "source": "legacy-ssh", "reachable": True, "identity_matches": True, "node_id": node_id, @@ -365,13 +414,265 @@ def run_worker_probe(profile: WorkerConnectionProfile) -> dict[str, Any]: } +def _metric_payload(sample: object) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: + row = _mapping(sample) + payload = _mapping(row.get("payload")) + return payload, _mapping(payload.get("fields")), _mapping(payload.get("tags")) + + +def _format_bytes(value: float | None) -> str: + if value is None: + return "—" + scaled = max(0.0, value) + units = ("B", "KiB", "MiB", "GiB", "TiB") + for unit in units: + if scaled < 1024 or unit == units[-1]: + precision = 0 if unit == "B" else 2 + return f"{scaled:.{precision}f}{unit}" + scaled /= 1024 + return "—" + + +def _agent_raw_document(document: dict[str, Any]) -> dict[str, Any]: + samples = [ + sample for sample in _items(document.get("samples")) if isinstance(sample, dict) + ] + node_ids = { + value + for sample in samples + if isinstance((value := sample.get("node_id")), str) + } + node_id = next(iter(node_ids)) if len(node_ids) == 1 else None + observed_values = [ + value + for sample in samples + if isinstance((value := sample.get("observed_at_utc")), str) + ] + raw: dict[str, Any] = { + "node_id": node_id, + "observed_at_utc": max(observed_values, default=_utc_now()), + "os": {}, + "cpu": {}, + "memory": {}, + "disks": [], + "gpu": None, + "network": [], + "docker_stats": {}, + "container_states": {}, + "triton": {}, + "perception": {}, + } + docker_stats: dict[str, dict[str, str]] = raw["docker_stats"] + container_states: dict[str, dict[str, object]] = raw["container_states"] + for sample in samples: + measurement = sample.get("measurement") + payload, fields, tags = _metric_payload(sample) + if sample.get("kind") == "pipeline": + raw["perception"] = _mapping(payload.get("payload")) or payload + continue + if sample.get("kind") == "runtime": + runtime_payload = _mapping(payload.get("payload")) or payload + if isinstance(runtime_payload.get("triton"), dict): + raw["triton"] = runtime_payload["triton"] + continue + if measurement == "cpu": + raw["cpu"] = { + "name": "Processor", + "logical_processors": None, + "load_percent": _number(fields.get("usage_active")), + } + elif measurement == "system": + cpu = _mapping(raw.get("cpu")) + cpu["logical_processors"] = _number(fields.get("n_cpus")) + raw["cpu"] = cpu + raw["os"] = { + "caption": fields.get("platform") or fields.get("os") or "Windows", + "version": fields.get("platform_version"), + "uptime_seconds": _number(fields.get("uptime")), + } + elif measurement == "mem": + total = _number(fields.get("total")) + available = _number(fields.get("available")) + if available is None: + available = _number(fields.get("free")) + raw["memory"] = { + "total_bytes": total, + "free_bytes": available, + } + elif measurement == "disk": + raw["disks"].append( + { + "name": tags.get("path") or tags.get("device"), + "size_bytes": _number(fields.get("total")), + "free_bytes": _number(fields.get("free")), + } + ) + elif measurement == "net": + raw["network"].append( + { + "name": tags.get("interface") or "unknown", + "description": None, + "status": "Up", + "link_speed_bps": _number(fields.get("speed")), + "mac_address": None, + "addresses": [], + "received_bytes": _number(fields.get("bytes_recv")), + "sent_bytes": _number(fields.get("bytes_sent")), + } + ) + elif measurement == "nvidia_smi": + memory_used = _number(fields.get("memory_used")) + memory_total = _number(fields.get("memory_total")) + raw["gpu"] = { + "name": tags.get("name") or tags.get("gpu_name"), + "utilization_percent": _number(fields.get("utilization_gpu")), + "memory_used_mib": memory_used, + "memory_total_mib": memory_total, + "power_watts": _number(fields.get("power_draw")), + "temperature_celsius": _number(fields.get("temperature_gpu")), + } + elif measurement == "missioncore_triton_health": + raw["triton"] = { + "ready": fields.get("response_status_code_match") == 1 + or fields.get("http_response_code") == 200, + "metrics": _items(_mapping(raw.get("triton")).get("metrics")), + } + elif measurement in { + "nv_inference_request_success", + "nv_inference_request_failure", + "nv_inference_count", + }: + value = next( + ( + candidate + for candidate in ( + _number(fields.get("gauge")), + _number(fields.get("counter")), + _number(fields.get("value")), + ) + if candidate is not None + ), + None, + ) + if value is not None: + triton = _mapping(raw.get("triton")) + metrics = [ + line for line in _items(triton.get("metrics")) if isinstance(line, str) + ] + metrics.append(f"{measurement} {value}") + triton["metrics"] = metrics + raw["triton"] = triton + elif isinstance(measurement, str) and measurement.startswith("docker_container_"): + container_name = tags.get("container_name") or tags.get("container") + if not isinstance(container_name, str) or not container_name: + continue + stats = docker_stats.setdefault(container_name, {"Name": container_name}) + container_states.setdefault( + container_name, + { + "state": {"Status": "running"}, + "image": tags.get("container_image"), + }, + ) + if measurement == "docker_container_cpu": + usage = _number(fields.get("usage_percent")) + if usage is not None: + stats["CPUPerc"] = f"{usage:.3f}%" + elif measurement == "docker_container_mem": + usage = _number(fields.get("usage")) + limit = _number(fields.get("limit")) + usage_percent = _number(fields.get("usage_percent")) + stats["MemUsage"] = f"{_format_bytes(usage)} / {_format_bytes(limit)}" + if usage_percent is not None: + stats["MemPerc"] = f"{usage_percent:.3f}%" + elif measurement == "docker_container_net": + received = _number(fields.get("rx_bytes")) + sent = _number(fields.get("tx_bytes")) + stats["NetIO"] = f"{_format_bytes(received)} / {_format_bytes(sent)}" + elif measurement == "docker_container_blkio": + total = _number(fields.get("io_service_bytes_recursive_total")) + stats["BlockIO"] = _format_bytes(total) + elif measurement == "win_system": + uptime = _number(fields.get("System_Up_Time")) + if not raw["os"]: + raw["os"] = { + "caption": "Windows", + "version": None, + "uptime_seconds": uptime, + } + return raw + + +def run_worker_agent_probe(profile: WorkerConnectionProfile) -> dict[str, Any]: + started = time.perf_counter() + base_url = os.environ.get( + "MISSIONCORE_TELEMETRY_QUERY_URL", + DEFAULT_TELEMETRY_QUERY_URL, + ).rstrip("/") + contour_id = os.environ.get("MISSIONCORE_TELEMETRY_CONTOUR_ID", "worker-006") + agent_id = os.environ.get("MISSIONCORE_TELEMETRY_AGENT_ID", "worker-006") + url = ( + f"{base_url}/v1/contours/{contour_id}/agents/{agent_id}/latest" + "?max_age_seconds=30" + ) + try: + with urllib.request.urlopen(url, timeout=3) as response: # noqa: S310 + document = json.loads(response.read()) + except ( + OSError, + urllib.error.URLError, + json.JSONDecodeError, + TypeError, + ValueError, + ): + return _failed_probe( + profile, + "telemetry-agent-unavailable", + started, + source="agent-mqtt", + ) + if not isinstance(document, dict) or document.get("schema_version") != ( + TELEMETRY_QUERY_SCHEMA + ): + return _failed_probe( + profile, + "invalid-telemetry-response", + started, + source="agent-mqtt", + ) + raw = _agent_raw_document(document) + node_id = raw.get("node_id") + if not isinstance(node_id, str): + return _failed_probe( + profile, + "telemetry-agent-stale", + started, + source="agent-mqtt", + ) + identity_matches = node_id == profile.expected_node_id + return { + "schema_version": PROBE_SCHEMA, + "source": "agent-mqtt", + "reachable": True, + "identity_matches": identity_matches, + "node_id": node_id, + "latency_ms": (time.perf_counter() - started) * 1000, + "observed_at_utc": raw.get("observed_at_utc") or _utc_now(), + "error_code": None if identity_matches else "worker-identity-mismatch", + "raw": raw if identity_matches else None, + } + + def _failed_probe( profile: WorkerConnectionProfile, error_code: str, started: float, + *, + source: str = "legacy-ssh", ) -> dict[str, Any]: return { "schema_version": PROBE_SCHEMA, + "source": source, "reachable": False, "identity_matches": False, "node_id": None, @@ -405,23 +706,30 @@ def _percent(used: float | None, total: float | None) -> float | None: def _container_document( - name: str, + canonical_name: str, + aliases: tuple[str, ...], + role: str, + owner: str, + external: bool, raw_stats: dict[str, Any], raw_states: dict[str, Any], ) -> dict[str, Any]: + name = next( + ( + alias + for alias in aliases + if isinstance(raw_states.get(alias), dict) + or isinstance(raw_stats.get(alias), dict) + ), + canonical_name, + ) stats = _mapping(raw_stats.get(name)) descriptor = _mapping(raw_states.get(name)) state = _mapping(descriptor.get("state")) health = _mapping(state.get("Health")) - roles = { - "mission-core-triton": ("Inference Runtime", "mission-core", False), - "mission-core-perception-worker": ("Perception Pipeline", "mission-core", False), - "sentinel-frigate": ("Sentinel Frigate", "external", True), - "sentinel-ollama": ("Sentinel Ollama", "external", True), - } - role, owner, external = roles[name] return { "name": name, + "canonical_name": canonical_name, "role": role, "owner": owner, "external": external, @@ -593,10 +901,12 @@ class WorkerTelemetryService: store: WorkerProfileStore, probe_runner: ProbeRunner = run_worker_probe, *, + telemetry_probe_runner: ProbeRunner | None = None, cache_seconds: float = 1.5, ) -> None: self.store = store self.probe_runner = probe_runner + self.telemetry_probe_runner = telemetry_probe_runner or probe_runner self.cache_seconds = cache_seconds self._lock = threading.Lock() self._cached_at = 0.0 @@ -654,7 +964,7 @@ class WorkerTelemetryService: "history": list(self._history)[-history_limit:], } profile = self.store.read() - probe = self.probe_runner(profile) + probe = self.telemetry_probe_runner(profile) document = self._telemetry_document(profile, probe, now) self._cached = document self._cached_at = now @@ -719,7 +1029,16 @@ class WorkerTelemetryService: raw_stats = _mapping(raw.get("docker_stats")) raw_states = _mapping(raw.get("container_states")) runtimes = [ - _container_document(name, raw_stats, raw_states) for name in CONTAINER_NAMES + _container_document( + canonical_name, + aliases, + role, + owner, + external, + raw_stats, + raw_states, + ) + for canonical_name, aliases, role, owner, external in CONTAINER_GROUPS ] triton = _mapping(raw.get("triton")) triton_document = { @@ -801,6 +1120,11 @@ class WorkerTelemetryService: def _public_probe(probe: dict[str, Any]) -> dict[str, Any]: return { "schema_version": PROBE_SCHEMA, + "source": ( + probe.get("source") + if probe.get("source") in {"agent-mqtt", "legacy-ssh"} + else "unknown" + ), "reachable": probe.get("reachable") is True, "identity_matches": probe.get("identity_matches") is True, "node_id": probe.get("node_id") if isinstance(probe.get("node_id"), str) else None, @@ -822,9 +1146,14 @@ def build_system_telemetry_router( *, root_provider: RootProvider, probe_runner: ProbeRunner = run_worker_probe, + telemetry_probe_runner: ProbeRunner = run_worker_agent_probe, ) -> APIRouter: store = WorkerProfileStore(root_provider()) - service = WorkerTelemetryService(store, probe_runner) + service = WorkerTelemetryService( + store, + probe_runner, + telemetry_probe_runner=telemetry_probe_runner, + ) router = APIRouter(prefix="/api/v1/system", tags=["system"]) @router.get("/worker-profile") diff --git a/tests/test_compute_contour_api.py b/tests/test_compute_contour_api.py new file mode 100644 index 0000000..7c01f5b --- /dev/null +++ b/tests/test_compute_contour_api.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import stat +from collections.abc import Callable +from pathlib import Path +from typing import Any + +import pytest +from fastapi import APIRouter, HTTPException +from fastapi.routing import APIRoute + +from k1link.web.compute_contour_api import ( + CATALOG_SCHEMA, + ComputeContourCreate, + ComputeContourPut, + ComputeContourStore, + build_compute_contour_router, +) + + +def _endpoint(router: APIRouter, path: str, method: str) -> Callable[..., Any]: + for route in router.routes: + if ( + isinstance(route, APIRoute) + and route.path == path + and route.methods is not None + and method in route.methods + ): + return route.endpoint + raise AssertionError(f"{method} {path} route is missing") + + +def test_contour_store_migrates_worker_006_as_first_configuration( + tmp_path: Path, +) -> None: + store = ComputeContourStore(tmp_path / "system") + contours = store.list_contours() + + assert len(contours) == 1 + assert contours[0].contour_id == "worker-006" + assert contours[0].telemetry_mode == "agent-mqtt" + assert not store.path.exists() + + +def test_contour_store_creates_and_updates_private_catalog(tmp_path: Path) -> None: + store = ComputeContourStore(tmp_path / "system") + created = store.create( + ComputeContourCreate( + display_name="Field Worker", + expected_node_id="FIELD-01", + platform="linux", + address="192.0.2.25", + mqtt_host="192.0.2.5", + ) + ) + updated = store.update( + created.contour_id, + ComputeContourPut( + revision=created.revision, + display_name="Field Worker 01", + expected_node_id="FIELD-01", + platform="linux", + telemetry_mode="agent-mqtt", + address="192.0.2.25", + ssh_port=22, + mqtt_host="192.0.2.5", + mqtt_port=1883, + ), + ) + + assert updated.display_name == "Field Worker 01" + assert updated.revision == 1 + assert stat.S_IMODE(store.path.stat().st_mode) == 0o600 + assert len(store.list_contours()) == 2 + with pytest.raises(RuntimeError, match="revision changed"): + store.update( + created.contour_id, + ComputeContourPut( + revision=0, + display_name="stale", + expected_node_id="FIELD-01", + platform="linux", + telemetry_mode="agent-mqtt", + address="", + ssh_port=22, + mqtt_host="127.0.0.1", + mqtt_port=1883, + ), + ) + + +def test_contour_router_exposes_catalog_and_safe_install_contract( + tmp_path: Path, +) -> None: + router = build_compute_contour_router( + root_provider=lambda: tmp_path / "system" + ) + list_contours = _endpoint(router, "/api/v1/system/contours", "GET") + install = _endpoint( + router, + "/api/v1/system/contours/{contour_id}/agent-install", + "GET", + ) + + catalog = list_contours() + assert catalog["schema_version"] == CATALOG_SCHEMA + assert catalog["contours"][0]["contour_id"] == "worker-006" + document = install("worker-006") + assert document["agent"]["distribution"] == "Telegraf" + assert "MQTT password" in document["command"] + assert "password" not in document["agent"]["environment"] + assert document["ready"] is False + + +def test_contour_router_returns_404_for_unknown_contour(tmp_path: Path) -> None: + router = build_compute_contour_router( + root_provider=lambda: tmp_path / "system" + ) + install = _endpoint( + router, + "/api/v1/system/contours/{contour_id}/agent-install", + "GET", + ) + + with pytest.raises(HTTPException) as error: + install("missing") + assert error.value.status_code == 404 diff --git a/tests/test_system_telemetry_api.py b/tests/test_system_telemetry_api.py index 0d9a080..829f61d 100644 --- a/tests/test_system_telemetry_api.py +++ b/tests/test_system_telemetry_api.py @@ -16,6 +16,7 @@ from k1link.web.system_telemetry_api import ( WorkerConnectionProfilePut, WorkerProfileStore, WorkerTelemetryService, + _agent_raw_document, _ssh_arguments, build_system_telemetry_router, ) @@ -42,6 +43,7 @@ def _probe( return { "reachable": True, "identity_matches": node_id == EXPECTED_NODE_ID, + "source": "agent-mqtt", "node_id": node_id, "latency_ms": 12.5, "observed_at_utc": "2026-07-27T12:00:00Z", @@ -208,6 +210,7 @@ def test_worker_telemetry_separates_mission_core_and_external_load( runtimes = {runtime["name"]: runtime for runtime in document["runtimes"]} assert document["connection"]["identity_matches"] is True + assert document["connection"]["source"] == "agent-mqtt" assert document["node"]["memory"]["used_percent"] == 60 assert document["node"]["gpu"]["memory_used_percent"] == 50 assert document["node"]["triton"]["requests_succeeded"] == 12 @@ -237,6 +240,106 @@ def test_worker_telemetry_separates_mission_core_and_external_load( )["share_percent"] is None +def test_worker_telemetry_prefers_ndc_container_names_during_migration( + tmp_path: Path, +) -> None: + probe = _probe() + raw = probe["raw"] + triton_stats = raw["docker_stats"].pop("mission-core-triton") + triton_stats["Name"] = "ndc-mission-core-triton" + raw["docker_stats"]["ndc-mission-core-triton"] = triton_stats + raw["container_states"]["ndc-mission-core-triton"] = raw[ + "container_states" + ].pop("mission-core-triton") + + service = WorkerTelemetryService( + WorkerProfileStore(tmp_path / "system"), + lambda _: probe, + cache_seconds=0, + ) + document = service.snapshot(10) + mission_core_runtimes = [ + runtime for runtime in document["runtimes"] if not runtime["external"] + ] + + assert len(mission_core_runtimes) == 2 + triton = next( + runtime + for runtime in mission_core_runtimes + if runtime["role"] == "Inference Runtime" + ) + assert triton["name"] == "ndc-mission-core-triton" + assert triton["canonical_name"] == "ndc-mission-core-triton" + + +def test_agent_metrics_are_mapped_to_the_existing_product_contract() -> None: + document = _agent_raw_document( + { + "samples": [ + { + "node_id": EXPECTED_NODE_ID, + "observed_at_utc": "2026-07-27T12:00:00Z", + "kind": "host", + "measurement": "cpu", + "payload": { + "fields": {"usage_active": 12.5}, + "tags": {"cpu": "cpu-total"}, + }, + }, + { + "node_id": EXPECTED_NODE_ID, + "observed_at_utc": "2026-07-27T12:00:00Z", + "kind": "host", + "measurement": "mem", + "payload": { + "fields": { + "total": 1_000, + "available": 400, + }, + "tags": {}, + }, + }, + { + "node_id": EXPECTED_NODE_ID, + "observed_at_utc": "2026-07-27T12:00:00Z", + "kind": "host", + "measurement": "net", + "payload": { + "fields": { + "bytes_recv": 2_000, + "bytes_sent": 1_000, + }, + "tags": {"interface": "Ethernet"}, + }, + }, + { + "node_id": EXPECTED_NODE_ID, + "observed_at_utc": "2026-07-27T12:00:00Z", + "kind": "host", + "measurement": "docker_container_cpu", + "payload": { + "fields": {"usage_percent": 2.5}, + "tags": { + "container_name": "ndc-mission-core-triton", + "container_image": "triton@sha256:accepted", + }, + }, + }, + ] + } + ) + + assert document["node_id"] == EXPECTED_NODE_ID + assert document["cpu"]["load_percent"] == 12.5 + assert document["memory"] == {"total_bytes": 1_000.0, "free_bytes": 400.0} + assert document["network"][0]["name"] == "Ethernet" + assert document["docker_stats"]["ndc-mission-core-triton"]["CPUPerc"] == "2.500%" + assert ( + document["container_states"]["ndc-mission-core-triton"]["image"] + == "triton@sha256:accepted" + ) + + def test_profile_apply_fails_closed_on_wrong_node_and_keeps_old_profile( tmp_path: Path, ) -> None: