NODEDC_MISSION_CORE/deploy/telemetry-plane/telegraf/Get-NdcMissionCorePipelineTelemetry.ps1

[CmdletBinding()]
param(
    [string]$ContainerName = "ndc-mission-core-perception-worker",
    [string]$HealthUrl = "http://127.0.0.1:18020/health",
    [string]$RuntimeSnapshot = "C:\ProgramData\NDC\MissionCore\telemetry-agent\perception\current.json"
)

$ErrorActionPreference = "Stop"
$stageIds = @(
    "source-ingress",
    "camera-decode",
    "preprocessing",
    "detector",
    "semantic-model",
    "sensor-fusion",
    "tracking",
    "temporal-state",
    "result-publication"
)
$health = $null
$collectorState = "unavailable"
# A fresh full-profile export supersedes the historical service. Presence of an
# expired/broken export is NOT permission to relabel a legacy model as current.
if (Test-Path -LiteralPath $RuntimeSnapshot -PathType Leaf) {
    $runtime = $null
    try {
        $item = Get-Item -LiteralPath $RuntimeSnapshot
        $age = ([DateTime]::UtcNow - $item.LastWriteTimeUtc).TotalSeconds
        if ($item.Length -le 8192 -and $age -ge -1 -and $age -le 5) {
            $stream = [IO.File]::Open($RuntimeSnapshot, 'Open', 'Read', 'ReadWrite')
            try {
                $bytes = New-Object byte[] 8193
                $count = $stream.Read($bytes, 0, $bytes.Length)
                if ($count -le 8192) {
                    $runtime = [Text.Encoding]::UTF8.GetString($bytes, 0, $count) | ConvertFrom-Json
                }
            } finally { $stream.Dispose() }
            if ($runtime.schema_version -ne 'missioncore.perception-runtime-observation/v1') {
                $runtime = $null
            }
        }
    } catch { $runtime = $null }
    $samples = foreach ($stageId in $stageIds) {
        $sample = [ordered]@{
            stage_id = $stageId
            stage_state = 'unavailable'
            service_state = 'unavailable'
            collector_state = 'unavailable'
        }
        if ($runtime) {
            $sample.collector_state = 'live'
            foreach ($name in @('service_state', 'profile_name', 'active_request_id',
                'input_state', 'wait_reason', 'live_children', 'input_pauses', 'pending_bundles', 'buffer_bytes')) {
                $sample[$name] = $runtime.$name
            }
        }
        [pscustomobject]$sample
    }
    @($samples) | ConvertTo-Json -Compress -Depth 4
    exit 0
}
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]-1
}
$failedRuns = if ($health -and $null -ne $health.failed_runs) {
    [int64]$health.failed_runs
}
else {
    [int64]-1
}
$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