feat(system): add compute contour telemetry APIs
This commit is contained in:
DCCONSTRUCTIONS
@@ -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"
|
||||
|
||||
@@ -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
|
||||
@@ -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"^(?P<name>nv_inference_(?:request_success|request_failure|count))"
|
||||
r"(?:\{[^}]*\})?\s+(?P<value>[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")
|
||||
|
||||
@@ -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
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user