From f3640026cfe9cacfbfa90e49dbb0d33002cbfa97 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Fri, 25 Sep 2026 23:24:27 +0300 Subject: [PATCH] fix(telemetry): recover worker agent and Tailscale transport automatically --- deploy/telemetry-plane/README.md | 31 +- deploy/telemetry-plane/prepare.py | 1 + .../Install-NdcMissionCoreTelegraf.ps1 | 9 + ...nstall-NdcMissionCoreTelemetryRecovery.ps1 | 87 +++++ .../Test-NdcMissionCoreTelemetryRecovery.ps1 | 26 ++ .../Update-NdcMissionCoreTelegraf.ps1 | 9 + scripts/check_telemetry_recovery.py | 99 ++++++ scripts/manage_telemetry_startup.py | 333 ++++++++++++++++++ src/k1link/web/compute_contour_api.py | 4 +- tests/test_compute_contour_api.py | 14 + tests/test_telemetry_startup.py | 146 ++++++++ 11 files changed, 755 insertions(+), 4 deletions(-) create mode 100644 deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelemetryRecovery.ps1 create mode 100644 deploy/telemetry-plane/telegraf/Test-NdcMissionCoreTelemetryRecovery.ps1 create mode 100644 scripts/check_telemetry_recovery.py create mode 100644 scripts/manage_telemetry_startup.py create mode 100644 tests/test_telemetry_startup.py diff --git a/deploy/telemetry-plane/README.md b/deploy/telemetry-plane/README.md index 00b72b9..a6b3cd3 100644 --- a/deploy/telemetry-plane/README.md +++ b/deploy/telemetry-plane/README.md @@ -149,9 +149,11 @@ globally unique because Mosquitto ACL ownership is username-based. ## Worker agent -MQTT outputs use `startup_error_behavior="retry"`, with a 2000-metric buffer and -the configured flush interval. A missing broker at agent startup must not end -the service. See [Telegraf's startup policy](https://docs.influxdata.com/telegraf/v1/configuration/plugin-options/). +MQTT outputs request `startup_error_behavior="retry"`, with a 2000-metric buffer +and the configured flush interval. This setting alone does not guarantee +recovery: Telegraf 1.38.4 can exit cleanly when MQTT is unavailable at startup. +The Tailscale profile below therefore also installs a system recovery task. +See [Telegraf's startup policy](https://docs.influxdata.com/telegraf/v1/configuration/plugin-options/). If Docker lost a published listener while the saved LAN address is unchanged, the explicit broker Apply action reconciles that listener; a telemetry GET never restarts infrastructure. Node connectivity does not prove a profile is ready. @@ -228,3 +230,26 @@ changing the K1 command sequence. The stack and agent are intentionally not started by repository tests. Provisioning a machine is a separate, explicit operation. + +## Tailscale operator profile and startup recovery + +The accepted 2026-09-25 operator profile no longer depends on a shared LAN or +DHCP address. A prepared Windows agent uses `127.0.0.1:1883`; the existing strict +Tailscale SSH identity carries a loopback reverse forward to the operator +broker. Operator deployment is owned by `scripts/manage_telemetry_startup.py` +(plan, hash-bound apply, rollback). This is a macOS **login** profile, not a +pre-login Docker daemon. Preserve the explicit prepared stack root and its +private credentials when moving the Core source checkout. + +For this profile the Windows install/update bundle includes and invokes +`Install-NdcMissionCoreTelemetryRecovery.ps1`: boot + once-per-minute SYSTEM +reconciliation for a stopped Telegraf service, independent of login. Its +maintenance/rollback boundary is described in the audit. Do not rely solely on +SCM failure actions or the MQTT `startup_error_behavior` setting: the pinned +1.38.4 release was observed to terminate with a clean service exit when the +broker was unavailable during boot. + +See [the measured recovery audit](../../docs/audits/2026-09-25-worker-telemetry-recovery.md) +for exact acceptance, remaining cold-boot/clean-host gates and bounded test +entrypoints. The earlier LAN workflow remains a legacy explicit configuration; +it is not the current Worker 006 transport. diff --git a/deploy/telemetry-plane/prepare.py b/deploy/telemetry-plane/prepare.py index 38cdc99..2a207e8 100644 --- a/deploy/telemetry-plane/prepare.py +++ b/deploy/telemetry-plane/prepare.py @@ -29,6 +29,7 @@ SAFE_IDENTIFIER: Final = re.compile( ) SAFE_INTERVAL: Final = re.compile(r"^[1-9][0-9]{0,2}s$") WINDOWS_BUNDLE_FILES: Final = ( + "Install-NdcMissionCoreTelemetryRecovery.ps1", "Get-NdcMissionCorePipelineTelemetry.ps1", "Install-NdcMissionCoreTelegraf.ps1", "Update-NdcMissionCoreTelegraf.ps1", diff --git a/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 b/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 index 2bb1145..71fcf0e 100644 --- a/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 +++ b/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelegraf.ps1 @@ -33,6 +33,10 @@ foreach ($name in @( Set-Item -Path "Env:$name" -Value ([string]$value) } +$recoveryInstaller = Join-Path $PSScriptRoot 'Install-NdcMissionCoreTelemetryRecovery.ps1' +if ($payload.MISSIONCORE_MQTT_HOST -eq '127.0.0.1' -and -not (Test-Path $recoveryInstaller -PathType Leaf)) { + throw 'Telemetry recovery installer missing from bundle' +} if (Get-Service -Name $serviceName -ErrorAction SilentlyContinue) { throw "Service '$serviceName' already exists; refusing an implicit replacement" } @@ -137,6 +141,10 @@ try { if ($LASTEXITCODE -ne 0) { throw "Failed to enable Telegraf recovery for non-crash failures" } + $recoveryTask = $null + if ($payload.MISSIONCORE_MQTT_HOST -eq '127.0.0.1') { + $recoveryTask = & $recoveryInstaller -ExpectedNodeId $env:COMPUTERNAME -Action Apply | ConvertFrom-Json + } Start-Service -Name $serviceName $service = Get-Service -Name $serviceName $service.WaitForStatus([ServiceProcess.ServiceControllerStatus]::Running, [TimeSpan]::FromSeconds(20)) @@ -150,6 +158,7 @@ try { Status = $service.Status.ToString() StartType = $service.StartType.ToString() RecoveryConfigured = $true + RecoveryTask = $recoveryTask Configuration = $configurationPath PipelineCollector = $collectorPath PipelineJournal = $PipelineJournal diff --git a/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelemetryRecovery.ps1 b/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelemetryRecovery.ps1 new file mode 100644 index 0000000..db48891 --- /dev/null +++ b/deploy/telemetry-plane/telegraf/Install-NdcMissionCoreTelemetryRecovery.ps1 @@ -0,0 +1,87 @@ +[CmdletBinding()] +param( + [Parameter(Mandatory = $true)][string]$ExpectedNodeId, + [ValidateSet('Plan','Apply','Rollback')][string]$Action = 'Plan', + [string]$BackupDirectory +) +# Telegraf 1.38.4 can report MQTT startup failure as a clean Windows-service exit. +# SCM failure actions alone cannot recover that stopped service. +$ErrorActionPreference = 'Stop' +$ProgressPreference = 'SilentlyContinue' +if ($env:COMPUTERNAME -cne $ExpectedNodeId) { throw 'Worker identity mismatch' } +$taskName = 'ndc-mission-core-telemetry-recovery' +$root = Join-Path $env:ProgramFiles 'NDC\Mission Core\TelemetryRecovery' +$guardPath = Join-Path $root 'Resume-Telemetry.ps1' +$service = Get-CimInstance Win32_Service -Filter "Name='telegraf'" +if (-not $service -or $service.PathName -notlike '*NDC\Mission Core\Telegraf\telegraf.exe*') { throw 'Managed Telegraf service required' } +$guard = @' +$ErrorActionPreference = 'Stop' +$service = Get-Service -Name 'telegraf' +if ($service.Status -eq 'Running') { exit 0 } +if ($service.Status -ne 'Stopped') { exit 0 } +$values = @{} +foreach ($entry in (Get-ItemProperty 'HKLM:\SYSTEM\CurrentControlSet\Services\telegraf').Environment) { + $name, $value = $entry -split '=', 2 + if ($name -in @('MISSIONCORE_MQTT_HOST','MISSIONCORE_MQTT_PORT')) { $values[$name] = $value } +} +if ($values['MISSIONCORE_MQTT_HOST'] -ne '127.0.0.1' -or $values['MISSIONCORE_MQTT_PORT'] -ne '1883') { exit 1 } +$client = [Net.Sockets.TcpClient]::new() +try { + $connect = $client.ConnectAsync('127.0.0.1', 1883) + if (-not $connect.Wait(3000) -or -not $client.Connected) { exit 0 } +} catch { exit 0 } finally { $client.Dispose() } +Start-Service -Name 'telegraf' +(Get-Service -Name 'telegraf').WaitForStatus([ServiceProcess.ServiceControllerStatus]::Running, [TimeSpan]::FromSeconds(15)) +'@ +if ($Action -eq 'Plan') { + [ordered]@{node=$env:COMPUTERNAME;task=$taskName;path=$guardPath;startMode=$service.StartMode;triggers=@('boot','every-minute');scope='Start stopped telemetry service only';requiresInteractiveLogin=$false}|ConvertTo-Json -Compress + exit 0 +} +if ($Action -eq 'Rollback') { + if (-not $BackupDirectory) { throw 'BackupDirectory required' } + $receipt = Get-Content (Join-Path $BackupDirectory 'receipt.json') -Raw | ConvertFrom-Json + if ($receipt.node -cne $ExpectedNodeId -or $receipt.task -ne $taskName) { throw 'Backup identity mismatch' } + if ((Get-FileHash $guardPath -Algorithm SHA256).Hash -ne $receipt.installedHash) { throw 'Installed recovery script changed' } + Unregister-ScheduledTask -TaskName $taskName -Confirm:$false -ErrorAction SilentlyContinue + if (Test-Path (Join-Path $BackupDirectory 'task.xml')) { + Register-ScheduledTask -TaskName $taskName -Xml (Get-Content (Join-Path $BackupDirectory 'task.xml') -Raw) -Force | Out-Null + } + if (Test-Path (Join-Path $BackupDirectory 'guard.ps1')) { + Copy-Item (Join-Path $BackupDirectory 'guard.ps1') $guardPath -Force + } else { Remove-Item $guardPath } + [ordered]@{restored=$true;task=$taskName}|ConvertTo-Json -Compress + exit 0 +} +$principal = [Security.Principal.WindowsPrincipal][Security.Principal.WindowsIdentity]::GetCurrent() +if (-not $principal.IsInRole([Security.Principal.WindowsBuiltInRole]::Administrator)) { throw 'Administrative installer required' } +if (Test-Path $root) { + if ((Get-Item $root -Force).Attributes -band [IO.FileAttributes]::ReparsePoint) { throw 'Recovery directory must not be a reparse point' } +} +New-Item -ItemType Directory -Path $root -Force | Out-Null +# SYSTEM executes this file: only SYSTEM and Administrators may modify it. +& icacls.exe $root /inheritance:r /grant:r '*S-1-5-18:(OI)(CI)F' '*S-1-5-32-544:(OI)(CI)F' | Out-Null +if ($LASTEXITCODE -ne 0) { throw 'Recovery directory ACL failed' } +$backup = Join-Path $root ('backups\' + [Guid]::NewGuid().ToString('N')) +New-Item -ItemType Directory -Path $backup -Force | Out-Null +$previousTask = Get-ScheduledTask -TaskName $taskName -ErrorAction SilentlyContinue +if ($previousTask) { Export-ScheduledTask -TaskName $taskName | Set-Content (Join-Path $backup 'task.xml') -Encoding UTF8 } +if (Test-Path $guardPath) { Copy-Item $guardPath (Join-Path $backup 'guard.ps1') } +try { + [IO.File]::WriteAllText($guardPath, $guard, [Text.UTF8Encoding]::new($false)) + $exe = Join-Path $env:SystemRoot 'System32\WindowsPowerShell\v1.0\powershell.exe' + $run = New-ScheduledTaskAction -Execute $exe -Argument ('-NoLogo -NoProfile -NonInteractive -ExecutionPolicy Bypass -File "' + $guardPath + '"') + $triggers = @((New-ScheduledTaskTrigger -AtStartup), (New-ScheduledTaskTrigger -Once -At (Get-Date).AddMinutes(1) -RepetitionInterval (New-TimeSpan -Minutes 1))) + $settings = New-ScheduledTaskSettingsSet -StartWhenAvailable -MultipleInstances IgnoreNew -ExecutionTimeLimit (New-TimeSpan -Seconds 45) -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries + $identity = New-ScheduledTaskPrincipal -UserId 'SYSTEM' -LogonType ServiceAccount -RunLevel Highest + Register-ScheduledTask -TaskName $taskName -Action $run -Trigger $triggers -Settings $settings -Principal $identity -Force | Out-Null + $receipt = [ordered]@{node=$env:COMPUTERNAME;task=$taskName;installedHash=(Get-FileHash $guardPath -Algorithm SHA256).Hash;backup=$backup} + $receipt|ConvertTo-Json|Set-Content (Join-Path $backup 'receipt.json') -Encoding UTF8 + Start-ScheduledTask -TaskName $taskName + $receipt|ConvertTo-Json -Compress +} catch { + if ($previousTask) { Register-ScheduledTask -TaskName $taskName -Xml (Get-Content (Join-Path $backup 'task.xml') -Raw) -Force | Out-Null } + else { Unregister-ScheduledTask -TaskName $taskName -Confirm:$false -ErrorAction SilentlyContinue } + if (Test-Path (Join-Path $backup 'guard.ps1')) { Copy-Item (Join-Path $backup 'guard.ps1') $guardPath -Force } + else { Remove-Item $guardPath -ErrorAction SilentlyContinue } + throw +} diff --git a/deploy/telemetry-plane/telegraf/Test-NdcMissionCoreTelemetryRecovery.ps1 b/deploy/telemetry-plane/telegraf/Test-NdcMissionCoreTelemetryRecovery.ps1 new file mode 100644 index 0000000..d9ada29 --- /dev/null +++ b/deploy/telemetry-plane/telegraf/Test-NdcMissionCoreTelemetryRecovery.ps1 @@ -0,0 +1,26 @@ +[CmdletBinding()] +param([Parameter(Mandatory=$true)][string]$ExpectedNodeId) +$ErrorActionPreference='Stop' +$ProgressPreference='SilentlyContinue' +if ($env:COMPUTERNAME -cne $ExpectedNodeId) { throw 'Worker identity mismatch' } +$task=Get-ScheduledTask -TaskName 'ndc-mission-core-telemetry-recovery' +$account=[Security.Principal.NTAccount]::new($task.Principal.UserId) +$sid=$account.Translate([Security.Principal.SecurityIdentifier]).Value +if (-not $task.Settings.Enabled -or $sid -ne 'S-1-5-18') { throw 'Recovery task not enabled as SYSTEM' } +$before=Get-CimInstance Win32_Service -Filter "Name='telegraf'" +if ($before.State -ne 'Running') { throw 'A running baseline is required' } +$started=[DateTime]::UtcNow +Stop-Service telegraf +(Get-Service telegraf).WaitForStatus([ServiceProcess.ServiceControllerStatus]::Stopped,[TimeSpan]::FromSeconds(20)) +$automatic=$false +try { + while (([DateTime]::UtcNow-$started).TotalSeconds -lt 80) { + Start-Sleep -Seconds 2 + $after=Get-CimInstance Win32_Service -Filter "Name='telegraf'" + if ($after.State -eq 'Running' -and $after.ProcessId -ne $before.ProcessId) { $automatic=$true;break } + } + [ordered]@{node=$ExpectedNodeId;test='clean-service-stop';automaticRecovery=$automatic;elapsedSeconds=[Math]::Round(([DateTime]::UtcNow-$started).TotalSeconds,2);previousPid=$before.ProcessId;newPid=$after.ProcessId;state=$after.State}|ConvertTo-Json -Compress + if (-not $automatic) { throw 'Automatic recovery failed; restoring baseline' } +} finally { + if ((Get-Service telegraf).Status -ne 'Running') { Start-Service telegraf } +} diff --git a/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 b/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 index 67949bc..ec108be 100644 --- a/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 +++ b/deploy/telemetry-plane/telegraf/Update-NdcMissionCoreTelegraf.ps1 @@ -67,6 +67,10 @@ if (-not ($serviceEnvironmentCandidate | Where-Object { ) } +$recoveryInstaller = Join-Path $PSScriptRoot 'Install-NdcMissionCoreTelemetryRecovery.ps1' +if ($env:MISSIONCORE_MQTT_HOST -eq '127.0.0.1' -and -not (Test-Path $recoveryInstaller -PathType Leaf)) { + throw 'Telemetry recovery installer missing from bundle' +} $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" @@ -127,6 +131,10 @@ try { if ($LASTEXITCODE -ne 0) { throw "Failed to enable Telegraf recovery for non-crash failures" } + $recoveryTask = $null + if ($env:MISSIONCORE_MQTT_HOST -eq '127.0.0.1') { + $recoveryTask = & $recoveryInstaller -ExpectedNodeId $env:COMPUTERNAME -Action Apply | ConvertFrom-Json + } } catch { Stop-Service -Name $serviceName -Force -ErrorAction SilentlyContinue @@ -155,6 +163,7 @@ try { ServiceName = $serviceName Status = (Get-Service -Name $serviceName).Status.ToString() RecoveryConfigured = $true + RecoveryTask = $recoveryTask Configuration = $configurationPath Backup = $backupPath PipelineCollector = $collectorPath diff --git a/scripts/check_telemetry_recovery.py b/scripts/check_telemetry_recovery.py new file mode 100644 index 0000000..5539010 --- /dev/null +++ b/scripts/check_telemetry_recovery.py @@ -0,0 +1,99 @@ +"""Bounded operator acceptance of installed telemetry recovery, no inference jobs. + +Each case injects one failure in an exact Mission Core telemetry component. +An existing service's automatic supervisor must restore fresh worker telemetry. +""" + +import argparse +import json +import os +import re +import subprocess +import time +import urllib.request +from datetime import datetime + +DOCKER = "/Applications/Docker.app/Contents/Resources/bin/docker" +BASE = "http://127.0.0.1:8000/api/v1/system/contours/worker-006/telemetry?history=1" + + +def read(): + return json.load(urllib.request.urlopen(BASE, timeout=5)) + + +def tunnel_pid(): + r = subprocess.run( + ["launchctl", "print", f"gui/{os.getuid()}/com.nodedc.telemetry-tunnel.local"], + capture_output=True, + text=True, + ) + m = re.search(r"\bpid = (\d+)", r.stdout) + return int(m.group(1)) if m else None + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("case", choices=["tunnel", "receiver"]) + args = parser.parse_args() + baseline = read()["connection"] + if not baseline["reachable"] or not baseline["identity_matches"]: + raise RuntimeError("Fresh confirmed baseline required") + started = time.monotonic() + baseline_time = datetime.fromisoformat(baseline["observed_at_utc"].replace("Z", "+00:00")) + before_pid = tunnel_pid() + if args.case == "tunnel": + subprocess.run( + [ + "launchctl", + "kill", + "SIGKILL", + f"gui/{os.getuid()}/com.nodedc.telemetry-tunnel.local", + ], + check=True, + capture_output=True, + ) + else: + subprocess.run( + [DOCKER, "stop", "--time", "5", "ndc-mission-core-telemetry-normalizer"], + check=True, + capture_output=True, + timeout=15, + ) + seen = set() + while time.monotonic() - started < 90: + time.sleep(2) + try: + probe = read()["connection"] + code = probe.get("error_code") + seen.add(code or "fresh") + observed = datetime.fromisoformat(probe["observed_at_utc"].replace("Z", "+00:00")) + if ( + probe["reachable"] + and probe["identity_matches"] + and (observed - baseline_time).total_seconds() >= 10 + and time.monotonic() - started >= 10 + and ( + args.case != "tunnel" + or (tunnel_pid() is not None and tunnel_pid() != before_pid) + ) + and (args.case != "receiver" or "telemetry-receiver-unavailable" in seen) + ): + print( + json.dumps( + { + "case": args.case, + "automaticRecovery": True, + "elapsedSeconds": round(time.monotonic() - started, 2), + "observedStates": sorted(seen), + "observedAt": probe["observed_at_utc"], + } + ) + ) + return + except (OSError, ValueError): + seen.add("query-unavailable") + raise RuntimeError("Automatic telemetry recovery not accepted within 90 seconds") + + +if __name__ == "__main__": + main() diff --git a/scripts/manage_telemetry_startup.py b/scripts/manage_telemetry_startup.py new file mode 100644 index 0000000..d00abd3 --- /dev/null +++ b/scripts/manage_telemetry_startup.py @@ -0,0 +1,333 @@ +#!/usr/bin/env python3 +"""Versioned macOS telemetry startup: plan/apply/rollback, no worker job launch. + +The prepared stack owns credentials and volumes. SSH provides only a loopback +MQTT forward through an existing strictly pinned Tailscale SSH profile. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import plistlib +import re +import subprocess +import time +import urllib.request +from pathlib import Path + +from manage_mission_core_launch_agent import _wait_for_health, _write_atomic +from migrate_worker_tunnel_logs import reload_agent + +from k1link.launchd_logs import launchd_log_path, prepare_launchd_log +from k1link.web.compute_contour_network import _replace_environment_value + +STARTUP = "com.nodedc.telemetry-startup.local" +TUNNEL = "com.nodedc.telemetry-tunnel.local" +CORE = "com.nodedc.mission-core.local" +DOCKER = "/Applications/Docker.app/Contents/Resources/bin/docker" + + +def command(args, timeout=30): + # Never echo Compose output: expanded environments can contain credentials. + return subprocess.run(args, capture_output=True, timeout=timeout, check=False) + + +def receiver_ready(): + try: + with urllib.request.urlopen("http://127.0.0.1:18030/health", timeout=3) as r: + d = json.loads(r.read()) + return ( + d.get("ok") is True + and d.get("mqtt_connected") is True + and d.get("database_reachable") is True + ) + except (OSError, ValueError): + return False + + +def broker_loopback(): + result = command( + [ + DOCKER, + "inspect", + "--format", + "{{json .NetworkSettings.Ports}}", + "ndc-mission-core-mqtt-broker", + ], + 10, + ) + try: + bindings = json.loads(result.stdout).get("1883/tcp", []) + return result.returncode == 0 and bindings == [{"HostIp": "127.0.0.1", "HostPort": "1883"}] + except (ValueError, AttributeError): + return False + + +def reconcile(stack: Path): + """One bounded retry, repeated by launchd after delayed Docker/network start.""" + if receiver_ready() and broker_loopback(): + return "ready" + if command([DOCKER, "info", "--format", "{{.ServerVersion}}"], 10).returncode: + # Desktop's own login setting is not a dependency manager. No VM/resource + # setting is changed and no user inference job is started here. + command(["/usr/bin/open", "-g", "-a", "/Applications/Docker.app"], 10) + return "waiting-for-docker" + compose = [ + DOCKER, + "compose", + "--project-directory", + str(stack), + "--env-file", + str(stack / ".env"), + "-f", + str(stack / "compose.yaml"), + ] + if command([*compose, "config", "--quiet"], 20).returncode: + return "configuration-unavailable" + if command( + [ + *compose, + "up", + "-d", + "--no-build", + "--pull", + "never", + "--wait", + "--wait-timeout", + "45", + "broker", + "timescale", + "normalizer", + ], + 60, + ).returncode: + return "waiting-for-receiver" + return "ready" if receiver_ready() and broker_loopback() else "waiting-for-receiver" + + +def private_file(path): + if path.is_symlink() or not path.is_file() or path.stat().st_uid != os.getuid(): + raise ValueError("Expected owned regular configuration file") + return path.read_bytes() + + +def plan(stack: Path, ssh_alias: str, repository: Path, agents: Path): + if not stack.is_absolute() or stack.resolve() != stack or not repository.is_absolute(): + raise ValueError("Canonical absolute installation paths required") + if re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_.-]{0,80}", ssh_alias) is None: + raise ValueError("Invalid SSH profile") + for name in ( + "compose.yaml", + "runtime/agents.json", + "runtime/mosquitto/acl", + "runtime/mosquitto/passwords", + ): + private_file(stack / name) + env = private_file(stack / ".env") + public = dict( + line.split("=", 1) + for line in env.decode().splitlines() + if "=" in line and not line.startswith("#") + ) + if ( + public.get("MISSIONCORE_MQTT_PORT", "1883") != "1883" + or public.get("MISSIONCORE_TELEMETRY_QUERY_PORT", "18030") != "18030" + ): + raise ValueError("This deployment profile requires MQTT 1883 and query 18030") + # Verify the existing trust boundary; never enroll a host or accept a new key. + ssh = command(["/usr/bin/ssh", "-G", ssh_alias]) + settings = dict(line.split(" ", 1) for line in ssh.stdout.decode().splitlines() if " " in line) + if ( + ssh.returncode + or settings.get("stricthostkeychecking") not in ("true", "yes") + or not settings.get("hostname", "").endswith(".ts.net") + ): + raise ValueError("A strictly pinned Tailscale SSH profile is required") + python = repository / ".venv/bin/python" + if not python.is_file() or not Path(DOCKER).is_file(): + raise ValueError("Prepared application Python and Docker Desktop are required") + common = dict( + RunAtLoad=True, AbandonProcessGroup=False, ProcessType="Background", ExitTimeOut=15 + ) + startup = dict( + common, + Label=STARTUP, + StartInterval=30, + ProgramArguments=[ + str(python), + str(repository / "scripts/manage_telemetry_startup.py"), + "reconcile", + "--stack-root", + str(stack), + ], + StandardOutPath="/dev/null", + StandardErrorPath=str(launchd_log_path("telemetry-startup.log")), + ) + tunnel = dict( + common, + Label=TUNNEL, + KeepAlive=True, + ThrottleInterval=10, + ProgramArguments=[ + "/usr/bin/ssh", + "-o", + "BatchMode=yes", + "-o", + "StrictHostKeyChecking=yes", + "-o", + "ExitOnForwardFailure=yes", + "-o", + "ConnectTimeout=8", + "-o", + "ServerAliveInterval=10", + "-o", + "ServerAliveCountMax=3", + "-N", + "-T", + "-R", + "127.0.0.1:1883:127.0.0.1:1883", + ssh_alias, + ], + StandardOutPath="/dev/null", + StandardErrorPath=str(launchd_log_path("telemetry-tunnel.log")), + ) + core_path = agents / (CORE + ".plist") + core = plistlib.loads(private_file(core_path)) + if core.get("Label") != CORE or core.get("WorkingDirectory") != str(repository): + raise ValueError("Canonical Mission Core installation changed") + core["EnvironmentVariables"]["MISSIONCORE_TELEMETRY_PLANE_ROOT"] = str(stack) + desired = { + stack / ".env": _replace_environment_value( + env.decode(), "MISSIONCORE_MQTT_BIND_ADDRESS", "127.0.0.1" + ).encode() + } + for label, doc in ((STARTUP, startup), (TUNNEL, tunnel), (CORE, core)): + path = agents / (label + ".plist") + if path.exists() and plistlib.loads(private_file(path)).get("Label") != label: + raise ValueError("Foreign LaunchAgent at target path") + desired[path] = plistlib.dumps(doc, sort_keys=True) + changes = [] + for path, data in desired.items(): + before = private_file(path) if path.exists() else None + changes.append( + dict( + path=str(path), + before=hashlib.sha256(before).hexdigest() if before is not None else None, + after=hashlib.sha256(data).hexdigest(), + ) + ) + document = dict( + schema_version="missioncore.telemetry-startup-plan/v1", + changes=changes, + stack_root=str(stack), + ssh_profile=ssh_alias, + mqtt="loopback-over-ssh-tailscale", + startup="macOS user login; retry every 30s", + worker_jobs_started=False, + ) + document["artifact_sha256"] = hashlib.sha256(Path(__file__).read_bytes()).hexdigest() + document["compose_sha256"] = hashlib.sha256((stack / "compose.yaml").read_bytes()).hexdigest() + document["sha256"] = hashlib.sha256(json.dumps(document, sort_keys=True).encode()).hexdigest() + return document, desired + + +def restore(backup: Path, agents: Path): + manifest = json.loads(private_file(backup / "manifest.json")) + for item in manifest["changes"]: + path = Path(item["path"]) + current = hashlib.sha256(private_file(path)).hexdigest() if path.exists() else None + if current not in (item["before"], item["after"]): + raise ValueError("Installed configuration changed; refusing rollback") + for label in (STARTUP, TUNNEL): + command(["launchctl", "bootout", f"gui/{os.getuid()}/{label}"]) + for i, item in enumerate(manifest["changes"]): + path = Path(item["path"]) + if item["before"] is None: + path.unlink(missing_ok=True) + else: + old = private_file(backup / str(i)) + if hashlib.sha256(old).hexdigest() != item["before"]: + raise ValueError("Rollback payload changed") + _write_atomic(path, old) + for label in (STARTUP, TUNNEL, CORE): + path = agents / (label + ".plist") + if path.exists(): + reload_agent(path, label) + if not _wait_for_health(45): + raise RuntimeError("Core health not accepted after rollback") + return { + "restored": True, + "container_state": "not changed; declared endpoint restored on next reconciliation", + } + + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("action", choices=["plan", "apply", "reconcile", "rollback"]) + p.add_argument("--stack-root", type=Path) + p.add_argument("--repository-root", type=Path, default=Path(__file__).resolve().parents[1]) + p.add_argument("--ssh-alias", default="mission-gpu") + p.add_argument("--expected-sha256") + p.add_argument("--backup", type=Path) + a = p.parse_args() + agents = Path.home() / "Library/LaunchAgents" + if a.action == "rollback": + if a.backup is None: + p.error("--backup required") + print(json.dumps(restore(a.backup, agents))) + return + if a.stack_root is None: + p.error("--stack-root required") + if a.action == "reconcile": + try: + phase = reconcile(a.stack_root) + except (OSError, subprocess.SubprocessError, ValueError): + phase = "receiver-unavailable" + # No repeating stdout log; health remains independently observable. + print(json.dumps({"phase": phase})) + return + document, desired = plan(a.stack_root, a.ssh_alias, a.repository_root, agents) + if a.action == "plan": + print(json.dumps(document, indent=2)) + return + if a.expected_sha256 != document["sha256"]: + raise ValueError("Startup plan changed before apply") + backup = ( + a.repository_root + / ".runtime/mission-core/telemetry-service-backups" + / (str(time.time_ns())) + ) + backup.mkdir(parents=True, mode=0o700) + _write_atomic(backup / "manifest.json", json.dumps(document).encode()) + for i, path in enumerate(desired): + if path.exists(): + _write_atomic(backup / str(i), private_file(path)) + for name in ("telemetry-startup.log", "telemetry-tunnel.log"): + prepare_launchd_log(launchd_log_path(name)) + try: + for path, data in desired.items(): + _write_atomic(path, data) + for label in (STARTUP, TUNNEL, CORE): + reload_agent(agents / (label + ".plist"), label) + if not _wait_for_health(45): + raise RuntimeError("Canonical Core failed health acceptance") + except BaseException: + restore(backup, agents) + raise + print( + json.dumps( + { + "installed": True, + "backup": str(backup), + "plan_sha256": document["sha256"], + "telemetry_acceptance": "pending fresh worker sample", + } + ) + ) + + +if __name__ == "__main__": + main() diff --git a/src/k1link/web/compute_contour_api.py b/src/k1link/web/compute_contour_api.py index 8a7b0b7..914c0c0 100644 --- a/src/k1link/web/compute_contour_api.py +++ b/src/k1link/web/compute_contour_api.py @@ -423,7 +423,9 @@ def build_compute_contour_router(*, root_provider: RootProvider) -> APIRouter: try: contour = store.get(contour_id) telemetry_plane_root = ( - Path(__file__).resolve().parents[3] / "deploy" / "telemetry-plane" + Path(os.environ["MISSIONCORE_TELEMETRY_PLANE_ROOT"]) + if os.environ.get("MISSIONCORE_TELEMETRY_PLANE_ROOT") + else Path(__file__).resolve().parents[3] / "deploy" / "telemetry-plane" ) return apply_broker_network( _network_target(contour), diff --git a/tests/test_compute_contour_api.py b/tests/test_compute_contour_api.py index d14ec0b..2b9a540 100644 --- a/tests/test_compute_contour_api.py +++ b/tests/test_compute_contour_api.py @@ -144,3 +144,17 @@ def test_contour_router_returns_404_for_unknown_contour(tmp_path: Path) -> None: with pytest.raises(HTTPException) as error: install("missing") assert error.value.status_code == 404 + + +def test_broker_apply_uses_explicit_installation_root(tmp_path, monkeypatch): + from k1link.web import compute_contour_api as api + root = tmp_path / 'prepared-stack' + monkeypatch.setenv('MISSIONCORE_TELEMETRY_PLANE_ROOT', str(root)) + seen = [] + def apply(target, installation): + seen.append(installation) + return {'ready': True} + monkeypatch.setattr(api, 'apply_broker_network', apply) + router = api.build_compute_contour_router(root_provider=lambda: tmp_path / 'system') + _endpoint(router, '/api/v1/system/contours/{contour_id}/network/broker', 'POST')('worker-006') + assert seen == [root] diff --git a/tests/test_telemetry_startup.py b/tests/test_telemetry_startup.py new file mode 100644 index 0000000..d2246ce --- /dev/null +++ b/tests/test_telemetry_startup.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +import importlib.util +import json +import plistlib +import subprocess +import sys +from pathlib import Path + +import pytest + +SCRIPTS = Path(__file__).parents[1] / "scripts" +sys.path.insert(0, str(SCRIPTS)) +try: + spec = importlib.util.spec_from_file_location( + "telemetry_startup", SCRIPTS / "manage_telemetry_startup.py" + ) + manager = importlib.util.module_from_spec(spec) + spec.loader.exec_module(manager) +finally: + sys.path.pop(0) + + +def setup(tmp_path, monkeypatch): + stack = tmp_path / "state" + repo = tmp_path / "release" + agents = tmp_path / "agents" + for p in (stack, repo, agents): + p.mkdir() + for name in ( + "compose.yaml", + "runtime/agents.json", + "runtime/mosquitto/acl", + "runtime/mosquitto/passwords", + ): + p = stack / name + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text("fixture") + (stack / ".env").write_text("MISSIONCORE_MQTT_BIND_ADDRESS=192.168.1.5\nSECRET=preserve-me\n") + (repo / ".venv/bin").mkdir(parents=True) + (repo / ".venv/bin/python").write_text("fixture") + docker = tmp_path / "docker" + docker.write_text("fixture") + monkeypatch.setattr(manager, "DOCKER", str(docker)) + (agents / (manager.CORE + ".plist")).write_bytes( + plistlib.dumps( + { + "Label": manager.CORE, + "WorkingDirectory": str(repo), + "EnvironmentVariables": {"MISSIONCORE_DATA_DIR": "/existing-data"}, + } + ) + ) + monkeypatch.setattr( + manager, + "command", + lambda *args: subprocess.CompletedProcess( + [], 0, b"hostname worker.example.ts.net\nstricthostkeychecking true\n", b"" + ), + ) + return stack, repo, agents + + +def test_plan_preserves_credentials_and_pins_loopback_transport(tmp_path, monkeypatch): + stack, repo, agents = setup(tmp_path, monkeypatch) + before = (stack / ".env").read_bytes() + doc, files = manager.plan(stack, "worker-test", repo, agents) + assert (stack / ".env").read_bytes() == before + assert files[stack / ".env"] == b"MISSIONCORE_MQTT_BIND_ADDRESS=127.0.0.1\nSECRET=preserve-me\n" + assert "preserve-me" not in json.dumps(doc) + startup = plistlib.loads(files[agents / (manager.STARTUP + ".plist")]) + assert startup["RunAtLoad"] and startup["StartInterval"] == 30 + tunnel = plistlib.loads(files[agents / (manager.TUNNEL + ".plist")]) + assert "127.0.0.1:1883:127.0.0.1:1883" in tunnel["ProgramArguments"] + assert "StrictHostKeyChecking=yes" in tunnel["ProgramArguments"] + assert tunnel["KeepAlive"] and tunnel["ThrottleInterval"] >= 5 + core = plistlib.loads(files[agents / (manager.CORE + ".plist")]) + assert core["EnvironmentVariables"]["MISSIONCORE_DATA_DIR"] == "/existing-data" + assert core["EnvironmentVariables"]["MISSIONCORE_TELEMETRY_PLANE_ROOT"] == str(stack) + + +def test_plan_refuses_non_tailscale_and_changed_input(tmp_path, monkeypatch): + stack, repo, agents = setup(tmp_path, monkeypatch) + before, _ = manager.plan(stack, "worker-test", repo, agents) + with (stack / ".env").open("a") as f: + f.write("NEW=setting\n") + after, _ = manager.plan(stack, "worker-test", repo, agents) + assert before["sha256"] != after["sha256"] + monkeypatch.setattr( + manager, + "command", + lambda *args: subprocess.CompletedProcess( + [], 0, b"hostname worker.local\nstricthostkeychecking true\n", b"" + ), + ) + with pytest.raises(ValueError, match="Tailscale"): + manager.plan(stack, "worker-test", repo, agents) + + +def test_delayed_docker_retries_without_compose_or_workload_launch(monkeypatch, tmp_path): + calls = [] + monkeypatch.setattr(manager, "receiver_ready", lambda: False) + + def run(args, timeout): + calls.append(args) + return subprocess.CompletedProcess(args, 1 if "info" in args else 0) + + monkeypatch.setattr(manager, "command", run) + assert manager.reconcile(tmp_path) == "waiting-for-docker" + assert len(calls) == 2 and calls[1][0] == "/usr/bin/open" + assert not any("compose" in c for c in calls) + + +def test_ready_receiver_is_not_restarted(monkeypatch, tmp_path): + monkeypatch.setattr(manager, "receiver_ready", lambda: True) + monkeypatch.setattr(manager, "broker_loopback", lambda: True) + monkeypatch.setattr( + manager, "command", lambda *args: pytest.fail("Must not restart healthy services") + ) + assert manager.reconcile(tmp_path) == "ready" + + +def test_old_lan_binding_is_reconciled_even_when_receiver_healthy(monkeypatch, tmp_path): + calls = [] + bindings = iter((False, True)) + monkeypatch.setattr(manager, "receiver_ready", lambda: True) + monkeypatch.setattr(manager, "broker_loopback", lambda: next(bindings)) + + def run(args, timeout): + calls.append(args) + return subprocess.CompletedProcess(args, 0) + + monkeypatch.setattr(manager, "command", run) + assert manager.reconcile(tmp_path) == "ready" + assert calls[-1][-3:] == ["broker", "timescale", "normalizer"] + assert "--no-build" in calls[-1] and "--pull" in calls[-1] and "never" in calls[-1] + + +def test_failed_compose_reports_waiting_without_ready(monkeypatch, tmp_path): + monkeypatch.setattr(manager, "receiver_ready", lambda: False) + monkeypatch.setattr( + manager, + "command", + lambda args, timeout: subprocess.CompletedProcess(args, int("up" in args)), + ) + assert manager.reconcile(tmp_path) == "waiting-for-receiver"