From 79eb4b46f7443fa2a1e05066759f901cfc27b755 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 29 Jul 2026 01:53:31 +0300 Subject: [PATCH] feat(telemetry): expose compute pipeline stage metrics --- .../system/ComputeContourSettingsWindow.tsx | 64 ++- .../src/core/system/computeContours.ts | 3 + .../src/core/system/useWorkerTelemetry.ts | 7 +- .../src/core/system/workerTelemetry.ts | 14 +- .../src/styles/system-telemetry.css | 5 +- .../system/ComputeModulesWorkspace.tsx | 4 +- .../workspaces/system/NetworkWorkspace.tsx | 4 +- .../test/systemTelemetry.test.mjs | 8 +- deploy/telemetry-plane/README.md | 12 + .../telemetry-plane/normalizer/normalizer.py | 4 +- .../Get-NdcMissionCorePipelineTelemetry.ps1 | 130 ++++++ .../Install-NdcMissionCoreTelegraf.ps1 | 9 +- .../Update-NdcMissionCorePerceptionRunner.ps1 | 103 +++++ .../Update-NdcMissionCoreTelegraf.ps1 | 60 ++- .../telegraf/mission-core-windows.conf.tmpl | 25 + .../e23-runner-d35df503-stage-telemetry.patch | 434 ++++++++++++++++++ src/k1link/web/system_telemetry_api.py | 52 ++- tests/test_system_telemetry_api.py | 51 ++ tests/test_telemetry_normalizer_contract.py | 35 ++ 19 files changed, 1000 insertions(+), 24 deletions(-) create mode 100644 deploy/telemetry-plane/telegraf/Get-NdcMissionCorePipelineTelemetry.ps1 create mode 100644 deploy/telemetry-plane/telegraf/Update-NdcMissionCorePerceptionRunner.ps1 create mode 100644 deploy/telemetry-plane/telegraf/patches/e23-runner-d35df503-stage-telemetry.patch diff --git a/apps/control-station/src/components/system/ComputeContourSettingsWindow.tsx b/apps/control-station/src/components/system/ComputeContourSettingsWindow.tsx index 0b21315..63f2679 100644 --- a/apps/control-station/src/components/system/ComputeContourSettingsWindow.tsx +++ b/apps/control-station/src/components/system/ComputeContourSettingsWindow.tsx @@ -57,6 +57,7 @@ function emptyDraft(): ComputeContourDraft { mqtt_host: "127.0.0.1", mqtt_port: 1883, telemetry_poll_interval_seconds: DEFAULT_TELEMETRY_POLL_INTERVAL_SECONDS, + mqtt_publish_interval_seconds: 2, }; } @@ -72,6 +73,7 @@ function draftFromContour(contour: ComputeContour | null): ComputeContourDraft { mqtt_host: contour.mqtt_host, mqtt_port: contour.mqtt_port, telemetry_poll_interval_seconds: contour.telemetry_poll_interval_seconds, + mqtt_publish_interval_seconds: contour.mqtt_publish_interval_seconds, }; } @@ -87,6 +89,9 @@ export function ComputeContourSettingsWindow({ const [telemetryPollIntervalDraft, setTelemetryPollIntervalDraft] = useState( () => String(draftFromContour(contour).telemetry_poll_interval_seconds), ); + const [mqttPublishIntervalDraft, setMqttPublishIntervalDraft] = useState( + () => String(draftFromContour(contour).mqtt_publish_interval_seconds), + ); const [activeSection, setActiveSection] = useState<"connection" | "agent">("connection"); const [install, setInstall] = useState(null); const [busy, setBusy] = useState(false); @@ -97,6 +102,7 @@ export function ComputeContourSettingsWindow({ const nextDraft = draftFromContour(mode === "edit" ? contour : null); setDraft(nextDraft); setTelemetryPollIntervalDraft(String(nextDraft.telemetry_poll_interval_seconds)); + setMqttPublishIntervalDraft(String(nextDraft.mqtt_publish_interval_seconds)); setActiveSection("connection"); setInstall(null); setError(null); @@ -121,6 +127,10 @@ export function ComputeContourSettingsWindow({ () => parseTelemetryPollIntervalDraft(telemetryPollIntervalDraft), [telemetryPollIntervalDraft], ); + const mqttPublishIntervalSeconds = useMemo( + () => parseTelemetryPollIntervalDraft(mqttPublishIntervalDraft), + [mqttPublishIntervalDraft], + ); const valid = useMemo(() => ( Boolean(draft.display_name.trim()) @@ -130,7 +140,8 @@ export function ComputeContourSettingsWindow({ && Number.isInteger(draft.mqtt_port) && draft.mqtt_port > 0 && telemetryPollIntervalSeconds !== null - ), [draft, telemetryPollIntervalSeconds]); + && mqttPublishIntervalSeconds !== null + ), [draft, mqttPublishIntervalSeconds, telemetryPollIntervalSeconds]); const commitTelemetryPollInterval = () => { const resolution = resolveTelemetryPollIntervalDraft( @@ -149,11 +160,34 @@ export function ComputeContourSettingsWindow({ )); }; + const commitMqttPublishInterval = () => { + const resolution = resolveTelemetryPollIntervalDraft( + mqttPublishIntervalDraft, + draft.mqtt_publish_interval_seconds, + ); + setMqttPublishIntervalDraft(resolution.draft); + if (!resolution.accepted) return; + setDraft((current) => ( + current.mqtt_publish_interval_seconds === resolution.seconds + ? current + : { + ...current, + mqtt_publish_interval_seconds: resolution.seconds, + } + )); + }; + const save = async () => { - if (!valid || busy || telemetryPollIntervalSeconds === null) return; + if ( + !valid + || busy + || telemetryPollIntervalSeconds === null + || mqttPublishIntervalSeconds === null + ) return; const committedDraft = { ...draft, telemetry_poll_interval_seconds: telemetryPollIntervalSeconds, + mqtt_publish_interval_seconds: mqttPublishIntervalSeconds, }; setBusy(true); setError(null); @@ -273,7 +307,7 @@ export function ComputeContourSettingsWindow({ }))} /> + setMqttPublishIntervalDraft(event.currentTarget.value)} + onBlur={commitMqttPublishInterval} + onKeyDown={(event) => { + if (event.key === "Enter") { + event.preventDefault(); + event.stopPropagation(); + commitMqttPublishInterval(); + } else if (event.key === "Escape") { + event.preventDefault(); + event.stopPropagation(); + setMqttPublishIntervalDraft( + String(draft.mqtt_publish_interval_seconds), + ); + } + }} + /> {draft.telemetry_mode === "legacy-ssh" ? ( setGeneration((value) => value + 1), []); useEffect(() => { - if (!enabled) { + if (!enabled || !contourId) { setTelemetry(null); setLoading(false); setError(null); @@ -38,7 +39,7 @@ export function useWorkerTelemetry( } const controller = new AbortController(); setLoading(true); - void fetchWorkerTelemetry(controller.signal) + void fetchWorkerTelemetry(contourId, controller.signal) .then((document) => { if (controller.signal.aborted) return; setTelemetry(document); @@ -52,7 +53,7 @@ export function useWorkerTelemetry( if (!controller.signal.aborted) setLoading(false); }); return () => controller.abort(); - }, [enabled, externalRefreshGeneration, generation]); + }, [contourId, enabled, externalRefreshGeneration, generation]); useEffect(() => { if (!enabled || loading) return; diff --git a/apps/control-station/src/core/system/workerTelemetry.ts b/apps/control-station/src/core/system/workerTelemetry.ts index f12eeaf..8b8db9f 100644 --- a/apps/control-station/src/core/system/workerTelemetry.ts +++ b/apps/control-station/src/core/system/workerTelemetry.ts @@ -181,13 +181,17 @@ async function requestJson( } export async function fetchWorkerTelemetry( + contourId: string, signal?: AbortSignal, ): Promise { const document = requireJsonRecord( - await requestJson("/api/v1/system/worker-telemetry?history=90", { - method: "GET", - signal, - }), + await requestJson( + `/api/v1/system/contours/${encodeURIComponent(contourId)}/telemetry?history=90`, + { + method: "GET", + signal, + }, + ), "missioncore.worker-telemetry/v1", ); if ( @@ -198,7 +202,7 @@ export async function fetchWorkerTelemetry( || !isRecord(document.network) || !Array.isArray(document.history) ) { - throw new Error("Срез Worker 006 неполон."); + throw new Error("Срез вычислительного контура неполон."); } return document as unknown as WorkerTelemetry; } diff --git a/apps/control-station/src/styles/system-telemetry.css b/apps/control-station/src/styles/system-telemetry.css index c510f68..6e7089a 100644 --- a/apps/control-station/src/styles/system-telemetry.css +++ b/apps/control-station/src/styles/system-telemetry.css @@ -380,7 +380,10 @@ .worker-stage-list { display: grid; min-width: 0; - grid-template-columns: repeat(3, minmax(0, 1fr)); + grid-template-columns: repeat( + auto-fit, + minmax(min(100%, 24rem), 1fr) + ); margin: 0; padding: 0; gap: 0.48rem; diff --git a/apps/control-station/src/workspaces/system/ComputeModulesWorkspace.tsx b/apps/control-station/src/workspaces/system/ComputeModulesWorkspace.tsx index dd882c8..94aa2f7 100644 --- a/apps/control-station/src/workspaces/system/ComputeModulesWorkspace.tsx +++ b/apps/control-station/src/workspaces/system/ComputeModulesWorkspace.tsx @@ -64,10 +64,10 @@ export function ComputeModulesWorkspace() { } = useComputeContours(); const legacyDiagnostic = selectedContour?.contour_id === "worker-006" && selectedContour.telemetry_mode === "legacy-ssh"; - const supportsLiveTelemetry = selectedContour?.contour_id === "worker-006"; const { telemetry, error } = useWorkerTelemetry( + selectedContour?.contour_id ?? null, (selectedContour?.telemetry_poll_interval_seconds ?? 3) * 1_000, - supportsLiveTelemetry, + selectedContour !== null, telemetryRefreshGeneration, ); const node = telemetry?.node ?? null; diff --git a/apps/control-station/src/workspaces/system/NetworkWorkspace.tsx b/apps/control-station/src/workspaces/system/NetworkWorkspace.tsx index da4fd62..a023add 100644 --- a/apps/control-station/src/workspaces/system/NetworkWorkspace.tsx +++ b/apps/control-station/src/workspaces/system/NetworkWorkspace.tsx @@ -22,10 +22,10 @@ export function NetworkWorkspace() { } = useComputeContours(); const legacyDiagnostic = selectedContour?.contour_id === "worker-006" && selectedContour.telemetry_mode === "legacy-ssh"; - const supportsLiveTelemetry = selectedContour?.contour_id === "worker-006"; const { telemetry, error } = useWorkerTelemetry( + selectedContour?.contour_id ?? null, (selectedContour?.telemetry_poll_interval_seconds ?? 3) * 1_000, - supportsLiveTelemetry, + selectedContour !== null, telemetryRefreshGeneration, ); const aggregate = telemetry?.network.aggregate ?? null; diff --git a/apps/control-station/test/systemTelemetry.test.mjs b/apps/control-station/test/systemTelemetry.test.mjs index 619f00a..4dd9a54 100644 --- a/apps/control-station/test/systemTelemetry.test.mjs +++ b/apps/control-station/test/systemTelemetry.test.mjs @@ -49,7 +49,7 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () => assert.match(workspaceHub, //); assert.match(workspaceHub, //); assert.doesNotMatch(workspaceHub, /worker-telemetry|worker-profile|DESKTOP-OPJ8J04/); - assert.match(core, /\/api\/v1\/system\/worker-telemetry/); + assert.match(core, /\/api\/v1\/system\/contours\/.*\/telemetry/); assert.match(core, /\/api\/v1\/system\/worker-profile/); assert.match(core, /share_percent/); assert.doesNotMatch(core, /@nodedc\/ui-react/); @@ -68,7 +68,7 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () => ); assert.match( telemetryStyles, - /\.worker-stage-list\s*\{[^}]*grid-template-columns:\s*repeat\(3,/s, + /\.worker-stage-list\s*\{[^}]*grid-template-columns:\s*repeat\(\s*auto-fit,\s*minmax\(min\(100%,\s*24rem\),\s*1fr\)/s, ); assert.doesNotMatch( telemetryStyles, @@ -87,7 +87,8 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () => assert.match(contourSettingsHook, /icon: "refresh"/); assert.match(contourSettingsHook, /icon: "settings"/); assert.match(contourSettings, /FieldFrame label="Операционная система"/); - assert.match(contourSettings, /label="Интервал MQTT"/); + assert.match(contourSettings, /label="Обновление интерфейса"/); + assert.match(contourSettings, /label="Интервал MQTT агента"/); assert.match(contourSettings, /value=\{telemetryPollIntervalDraft\}/); assert.match(contourSettings, /onBlur=\{commitTelemetryPollInterval\}/); assert.match(contourSettings, /event\.key === "Escape"/); @@ -96,6 +97,7 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () => /telemetry_poll_interval_seconds:\s*Number\(event\.currentTarget\.value\)/, ); assert.match(contourContract, /telemetry_poll_interval_seconds: number/); + assert.match(contourContract, /mqtt_publish_interval_seconds: number/); assert.match(pollIntervalContract, /MIN_TELEMETRY_POLL_INTERVAL_SECONDS\s*=\s*1/); assert.match(telemetryPolling, /normalizeWorkerTelemetryPollMilliseconds/); assert.match(styles, /system-telemetry\.css/); diff --git a/deploy/telemetry-plane/README.md b/deploy/telemetry-plane/README.md index ead6331..c598aa5 100644 --- a/deploy/telemetry-plane/README.md +++ b/deploy/telemetry-plane/README.md @@ -97,5 +97,17 @@ 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 same host service reads the existing perception worker's loopback `/health` +contract with `Get-NdcMissionCorePipelineTelemetry.ps1` and publishes nine +stage-keyed snapshots to the contour's `pipeline` topic. The perception container +does not receive broker credentials and no second agent container is introduced. +These snapshots expose current durable-worker state and cumulative stage timing; +native per-run lifecycle events remain a separate compute contract. + +When the mounted perception runner itself changes, use +`Update-NdcMissionCorePerceptionRunner.ps1` with exact predecessor and candidate +digests. It backs up the mounted runner, restarts the same container, accepts only a +ready health document with stage metrics, and restores the predecessor on failure. + 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/normalizer/normalizer.py b/deploy/telemetry-plane/normalizer/normalizer.py index daf93d6..ab2c818 100644 --- a/deploy/telemetry-plane/normalizer/normalizer.py +++ b/deploy/telemetry-plane/normalizer/normalizer.py @@ -151,7 +151,7 @@ def _normalize(topic: str, payload: bytes) -> tuple[object, ...]: observed_value = document.get("observed_at_utc") measurement = kind elif ( - kind == "host" + kind in {"host", "pipeline"} and isinstance(document.get("name"), str) and isinstance(document.get("fields"), dict) ): @@ -160,6 +160,8 @@ def _normalize(topic: str, payload: bytes) -> tuple[object, ...]: measurement = document["name"].strip() if not measurement or len(measurement) > 128: raise ValueError("Telegraf measurement name is required") + if kind == "pipeline" and measurement != "missioncore_pipeline": + raise ValueError("Telegraf pipeline measurement is not recognized") else: raise ValueError("payload schema does not match topic kind") node_id = document.get("node_id") or _tag_text(document, "node_id") diff --git a/deploy/telemetry-plane/telegraf/Get-NdcMissionCorePipelineTelemetry.ps1 b/deploy/telemetry-plane/telegraf/Get-NdcMissionCorePipelineTelemetry.ps1 new file mode 100644 index 0000000..7094895 --- /dev/null +++ b/deploy/telemetry-plane/telegraf/Get-NdcMissionCorePipelineTelemetry.ps1 @@ -0,0 +1,130 @@ +[CmdletBinding()] +param( + [string]$ContainerName = "ndc-mission-core-perception-worker", + [string]$HealthUrl = "http://127.0.0.1:18020/health" +) + +$ErrorActionPreference = "Stop" +$stageIds = @( + "source-ingress", + "camera-decode", + "preprocessing", + "detector", + "semantic-model", + "sensor-fusion", + "tracking", + "temporal-state", + "result-publication" +) +$health = $null +$collectorState = "unavailable" +try { + $healthJson = docker exec $ContainerName python3 -c ` + "import urllib.request;print(urllib.request.urlopen('$HealthUrl',timeout=2).read().decode())" ` + 2>$null + if ($LASTEXITCODE -eq 0 -and $healthJson) { + $health = $healthJson | ConvertFrom-Json + $collectorState = "live" + } +} +catch { + $health = $null +} + +$serviceState = if ($health -and $health.state) { + [string]$health.state +} +else { + "unavailable" +} +$currentStage = if ($health -and $health.current_stage) { + [string]$health.current_stage +} +else { + "" +} +$activeStages = if ($health -and $health.active_stages) { + @($health.active_stages | ForEach-Object { [string]$_ }) +} +else { + @() +} +$activeRequestId = if ($health -and $health.active_request_id) { + [string]$health.active_request_id +} +else { + "" +} +$activeFrameIndex = if ( + $health -and + $null -ne $health.active_frame_index +) { + [int64]$health.active_frame_index +} +else { + [int64]-1 +} +$completedRuns = if ($health -and $null -ne $health.completed_runs) { + [int64]$health.completed_runs +} +else { + [int64]0 +} +$failedRuns = if ($health -and $null -ne $health.failed_runs) { + [int64]$health.failed_runs +} +else { + [int64]0 +} +$modelLoadSeconds = if ($health -and $null -ne $health.model_load_seconds) { + [double]$health.model_load_seconds +} +else { + [double]0 +} + +$samples = foreach ($stageId in $stageIds) { + $stageState = if ($collectorState -ne "live") { + "unavailable" + } + elseif ($currentStage -eq $stageId -or $activeStages -contains $stageId) { + "active" + } + elseif ($serviceState -eq "busy") { + "waiting" + } + else { + "ready" + } + $sample = [ordered]@{ + stage_id = $stageId + stage_state = $stageState + service_state = $serviceState + current_stage = $currentStage + active_request_id = $activeRequestId + active_frame_index = $activeFrameIndex + completed_runs = $completedRuns + failed_runs = $failedRuns + model_load_seconds = $modelLoadSeconds + collector_state = $collectorState + health_ok = [bool]($health -and $health.ok) + } + if ($health -and $health.stage_metrics) { + $property = $health.stage_metrics.PSObject.Properties[$stageId] + if ($property) { + $metric = $property.Value + if ($null -ne $metric.elapsed_seconds) { + $sample.elapsed_seconds = [double]$metric.elapsed_seconds + } + if ($null -ne $metric.activations) { + $sample.activations = [int64]$metric.activations + } + if ($null -ne $metric.share_percent) { + $sample.share_percent = [double]$metric.share_percent + } + } + } + [pscustomobject]$sample +} + +@($samples) | ConvertTo-Json -Compress -Depth 8 diff --git a/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 b/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 index 8c7a629..747c268 100644 --- a/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 +++ b/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 @@ -2,7 +2,8 @@ param( [string]$Version = "1.38.4", [string]$ExpectedSha256 = "6c7878ec319471ac85b82443baec2f3fa5dbcf1b6e2da5d5cd2cbb60fff2bb45", - [string]$ConfigurationTemplate = "$PSScriptRoot\mission-core-windows.conf.tmpl" + [string]$ConfigurationTemplate = "$PSScriptRoot\mission-core-windows.conf.tmpl", + [string]$PipelineCollector = "$PSScriptRoot\Get-NdcMissionCorePipelineTelemetry.ps1" ) $ErrorActionPreference = "Stop" @@ -10,6 +11,7 @@ $serviceName = "telegraf" $installRoot = "C:\Program Files\NDC\Mission Core\Telegraf" $configurationRoot = "C:\ProgramData\NDC\MissionCore\telemetry-agent" $configurationPath = Join-Path $configurationRoot "telegraf.conf" +$collectorPath = Join-Path $configurationRoot "Get-NdcMissionCorePipelineTelemetry.ps1" $archiveUrl = "https://dl.influxdata.com/telegraf/releases/telegraf-$($Version)_windows_amd64.zip" $payload = [Console]::In.ReadToEnd() | ConvertFrom-Json @@ -36,6 +38,9 @@ if (Get-Service -Name $serviceName -ErrorAction SilentlyContinue) { if (-not (Test-Path -LiteralPath $ConfigurationTemplate -PathType Leaf)) { throw "Configuration template not found: $ConfigurationTemplate" } +if (-not (Test-Path -LiteralPath $PipelineCollector -PathType Leaf)) { + throw "Pipeline collector not found: $PipelineCollector" +} $temporaryRoot = Join-Path $env:TEMP "ndc-mission-core-telegraf-$([Guid]::NewGuid().ToString('N'))" $archivePath = Join-Path $temporaryRoot "telegraf.zip" @@ -57,6 +62,7 @@ try { 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 + Copy-Item -LiteralPath $PipelineCollector -Destination $collectorPath & 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) { @@ -114,6 +120,7 @@ try { Status = $service.Status.ToString() StartType = $service.StartType.ToString() Configuration = $configurationPath + PipelineCollector = $collectorPath Sha256 = $actualSha256 } | ConvertTo-Json -Compress } diff --git a/deploy/telemetry-plane/telegraf/Update-NdcMissionCorePerceptionRunner.ps1 b/deploy/telemetry-plane/telegraf/Update-NdcMissionCorePerceptionRunner.ps1 new file mode 100644 index 0000000..3edf6bf --- /dev/null +++ b/deploy/telemetry-plane/telegraf/Update-NdcMissionCorePerceptionRunner.ps1 @@ -0,0 +1,103 @@ +[CmdletBinding()] +param( + [Parameter(Mandatory = $true)] + [string]$Candidate, + [Parameter(Mandatory = $true)] + [string]$ExpectedPredecessorSha256, + [Parameter(Mandatory = $true)] + [string]$ExpectedCandidateSha256, + [string]$ContainerName = "ndc-mission-core-perception-worker", + [string]$RunnerRoot = "D:\NDC_MISSIONCORE\runtime\derived\e23-runner-20260724-002" +) + +$ErrorActionPreference = "Stop" +$target = Join-Path $RunnerRoot "run_e15_shadow_inference.py" +$expectedPredecessor = $ExpectedPredecessorSha256.ToLowerInvariant() +$expectedCandidate = $ExpectedCandidateSha256.ToLowerInvariant() +$predecessor = ( + Get-FileHash -Algorithm SHA256 -LiteralPath $target +).Hash.ToLowerInvariant() +if ($predecessor -ne $expectedPredecessor) { + throw "Runner predecessor digest changed" +} +if ( + (Get-FileHash -Algorithm SHA256 -LiteralPath $Candidate).Hash.ToLowerInvariant() ` + -ne $expectedCandidate +) { + throw "Runner candidate digest changed" +} +$backup = Join-Path $RunnerRoot ( + "run_e15_shadow_inference.py.rollback-" + + [DateTime]::UtcNow.ToString("yyyyMMddTHHmmssZ") + + "-" + + $predecessor.Substring(0, 12) +) +$healthCode = "import urllib.request;print(urllib.request.urlopen('http://127.0.0.1:18020/health',timeout=2).read().decode())" + +function Wait-PerceptionHealth { + param( + [bool]$RequireStageMetrics, + [int]$TimeoutSeconds = 150 + ) + + $deadline = [DateTime]::UtcNow.AddSeconds($TimeoutSeconds) + while ([DateTime]::UtcNow -lt $deadline) { + Start-Sleep -Seconds 3 + try { + $healthJson = docker exec $ContainerName python3 -c $healthCode 2>$null + if ($LASTEXITCODE -ne 0 -or -not $healthJson) { + continue + } + $health = $healthJson | ConvertFrom-Json + if ( + $health.ok -and + $health.state -eq "ready" -and + (-not $RequireStageMetrics -or $health.stage_metrics) + ) { + return $health + } + } + catch {} + } + return $null +} + +Copy-Item -LiteralPath $target -Destination $backup +try { + Copy-Item -LiteralPath $Candidate -Destination $target -Force + if ( + (Get-FileHash -Algorithm SHA256 -LiteralPath $target).Hash.ToLowerInvariant() ` + -ne $expectedCandidate + ) { + throw "Runner replacement digest changed" + } + docker restart --time 20 $ContainerName | Out-Null + if ($LASTEXITCODE -ne 0) { + throw "Perception container restart failed" + } + $health = Wait-PerceptionHealth -RequireStageMetrics $true + if (-not $health) { + throw "Stage-instrumented runner did not become ready" + } +} +catch { + Copy-Item -LiteralPath $backup -Destination $target -Force + docker restart --time 20 $ContainerName | Out-Null + $rollbackHealth = Wait-PerceptionHealth -RequireStageMetrics $false + if (-not $rollbackHealth) { + throw "Candidate failed and predecessor rollback did not become ready" + } + throw +} + +[ordered]@{ + SchemaVersion = "missioncore.worker-runner-deploy-result/v1" + Container = $ContainerName + State = (docker inspect --format "{{.State.Status}}" $ContainerName) + PredecessorSha256 = $predecessor + CandidateSha256 = ( + Get-FileHash -Algorithm SHA256 -LiteralPath $target + ).Hash.ToLowerInvariant() + Backup = $backup + Health = $health +} | ConvertTo-Json -Depth 12 -Compress diff --git a/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 b/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 index 1a49bbe..34e4ab2 100644 --- a/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 +++ b/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 @@ -1,7 +1,8 @@ [CmdletBinding()] param( [Parameter(Mandatory = $true)] - [string]$ConfigurationTemplate + [string]$ConfigurationTemplate, + [string]$PipelineCollector = "$PSScriptRoot\Get-NdcMissionCorePipelineTelemetry.ps1" ) $ErrorActionPreference = "Stop" @@ -9,6 +10,7 @@ $serviceName = "telegraf" $installRoot = "C:\Program Files\NDC\Mission Core\Telegraf" $configurationRoot = "C:\ProgramData\NDC\MissionCore\telemetry-agent" $configurationPath = Join-Path $configurationRoot "telegraf.conf" +$collectorPath = Join-Path $configurationRoot "Get-NdcMissionCorePipelineTelemetry.ps1" $executable = Join-Path $installRoot "telegraf.exe" $serviceRegistryPath = "HKLM:\SYSTEM\CurrentControlSet\Services\$serviceName" @@ -19,8 +21,12 @@ if (-not (Test-Path -LiteralPath $executable -PathType Leaf)) { if (-not (Test-Path -LiteralPath $ConfigurationTemplate -PathType Leaf)) { throw "Configuration template not found: $ConfigurationTemplate" } +if (-not (Test-Path -LiteralPath $PipelineCollector -PathType Leaf)) { + throw "Pipeline collector not found: $PipelineCollector" +} -foreach ($entry in @((Get-ItemProperty -Path $serviceRegistryPath).Environment)) { +$serviceEnvironmentBefore = @((Get-ItemProperty -Path $serviceRegistryPath).Environment) +foreach ($entry in $serviceEnvironmentBefore) { $name, $value = $entry -split "=", 2 if ($name -and $value) { Set-Item -Path "Env:$name" -Value $value @@ -29,14 +35,38 @@ foreach ($entry in @((Get-ItemProperty -Path $serviceRegistryPath).Environment)) if (-not $env:MISSIONCORE_TELEMETRY_INTERVAL) { $env:MISSIONCORE_TELEMETRY_INTERVAL = "2s" } +$serviceEnvironmentCandidate = @($serviceEnvironmentBefore) +if (-not ($serviceEnvironmentCandidate | Where-Object { + $_ -like "MISSIONCORE_TELEMETRY_INTERVAL=*" +})) { + $serviceEnvironmentCandidate += ( + "MISSIONCORE_TELEMETRY_INTERVAL=$($env:MISSIONCORE_TELEMETRY_INTERVAL)" + ) +} $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" +$backupCollectorPath = Join-Path $backupRoot ` + "pipeline-collector-$([DateTime]::UtcNow.ToString('yyyyMMddTHHmmssZ')).ps1" +$hadCollector = Test-Path -LiteralPath $collectorPath -PathType Leaf try { New-Item -ItemType Directory -Path $temporaryRoot, $backupRoot -Force | Out-Null + $collectorProbe = & powershell.exe -NoLogo -NoProfile -NonInteractive ` + -ExecutionPolicy Bypass -File $PipelineCollector + if ($LASTEXITCODE -ne 0) { + throw "Pipeline collector validation failed" + } + $collectorDocument = $collectorProbe | ConvertFrom-Json + if (@($collectorDocument).Count -ne 9) { + throw "Pipeline collector must publish exactly nine stage samples" + } + if ($hadCollector) { + Copy-Item -LiteralPath $collectorPath -Destination $backupCollectorPath + } + Copy-Item -LiteralPath $PipelineCollector -Destination $collectorPath $validation = Start-Process -FilePath $executable ` -ArgumentList @("--config", $ConfigurationTemplate, "--test") ` -NoNewWindow -Wait -PassThru ` @@ -49,7 +79,14 @@ try { } Copy-Item -LiteralPath $configurationPath -Destination $backupPath + Set-ItemProperty -Path $serviceRegistryPath -Name Environment ` + -Type MultiString -Value $serviceEnvironmentCandidate Stop-Service -Name $serviceName + $service = Get-Service -Name $serviceName + $service.WaitForStatus( + [ServiceProcess.ServiceControllerStatus]::Stopped, + [TimeSpan]::FromSeconds(20) + ) try { Copy-Item -LiteralPath $ConfigurationTemplate -Destination $configurationPath Start-Service -Name $serviceName @@ -61,6 +98,12 @@ try { } catch { Copy-Item -LiteralPath $backupPath -Destination $configurationPath + if ($hadCollector) { + Copy-Item -LiteralPath $backupCollectorPath -Destination $collectorPath + } + else { + Remove-Item -LiteralPath $collectorPath -Force -ErrorAction SilentlyContinue + } Start-Service -Name $serviceName throw } @@ -71,8 +114,21 @@ try { Status = (Get-Service -Name $serviceName).Status.ToString() Configuration = $configurationPath Backup = $backupPath + PipelineCollector = $collectorPath + PipelineCollectorBackup = if ($hadCollector) { $backupCollectorPath } else { $null } } | ConvertTo-Json -Compress } +catch { + Set-ItemProperty -Path $serviceRegistryPath -Name Environment ` + -Type MultiString -Value $serviceEnvironmentBefore + if ($hadCollector -and (Test-Path -LiteralPath $backupCollectorPath -PathType Leaf)) { + Copy-Item -LiteralPath $backupCollectorPath -Destination $collectorPath + } + elseif (-not $hadCollector) { + Remove-Item -LiteralPath $collectorPath -Force -ErrorAction SilentlyContinue + } + throw +} finally { Remove-Item -LiteralPath $temporaryRoot -Recurse -Force -ErrorAction SilentlyContinue } diff --git a/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl b/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl index db178cb..1a76e3c 100644 --- a/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl +++ b/deploy/telemetry-plane/telegraf/mission-core-windows.conf.tmpl @@ -47,6 +47,20 @@ "nv_inference_count", ] +[[inputs.exec]] + commands = ['powershell.exe -NoLogo -NoProfile -NonInteractive -ExecutionPolicy Bypass -File "C:\ProgramData\NDC\MissionCore\telemetry-agent\Get-NdcMissionCorePipelineTelemetry.ps1"'] + timeout = "4s" + name_override = "missioncore_pipeline" + data_format = "json" + tag_keys = ["stage_id"] + json_string_fields = [ + "stage_state", + "service_state", + "current_stage", + "active_request_id", + "collector_state", + ] + [[outputs.mqtt]] servers = ["tcp://${MISSIONCORE_MQTT_HOST}:${MISSIONCORE_MQTT_PORT}"] topic = "mission-core/v1/contours/${MISSIONCORE_CONTOUR_ID}/agents/${MISSIONCORE_AGENT_ID}/host" @@ -55,3 +69,14 @@ qos = 1 data_format = "json" json_timestamp_units = "1s" + namedrop = ["missioncore_pipeline"] + +[[outputs.mqtt]] + servers = ["tcp://${MISSIONCORE_MQTT_HOST}:${MISSIONCORE_MQTT_PORT}"] + topic = "mission-core/v1/contours/${MISSIONCORE_CONTOUR_ID}/agents/${MISSIONCORE_AGENT_ID}/pipeline" + username = "${MISSIONCORE_MQTT_USERNAME}" + password = "${MISSIONCORE_MQTT_PASSWORD}" + qos = 1 + data_format = "json" + json_timestamp_units = "1s" + namepass = ["missioncore_pipeline"] diff --git a/deploy/telemetry-plane/telegraf/patches/e23-runner-d35df503-stage-telemetry.patch b/deploy/telemetry-plane/telegraf/patches/e23-runner-d35df503-stage-telemetry.patch new file mode 100644 index 0000000..bbb3de3 --- /dev/null +++ b/deploy/telemetry-plane/telegraf/patches/e23-runner-d35df503-stage-telemetry.patch @@ -0,0 +1,434 @@ +--- run_e15_shadow_inference.py ++++ run_e15_shadow_inference.py +@@ -20,7 +20,8 @@ + import threading + import time + from collections import Counter, deque +-from contextlib import suppress ++from collections.abc import Iterator ++from contextlib import contextmanager, suppress + from dataclasses import dataclass + from datetime import UTC, datetime + from pathlib import Path +@@ -108,6 +109,17 @@ + WORLD_SCHEMA = "missioncore.live-perception-world-state/v1" + SEMANTIC_SCHEMA = "missioncore.e15-shadow-semantic-frame/v1" + PIPELINE_ID = "shadow-fmp4-yolox-eomt-kb4-amodal-world-state/v1" ++PIPELINE_STAGE_IDS = ( ++ "source-ingress", ++ "camera-decode", ++ "preprocessing", ++ "detector", ++ "semantic-model", ++ "sensor-fusion", ++ "tracking", ++ "temporal-state", ++ "result-publication", ++) + + + def arguments() -> argparse.Namespace: +@@ -539,6 +551,76 @@ + summary["rss_growth_mib"] = 0.0 + summary["final_queues"] = {} + return summary ++ ++ ++class _StageExecutionTelemetry: ++ """Measure named pipeline spans without pretending they are OS processes.""" ++ ++ def __init__(self, stage_ids: tuple[str, ...] = PIPELINE_STAGE_IDS) -> None: ++ if not stage_ids or len(stage_ids) != len(set(stage_ids)): ++ raise RuntimeError("pipeline stage identities are invalid") ++ self._stage_ids = stage_ids ++ self._lock = threading.Lock() ++ self._next_token = 0 ++ self._active: dict[int, tuple[str, float, int | None]] = {} ++ self._elapsed_seconds = dict.fromkeys(stage_ids, 0.0) ++ self._activations = dict.fromkeys(stage_ids, 0) ++ self._last_frame_index: int | None = None ++ ++ @contextmanager ++ def measure( ++ self, ++ stage_id: str, ++ frame_index: int | None = None, ++ ) -> Iterator[None]: ++ if stage_id not in self._elapsed_seconds: ++ raise RuntimeError(f"unknown pipeline stage: {stage_id}") ++ started = time.perf_counter() ++ with self._lock: ++ self._next_token += 1 ++ token = self._next_token ++ self._active[token] = (stage_id, started, frame_index) ++ self._activations[stage_id] += 1 ++ if frame_index is not None: ++ self._last_frame_index = frame_index ++ try: ++ yield ++ finally: ++ finished = time.perf_counter() ++ with self._lock: ++ active = self._active.pop(token, None) ++ if active is not None: ++ self._elapsed_seconds[stage_id] += max(0.0, finished - active[1]) ++ ++ def snapshot(self) -> dict[str, Any]: ++ now = time.perf_counter() ++ with self._lock: ++ elapsed = dict(self._elapsed_seconds) ++ active_rows = list(self._active.values()) ++ for stage_id, started, _frame_index in active_rows: ++ elapsed[stage_id] += max(0.0, now - started) ++ total = sum(elapsed.values()) ++ active_stages = list( ++ dict.fromkeys(stage_id for stage_id, _started, _frame in active_rows) ++ ) ++ current_stage = active_rows[-1][0] if active_rows else None ++ return { ++ "current_stage": current_stage, ++ "active_stages": active_stages, ++ "active_frame_index": self._last_frame_index, ++ "stages": { ++ stage_id: { ++ "elapsed_seconds": round(elapsed[stage_id], 6), ++ "activations": self._activations[stage_id], ++ "share_percent": ( ++ round(elapsed[stage_id] / total * 100, 6) ++ if total > 0 ++ else None ++ ), ++ } ++ for stage_id in self._stage_ids ++ }, ++ } + + + def _send_client_binary_frame(stream: Any, payload: bytes) -> None: +@@ -571,6 +653,7 @@ + sensor_decode_ms: dict[str, list[float]], + result_queue: queue.Queue[bytes], + result_complete: threading.Event, ++ stage_telemetry: _StageExecutionTelemetry, + ) -> None: + import select + +@@ -632,7 +715,8 @@ + continue + if opcode != 0x2: + raise ShadowRuntimeError(f"unexpected websocket opcode: {opcode}") +- header, payload = _decode_event(frame) ++ with stage_telemetry.measure("source-ingress"): ++ header, payload = _decode_event(frame) + sequence = int(header["ingress_sequence"]) + if state.last_ingress_sequence is not None: + if sequence <= state.last_ingress_sequence: +@@ -883,7 +967,11 @@ + ) + + +-def run(args: argparse.Namespace, loaded: _LoadedModels | None = None) -> int: ++def run( ++ args: argparse.Namespace, ++ loaded: _LoadedModels | None = None, ++ runtime_state: dict[str, Any] | None = None, ++) -> int: + import av + import torch + import transformers +@@ -894,6 +982,15 @@ + token = sys.stdin.readline().strip() if args.token_stdin else args.token + if not token or len(token) < 40: + raise RuntimeError("LAB E15 shadow token is missing") ++ stage_telemetry = ( ++ runtime_state.get("_stage_telemetry") ++ if runtime_state is not None ++ else None ++ ) ++ if not isinstance(stage_telemetry, _StageExecutionTelemetry): ++ stage_telemetry = _StageExecutionTelemetry() ++ if runtime_state is not None: ++ runtime_state["_stage_telemetry"] = stage_telemetry + common = _common(args) if loaded is None else loaded.common + live = common["live"] + e14 = common["e14"] +@@ -982,6 +1079,10 @@ + height=600, + maximum_buffer_bytes=int(live["transport"]["maximum_media_buffer_bytes"]), + metadata_capacity=int(live["transport"]["camera_metadata_capacity"]), ++ measure_decode=lambda frame_index: stage_telemetry.measure( ++ "camera-decode", ++ frame_index, ++ ), + ) + transport = _TransportState(Counter(), Counter()) + result_queue: queue.Queue[bytes] = queue.Queue(maxsize=2) +@@ -1086,6 +1187,11 @@ + self.count += 1 + + completed_semantics = SemanticResultStream() ++ ++ def monitored_semantic_inference(image: Any) -> Any: ++ with stage_telemetry.measure("semantic-model"): ++ return infer_semantic(image) ++ + semantic_thread = threading.Thread( + target=semantic_worker, + kwargs={ +@@ -1094,7 +1200,7 @@ + "valid_mask": valid_mask, + "target_lut": target_lut, + "target_names": target_names, +- "infer": infer_semantic, ++ "infer": monitored_semantic_inference, + "latency": semantic_latency, + "completed": completed_semantics, + "failures": semantic_errors, +@@ -1125,6 +1231,7 @@ + "sensor_decode_ms": sensor_decode_ms, + "result_queue": result_queue, + "result_complete": result_complete, ++ "stage_telemetry": stage_telemetry, + }, + name="lab-e15-shadow-receiver", + daemon=True, +@@ -1157,10 +1264,21 @@ + ) + try: + detector_started = time.perf_counter() +- tensor = _preprocess(envelope.image, valid_mask, detector) +- output_tensor, _request_ms = _infer(args.triton_url, detector["model"], tensor) +- detections, _rejected = _detections(output_tensor, detector, valid_mask) +- tracks = tracker.update(detections, envelope.frame_index) ++ with stage_telemetry.measure("preprocessing", envelope.frame_index): ++ tensor = _preprocess(envelope.image, valid_mask, detector) ++ with stage_telemetry.measure("detector", envelope.frame_index): ++ output_tensor, _request_ms = _infer( ++ args.triton_url, ++ detector["model"], ++ tensor, ++ ) ++ detections, _rejected = _detections( ++ output_tensor, ++ detector, ++ valid_mask, ++ ) ++ with stage_telemetry.measure("tracking", envelope.frame_index): ++ tracks = tracker.update(detections, envelope.frame_index) + latency["detector_ms"].append( + (time.perf_counter() - detector_started) * 1000 + ) +@@ -1194,34 +1312,35 @@ + ).reshape((-1, 3)) + position = binding.pose.position_xyz + quaternion = binding.pose.orientation_xyzw +- projection_started = time.perf_counter() +- pixels, depths, source_indices, points_lidar = project_points( +- points_map, +- position, +- quaternion, +- projection, +- ) +- latency["projection_ms"].append( +- (time.perf_counter() - projection_started) * 1000 +- ) +- association_started = time.perf_counter() +- fusions = fuse_tracks( +- tracks=[_track_document(track) for track in tracks], +- semantic_map=current_semantic.mask, +- pixels=pixels, +- depths=depths, +- source_indices=source_indices, +- points_map=points_map, +- points_lidar=points_lidar, +- association=e14["association"], +- distance_history=history, +- completion_tracker=completion_tracker, +- sensor_position_map=position, +- session_seconds=frame_seconds, +- ) +- latency["association_ms"].append( +- (time.perf_counter() - association_started) * 1000 +- ) ++ with stage_telemetry.measure("sensor-fusion", envelope.frame_index): ++ projection_started = time.perf_counter() ++ pixels, depths, source_indices, points_lidar = project_points( ++ points_map, ++ position, ++ quaternion, ++ projection, ++ ) ++ latency["projection_ms"].append( ++ (time.perf_counter() - projection_started) * 1000 ++ ) ++ association_started = time.perf_counter() ++ fusions = fuse_tracks( ++ tracks=[_track_document(track) for track in tracks], ++ semantic_map=current_semantic.mask, ++ pixels=pixels, ++ depths=depths, ++ source_indices=source_indices, ++ points_map=points_map, ++ points_lidar=points_lidar, ++ association=e14["association"], ++ distance_history=history, ++ completion_tracker=completion_tracker, ++ sensor_position_map=position, ++ session_seconds=frame_seconds, ++ ) ++ latency["association_ms"].append( ++ (time.perf_counter() - association_started) * 1000 ++ ) + fusion_state = "fused" + fused_frames += 1 + fusion_state_counts[fusion_state] += 1 +@@ -1270,20 +1389,21 @@ + if temporal_stabilizer is None: + fusion_objects = raw_fusion_objects + else: +- temporal_started = time.perf_counter() +- fusion_objects = temporal_stabilizer.update( +- frame_index=envelope.frame_index, +- session_seconds=frame_seconds, +- objects=raw_fusion_objects, +- ) +- world = stabilize_world_state( +- world, +- fusion_objects, +- temporal_world_memory, +- ) +- latency["temporal_2d_3d_ms"].append( +- (time.perf_counter() - temporal_started) * 1000 +- ) ++ with stage_telemetry.measure("temporal-state", envelope.frame_index): ++ temporal_started = time.perf_counter() ++ fusion_objects = temporal_stabilizer.update( ++ frame_index=envelope.frame_index, ++ session_seconds=frame_seconds, ++ objects=raw_fusion_objects, ++ ) ++ world = stabilize_world_state( ++ world, ++ fusion_objects, ++ temporal_world_memory, ++ ) ++ latency["temporal_2d_3d_ms"].append( ++ (time.perf_counter() - temporal_started) * 1000 ++ ) + stabilized_cuboids += sum( + str(item.get("cuboid_status", "")).startswith("accepted-") + for item in fusion_objects +@@ -1351,28 +1471,29 @@ + quality=80, + optimize=False, + ) +- live_result = encode_live_perception_result( +- frame_index=envelope.frame_index, +- source_frame_index=int(envelope.timeline["source_frame_index"]), +- session_seconds=frame_seconds, +- captured_at_epoch_ns=int(envelope.timeline["captured_at_epoch_ns"]), +- image_jpeg=encoded_image.getvalue(), +- segmentation_mask=( +- current_semantic.mask +- if current_semantic is not None and semantic_status == "fresh" +- else None +- ), +- objects=fusion_objects, +- delivery=world["delivery"], +- ) +- try: +- result_queue.put_nowait(live_result) +- except queue.Full: +- with suppress(queue.Empty): +- result_queue.get_nowait() +- result_queue.task_done() +- transport.results_dropped += 1 +- result_queue.put_nowait(live_result) ++ with stage_telemetry.measure("result-publication", envelope.frame_index): ++ live_result = encode_live_perception_result( ++ frame_index=envelope.frame_index, ++ source_frame_index=int(envelope.timeline["source_frame_index"]), ++ session_seconds=frame_seconds, ++ captured_at_epoch_ns=int(envelope.timeline["captured_at_epoch_ns"]), ++ image_jpeg=encoded_image.getvalue(), ++ segmentation_mask=( ++ current_semantic.mask ++ if current_semantic is not None and semantic_status == "fresh" ++ else None ++ ), ++ objects=fusion_objects, ++ delivery=world["delivery"], ++ ) ++ try: ++ result_queue.put_nowait(live_result) ++ except queue.Full: ++ with suppress(queue.Empty): ++ result_queue.get_nowait() ++ result_queue.task_done() ++ transport.results_dropped += 1 ++ result_queue.put_nowait(live_result) + except Exception: + detector_failures += 1 + raise +@@ -1645,6 +1766,7 @@ + }, + "gpu_telemetry": gpu.summary(), + "runtime_telemetry": runtime_summary, ++ "stage_telemetry": stage_telemetry.snapshot(), + "process_cpu_seconds": time.process_time() - process_cpu_started, + "process_peak_rss_mib": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024, + "cuda_peak_memory_allocated_mib": torch.cuda.max_memory_allocated() / 2**20, +@@ -1816,7 +1938,13 @@ + loaded = _load_models(args, common) + model_load_seconds = time.perf_counter() - load_started + run_lock = threading.Lock() +- state = {"busy": False, "completed_runs": 0, "failed_runs": 0} ++ state = { ++ "busy": False, ++ "completed_runs": 0, ++ "failed_runs": 0, ++ "active_request_id": None, ++ "_stage_telemetry": _StageExecutionTelemetry(), ++ } + + class Handler(BaseHTTPRequestHandler): + server_version = "MissionCorePersistentPerception/1" +@@ -1844,6 +1972,7 @@ + if self.path != "/health": + self._send(404, {"ok": False, "error": "not-found"}) + return ++ stage_snapshot = state["_stage_telemetry"].snapshot() + self._send( + 200, + { +@@ -1853,6 +1982,11 @@ + "model_load_seconds": model_load_seconds, + "completed_runs": state["completed_runs"], + "failed_runs": state["failed_runs"], ++ "active_request_id": state["active_request_id"], ++ "active_frame_index": stage_snapshot["active_frame_index"], ++ "current_stage": stage_snapshot["current_stage"], ++ "active_stages": stage_snapshot["active_stages"], ++ "stage_metrics": stage_snapshot["stages"], + "authority": common["live"]["authority"], + "gpu": torch.cuda.get_device_name(), + }, +@@ -1882,7 +2016,9 @@ + request_id = str(document.get("request_id", "invalid")) + run_args = _persistent_run_arguments(args, document) + document["token"] = None +- exit_code = run(run_args, loaded) ++ state["active_request_id"] = request_id ++ state["_stage_telemetry"] = _StageExecutionTelemetry() ++ exit_code = run(run_args, loaded, state) + state["completed_runs"] += 1 + self._send( + 200, +@@ -1917,6 +2053,7 @@ + ) + finally: + state["busy"] = False ++ state["active_request_id"] = None + run_lock.release() + + server = ThreadingHTTPServer((args.listen_host, args.listen_port), Handler) diff --git a/src/k1link/web/system_telemetry_api.py b/src/k1link/web/system_telemetry_api.py index c8b4050..e0db2d7 100644 --- a/src/k1link/web/system_telemetry_api.py +++ b/src/k1link/web/system_telemetry_api.py @@ -474,11 +474,57 @@ def _agent_raw_document(document: dict[str, Any]) -> dict[str, Any]: } docker_stats: dict[str, dict[str, str]] = raw["docker_stats"] container_states: dict[str, dict[str, object]] = raw["container_states"] + perception: dict[str, Any] = raw["perception"] + pipeline_stage_metrics: dict[str, dict[str, Any]] = {} + pipeline_active_stages: set[str] = set() 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 + if measurement == "missioncore_pipeline": + stage_id = tags.get("stage_id") + if isinstance(stage_id, str) and stage_id: + stage_state = fields.get("stage_state") + if stage_state == "active": + pipeline_active_stages.add(stage_id) + pipeline_stage_metrics[stage_id] = { + "elapsed_seconds": _number(fields.get("elapsed_seconds")), + "activations": fields.get("activations"), + "share_percent": _number(fields.get("share_percent")), + } + for name in ( + "service_state", + "current_stage", + "active_request_id", + ): + value = fields.get(name) + if isinstance(value, str): + perception[name if name != "service_state" else "state"] = ( + value or None + ) + for name in ("active_frame_index", "completed_runs", "failed_runs"): + value = fields.get(name) + if isinstance(value, int) and not isinstance(value, bool): + perception[name] = value if value >= 0 else None + model_load_seconds = _number(fields.get("model_load_seconds")) + if model_load_seconds is not None: + perception["model_load_seconds"] = model_load_seconds + perception["collector_state"] = fields.get("collector_state") + else: + pipeline_payload = _mapping(payload.get("payload")) or payload + for name, value in pipeline_payload.items(): + if name == "stage_metrics": + for stage_id, metrics in _mapping(value).items(): + if isinstance(metrics, dict): + pipeline_stage_metrics[stage_id] = metrics + elif name == "active_stages": + pipeline_active_stages.update( + stage_id + for stage_id in _items(value) + if isinstance(stage_id, str) + ) + else: + perception[name] = value continue if sample.get("kind") == "runtime": runtime_payload = _mapping(payload.get("payload")) or payload @@ -610,6 +656,10 @@ def _agent_raw_document(document: dict[str, Any]) -> dict[str, Any]: "version": None, "uptime_seconds": uptime, } + if pipeline_stage_metrics: + perception["stage_metrics"] = pipeline_stage_metrics + if pipeline_active_stages: + perception["active_stages"] = sorted(pipeline_active_stages) return raw diff --git a/tests/test_system_telemetry_api.py b/tests/test_system_telemetry_api.py index b528020..a8b24a2 100644 --- a/tests/test_system_telemetry_api.py +++ b/tests/test_system_telemetry_api.py @@ -274,6 +274,57 @@ def test_worker_telemetry_prefers_ndc_container_names_during_migration( assert triton["canonical_name"] == "ndc-mission-core-triton" +def test_agent_pipeline_snapshot_merges_all_stage_series() -> None: + samples = [] + for stage_id, stage_state, elapsed, activations, share in ( + ("detector", "active", 2.5, 42, 62.5), + ("semantic-model", "waiting", 1.5, 11, 37.5), + ): + samples.append( + { + "node_id": EXPECTED_NODE_ID, + "kind": "pipeline", + "measurement": "missioncore_pipeline", + "observed_at_utc": "2026-07-28T12:00:00Z", + "payload": { + "name": "missioncore_pipeline", + "tags": {"stage_id": stage_id}, + "fields": { + "service_state": "busy", + "current_stage": "detector", + "active_request_id": "request-006", + "active_frame_index": 17, + "completed_runs": 3, + "failed_runs": 0, + "model_load_seconds": 10.5, + "collector_state": "live", + "stage_state": stage_state, + "elapsed_seconds": elapsed, + "activations": activations, + "share_percent": share, + }, + }, + } + ) + + raw = _agent_raw_document({"samples": samples}) + + assert raw["perception"]["state"] == "busy" + assert raw["perception"]["active_stages"] == ["detector"] + assert raw["perception"]["stage_metrics"] == { + "detector": { + "elapsed_seconds": 2.5, + "activations": 42, + "share_percent": 62.5, + }, + "semantic-model": { + "elapsed_seconds": 1.5, + "activations": 11, + "share_percent": 37.5, + }, + } + + def test_worker_telemetry_history_keeps_one_row_per_agent_observation( tmp_path: Path, ) -> None: diff --git a/tests/test_telemetry_normalizer_contract.py b/tests/test_telemetry_normalizer_contract.py index ab3c39e..4cd9ee1 100644 --- a/tests/test_telemetry_normalizer_contract.py +++ b/tests/test_telemetry_normalizer_contract.py @@ -181,6 +181,41 @@ def test_normalizer_preserves_native_pipeline_stage_tags() -> None: } +def test_normalizer_accepts_telegraf_pipeline_snapshot() -> None: + normalizer = _normalizer() + row = normalizer._normalize( + "mission-core/v1/contours/worker-006/agents/worker-006/pipeline", + json.dumps( + { + "name": "missioncore_pipeline", + "tags": { + "node_id": "DESKTOP-OPJ8J04", + "contour_id": "worker-006", + "agent_id": "worker-006", + "stage_id": "detector", + }, + "fields": { + "service_state": "busy", + "stage_state": "active", + "elapsed_seconds": 2.5, + "activations": 42, + }, + "timestamp": 1_785_179_600, + } + ).encode(), + ) + + assert row[1:8] == ( + "worker-006", + "worker-006", + "DESKTOP-OPJ8J04", + "pipeline", + "missioncore_pipeline", + '{"stage_id":"detector"}', + "telegraf.metric-json/v1", + ) + + def test_normalizer_rejects_oversized_payload_and_series_identity() -> None: normalizer = _normalizer() with pytest.raises(ValueError, match="1 MiB"):