diff --git a/apps/device-manager/server/device-core-client.mjs b/apps/device-manager/server/device-core-client.mjs index 830ce9a..4b496e5 100644 --- a/apps/device-manager/server/device-core-client.mjs +++ b/apps/device-manager/server/device-core-client.mjs @@ -204,7 +204,10 @@ export function createLocalPreviewDeviceCore({ fixture = null } = {}) { }, assets: projectValues(assets, projectRef), assetBindings: projectValues(assetBindings, projectRef), - hosts: projectHosts.map((host) => withHealth("host", host, host.hostRef)), + hosts: projectHosts.map((host) => ({ + ...withHealth("host", host, host.hostRef), + telemetry: host.telemetry ?? emptyPreviewHostTelemetry(), + })), endpoints: projectValues(endpoints, projectRef), deployments: projectValues(deployments, projectRef), serviceInstances: projectServices.map((service) => @@ -245,6 +248,10 @@ export function createLocalPreviewDeviceCore({ fixture = null } = {}) { routes, sessions, configurationStates, + hosts, + deployments, + serviceInstances, + healthObservations, }); else if (fixture != null && fixture !== "") { throw serviceError("device_manager_preview_fixture_invalid", 400); @@ -890,6 +897,88 @@ function previewOntology(entityId) { return { entityId, catalogHash: "229c61c02a790906" }; } +function emptyPreviewHostTelemetry() { + return { + state: "unobserved", + freshness: "missing", + observedAt: null, + receivedAt: null, + expiresAt: null, + current: null, + history: [], + observation: null, + }; +} + +function previewHostTelemetry(hostRef, serviceInstanceRef, edgeRef) { + const now = Date.now(); + const history = Array.from({ length: 60 }, (_, index) => { + const observedAt = new Date(now - (59 - index) * 2_000).toISOString(); + const phase = index / 7; + return { + observedAt, + cpuUsagePercent: 18 + Math.sin(phase) * 8 + (index % 5), + memoryUsedPercent: 42 + Math.sin(phase / 2) * 3, + network: [{ + interface: "eth0", + bytesReceived: 8_000_000 + index * (32_000 + (index % 4) * 4_000), + bytesSent: 3_000_000 + index * (14_000 + (index % 3) * 2_000), + }], + }; + }); + const observedAt = history.at(-1).observedAt; + const current = { + schemaVersion: "nodedc.infrastructure.host-telemetry.v1", + profile: "linux-host-telegraf-v1", + hostKey: "robot2b-b2-edge-vps", + observedAt, + source: { + agent: "telegraf", + agentVersion: "1.38.4", + collectorRef: "service:nodedc-host-telemetry-agent", + }, + hardware: { + hostname: "koffyvngij", + architecture: "x64", + platform: "linux", + kernelRelease: "6.8.0-79-generic", + cpuModel: "AMD EPYC Processor (KVM)", + logicalProcessors: 1, + }, + cpu: { usagePercent: history.at(-1).cpuUsagePercent, load1: 0.31, load5: 0.24, load15: 0.18 }, + memory: { totalBytes: 1_007_681_536, availableBytes: 570_425_344, freeBytes: null, usedBytes: 437_256_192, usedPercent: history.at(-1).memoryUsedPercent }, + swap: { totalBytes: 0, availableBytes: null, freeBytes: 0, usedBytes: 0, usedPercent: 0 }, + system: { uptimeSeconds: 723_419, users: 1, processes: { total: 118, running: 2, sleeping: 115, blocked: 0, zombies: 1 } }, + disks: [{ device: "/dev/vda1", mount: "/", filesystem: "ext4", totalBytes: 21_474_836_480, freeBytes: 14_495_514_624, usedBytes: 6_979_321_856, usedPercent: 32.5 }], + network: [{ interface: "eth0", bytesReceived: history.at(-1).network[0].bytesReceived, bytesSent: history.at(-1).network[0].bytesSent, packetsReceived: 91_482, packetsSent: 64_501, errorsReceived: 0, errorsSent: 0, droppedReceived: 0, droppedSent: 0 }], + services: [ + { name: "nodedc-device-edge-channel.service", loadState: "loaded", activeState: "active", subState: "running", memoryBytes: 71_303_168, restarts: 0, pid: 1482 }, + { name: "nodedc-host-telemetry-agent.service", loadState: "loaded", activeState: "active", subState: "running", memoryBytes: 35_651_584, restarts: 0, pid: 1510 }, + { name: "ssh.service", loadState: "loaded", activeState: "active", subState: "running", memoryBytes: 8_388_608, restarts: 0, pid: 712 }, + ], + }; + return { + state: "online", + freshness: "fresh", + observedAt, + receivedAt: new Date(now).toISOString(), + expiresAt: new Date(now + 15_000).toISOString(), + current, + history, + observation: { + observationRef: "observation:preview-host-telemetry", + entityId: "observation.observation", + catalogHash: "229c61c02a790906", + targetRef: hostRef, + serviceInstanceRef, + edgeRef, + profileRef: "linux-host-telegraf-v1", + source: { ...current.source, provenanceRef: `${edgeRef}:${current.source.collectorRef}` }, + observedProperties: ["host.cpu.utilization", "host.memory.utilization", "host.disk.utilization", "host.network.counters", "host.systemd.unit-state"], + }, + }; +} + function requirePlatformOwner(actor) { if (actor?.hubRole !== "owner") { throw serviceError("device_platform_catalog_access_denied", 403); @@ -905,6 +994,10 @@ function seedArusnaviB2Preview({ routes, sessions, configurationStates, + hosts, + deployments, + serviceInstances, + healthObservations, }) { const timestamp = new Date().toISOString(); const ownerScopeRef = "owner-scope:78da71d5-f48f-4de0-8e47-729f6d644151"; @@ -913,6 +1006,9 @@ function seedArusnaviB2Preview({ const edgeRef = "edge:73da0c42-a641-4559-b8f7-23509b60bfe9"; const routeRef = "route:fef9b7a0-a462-4d68-9991-af026203368b"; const sessionRef = "session:57ead610-47de-45f7-a42d-fbe4fa0aba38"; + const hostRef = "host:adf2a5b6-3c0b-4a39-998c-07dfb7818ad1"; + const deploymentRef = "deployment:49b296f8-4cc8-470f-9dc1-2e3a543fd224"; + const serviceInstanceRef = "service-instance:01f14736-f5c2-4867-9cbc-2d268996a871"; const scope = { ownerScopeRef, scopeKind: "personal", @@ -963,6 +1059,52 @@ function seedArusnaviB2Preview({ createdAt: timestamp, updatedAt: timestamp, }); + hosts.set(hostRef, { + hostRef, + projectRef, + hostKey: "robot2b-b2-edge-vps", + displayName: "Robot2B B2 Edge VPS", + providerRef: "provider:beget", + externalRef: "host:koffyvngij", + managementCredentialConfigured: true, + lifecycleState: "active", + telemetry: previewHostTelemetry(hostRef, serviceInstanceRef, edgeRef), + ontology: previewOntology("infrastructure.host"), + }); + deployments.set(deploymentRef, { + deploymentRef, + projectRef, + hostRef, + deploymentKey: "device-edge-vps-command-transport", + displayName: "Device Edge VPS runtime", + artifactRef: "device-edge-vps-command-transport-20260812-013", + artifactDigest: "sha256:c7486ec879681ddd706f229b628c8556ca8c9ccc4f152a85debb409c302759ef", + lifecycleState: "active", + ontology: previewOntology("infrastructure.deployment"), + }); + serviceInstances.set(serviceInstanceRef, { + serviceInstanceRef, + projectRef, + hostRef, + deploymentRef, + edgeRef, + serviceKey: "device-edge", + displayName: "Robot2B B2 Device Edge", + serviceRole: "device.edge", + lifecycleState: "active", + ontology: previewOntology("infrastructure.service_instance"), + }); + const healthObservationRef = "health-observation:b912a45a-2b97-4562-8f9e-824cc24e0710"; + healthObservations.set(healthObservationRef, { + healthObservationRef, + projectRef, + subjectKind: "host", + subjectRef: hostRef, + observedState: "reachable", + evidenceClass: "agent.telemetry", + observedAt: timestamp, + expiresAt: new Date(Date.now() + 60_000).toISOString(), + }); routes.set(routeRef, { routeRef, projectRef, diff --git a/apps/device-manager/src/DeviceControlViews.tsx b/apps/device-manager/src/DeviceControlViews.tsx index b627eec..ce4e7bd 100644 --- a/apps/device-manager/src/DeviceControlViews.tsx +++ b/apps/device-manager/src/DeviceControlViews.tsx @@ -3,6 +3,7 @@ import { Button, GlassSurface, Icon, + IconButton, Select, SettingsCard, StatusBadge, @@ -41,6 +42,7 @@ import type { DeviceManagerSession, EdgeView, InfrastructureHostView, + InfrastructureServiceInstanceView, ModelProfileView, ProjectWorkspace, } from "./types"; @@ -79,12 +81,14 @@ export function DeviceControlView({ workspace, session, onRefresh, + onPoll, onError, }: { view: ControlViewId; workspace: ProjectWorkspace; session: DeviceManagerSession; onRefresh: () => Promise; + onPoll: () => Promise; onError: (reason: unknown) => void; }) { const [dialog, setDialog] = useState(null); @@ -181,6 +185,8 @@ export function DeviceControlView({ assetBindingRef: binding.assetBindingRef, validTo: new Date().toISOString(), }))} + onPoll={onPoll} + onError={onError} /> ) : null} {view === "sessions" ? : null} @@ -472,6 +478,8 @@ function HostsView({ onCreateAsset, onCreateAssetBinding, onCloseAssetBinding, + onPoll, + onError, }: { workspace: ProjectWorkspace; canManageInfrastructure: boolean; @@ -485,8 +493,25 @@ function HostsView({ onCreateAsset: () => void; onCreateAssetBinding: () => void; onCloseAssetBinding: (binding: AssetBindingView) => void; + onPoll: () => Promise; + onError: (reason: unknown) => void; }) { + const [selectedHostRef, setSelectedHostRef] = useState(null); const topology = workspace.ontology; + const selectedHost = selectedHostRef + ? topology.hosts.find((host) => host.hostRef === selectedHostRef) ?? null + : null; + if (selectedHost) { + return ( + service.hostRef === selectedHost.hostRef)} + onBack={() => setSelectedHostRef(null)} + onPoll={onPoll} + onError={onError} + /> + ); + } return ( Health evidence - ) : null} + action={<> + + {canManageInfrastructure ? ( + + ) : null} + } /> ); })} @@ -624,6 +652,245 @@ function HostsView({ ); } +function HostTelemetryWorkspace({ + host, + services, + onBack, + onPoll, + onError, +}: { + host: InfrastructureHostView; + services: InfrastructureServiceInstanceView[]; + onBack: () => void; + onPoll: () => Promise; + onError: (reason: unknown) => void; +}) { + useEffect(() => { + let active = true; + const timer = window.setInterval(() => { + onPoll().catch((reason) => active && onError(reason)); + }, 3_000); + return () => { + active = false; + window.clearInterval(timer); + }; + }, [onError, onPoll]); + + const telemetry = host.telemetry; + const current = telemetry.current; + const networkRate = calculateNetworkRate(telemetry.history); + const cpuHistory = telemetry.history.map((sample) => sample.cpuUsagePercent); + const memoryHistory = telemetry.history.map((sample) => sample.memoryUsedPercent); + const receiveHistory = calculateNetworkRateHistory(telemetry.history, "received"); + const sendHistory = calculateNetworkRateHistory(telemetry.history, "sent"); + const runtimeServices = current?.services ?? []; + + return ( +
+
+ + + +
+ СИСТЕМА / ВЫЧИСЛИТЕЛЬНЫЕ МОДУЛИ +

{host.displayName}

+

Аппаратный и процессинговый срез VPS. Метрики снимает host-agent; Device Core хранит только канонические наблюдения.

+
+
+ + {telemetry.freshness === "fresh" ? "Свежие данные" : telemetry.freshness === "stale" ? "Данные устарели" : "Нет данных"} + + onPoll().catch(onError)}> + + +
+
+ +
+ + + + +
+ +
+
+
HARDWARE

{current?.hardware.hostname ?? host.hostKey}

+ {telemetry.state} +
+
+ + + + + + + + +
+
+ +
+
+
STORAGE

Файловые системы

{current?.disks.length ?? 0}
+
+ {(current?.disks ?? []).map((disk, index) => ( +
+ {disk.mount ?? disk.device ?? "Диск"}{[disk.device, disk.filesystem].filter(Boolean).join(" · ")} + {formatPercent(disk.usedPercent)}{formatUsedTotal(disk.usedBytes, disk.totalBytes)} +
+ ))} + {!current?.disks.length ?
Данные о дисках ещё не поступили.
: null} +
+
+ +
+
NETWORK

Сетевые интерфейсы

{current?.network.length ?? 0}
+
+ {(current?.network ?? []).filter((item) => item.interface !== "lo").map((item, index) => ( +
+ {item.interface ?? "Интерфейс"}{formatPackets(item.packetsReceived, item.packetsSent)} + ↓ {formatMetricBytes(item.bytesReceived)}↑ {formatMetricBytes(item.bytesSent)} +
+ ))} + {!current?.network.filter((item) => item.interface !== "lo").length ?
Сетевые счётчики ещё не поступили.
: null} +
+
+
+ +
+
+
PROCESSING RUNTIME

Сервисы VPS

Состояние systemd-юнитов и их ресурсный профиль.

+ service.activeState === "failed") ? "danger" : "success"}>{runtimeServices.length} units +
+
+ {runtimeServices.map((service, index) => ( +
+ {service.name ?? "systemd unit"}{service.subState ?? service.loadState ?? "—"} + {service.activeState ?? "unknown"}{formatMetricBytes(service.memoryBytes)} +
+ ))} + {!runtimeServices.length ?
Состояние сервисов ещё не поступило.
: null} +
+
+ +
+
ONTOLOGY / OBSERVATION{telemetry.observation?.entityId ?? "observation.observation"}
+
Источник{current ? `${current.source.agent} ${current.source.agentVersion}` : "—"}
+
Последнее наблюдение{formatDate(telemetry.observedAt)}
+
Связанные сервисы{services.length}
+
+
+ ); +} + +function TelemetryMetricCard({ label, value, points, detail }: { label: string; value: string; points: Array; detail: string }) { + return ( +
+ {label} + {value} + + {detail} +
+ ); +} + +function Sparkline({ values }: { values: Array }) { + const normalized = values.filter((value): value is number => typeof value === "number" && Number.isFinite(value)); + if (normalized.length < 2) return
; + const minimum = Math.min(...normalized); + const maximum = Math.max(...normalized); + const spread = Math.max(1, maximum - minimum); + const points = normalized.map((value, index) => { + const x = (index / (normalized.length - 1)) * 100; + const y = 28 - ((value - minimum) / spread) * 24; + return `${x.toFixed(2)},${y.toFixed(2)}`; + }).join(" "); + return ; +} + +function TelemetryFact({ label, value }: { label: string; value: string | null | undefined }) { + return
{label}{value || "—"}
; +} + +function calculateNetworkRate(history: InfrastructureHostView["telemetry"]["history"]) { + if (history.length < 2) return { received: null, sent: null }; + const previous = history[history.length - 2]; + const latest = history[history.length - 1]; + const seconds = (new Date(latest.observedAt).valueOf() - new Date(previous.observedAt).valueOf()) / 1000; + if (!Number.isFinite(seconds) || seconds <= 0) return { received: null, sent: null }; + const previousTotals = networkTotals(previous.network); + const latestTotals = networkTotals(latest.network); + return { + received: nonNegativeRate(latestTotals.received - previousTotals.received, seconds), + sent: nonNegativeRate(latestTotals.sent - previousTotals.sent, seconds), + }; +} + +function calculateNetworkRateHistory(history: InfrastructureHostView["telemetry"]["history"], direction: "received" | "sent") { + return history.slice(1).map((sample, index) => { + const previous = history[index]; + const seconds = (new Date(sample.observedAt).valueOf() - new Date(previous.observedAt).valueOf()) / 1000; + if (seconds <= 0) return null; + const currentTotals = networkTotals(sample.network); + const previousTotals = networkTotals(previous.network); + return nonNegativeRate(currentTotals[direction] - previousTotals[direction], seconds); + }); +} + +function networkTotals(network: InfrastructureHostView["telemetry"]["history"][number]["network"]) { + return network.filter((item) => item.interface !== "lo").reduce((total, item) => ({ + received: total.received + (item.bytesReceived ?? 0), + sent: total.sent + (item.bytesSent ?? 0), + }), { received: 0, sent: 0 }); +} + +function nonNegativeRate(bytes: number, seconds: number) { + const value = bytes / seconds; + return Number.isFinite(value) && value >= 0 ? value : null; +} + +function formatPercent(value: number | null | undefined) { + return value == null ? "—" : `${value.toFixed(value >= 10 ? 0 : 1)}%`; +} + +function formatRate(value: number | null) { + return value == null ? "—" : `${formatMetricBytes(value)}/s`; +} + +function formatMetricBytes(value: number | null | undefined) { + if (value == null || !Number.isFinite(value)) return "—"; + if (value < 1024) return `${Math.round(value)} B`; + if (value < 1024 ** 2) return `${(value / 1024).toFixed(1)} KiB`; + if (value < 1024 ** 3) return `${(value / 1024 ** 2).toFixed(1)} MiB`; + return `${(value / 1024 ** 3).toFixed(1)} GiB`; +} + +function formatUsedTotal(used: number | null | undefined, total: number | null | undefined) { + return used == null || total == null ? "—" : `${formatMetricBytes(used)} / ${formatMetricBytes(total)}`; +} + +function formatNullable(value: number | null | undefined) { + return value == null ? "—" : new Intl.NumberFormat("ru-RU").format(value); +} + +function formatDuration(seconds: number | null | undefined) { + if (seconds == null) return "—"; + const days = Math.floor(seconds / 86_400); + const hours = Math.floor((seconds % 86_400) / 3_600); + const minutes = Math.floor((seconds % 3_600) / 60); + return [days ? `${days} д` : null, hours ? `${hours} ч` : null, `${minutes} мин`].filter(Boolean).join(" "); +} + +function formatLoad(cpu: { load1: number | null; load5: number | null; load15: number | null } | null | undefined) { + if (!cpu || cpu.load1 == null) return "load average —"; + return `load ${[cpu.load1, cpu.load5, cpu.load15].map((value) => value?.toFixed(2) ?? "—").join(" / ")}`; +} + +function formatPackets(received: number | null | undefined, sent: number | null | undefined) { + return `↓ ${formatNullable(received)} пакетов · ↑ ${formatNullable(sent)} пакетов`; +} + function SessionsView({ workspace }: { workspace: ProjectWorkspace }) { return ( diff --git a/apps/device-manager/src/DeviceManagerApp.tsx b/apps/device-manager/src/DeviceManagerApp.tsx index 18f2725..2909ab0 100644 --- a/apps/device-manager/src/DeviceManagerApp.tsx +++ b/apps/device-manager/src/DeviceManagerApp.tsx @@ -806,6 +806,7 @@ function ProjectView({ view, workspace, canManageCollections, canClaim, canConfi workspace={workspace} session={session} onRefresh={onRefresh} + onPoll={onPoll} onError={onError} />; } diff --git a/apps/device-manager/src/styles.css b/apps/device-manager/src/styles.css index e21ec2e..6095119 100644 --- a/apps/device-manager/src/styles.css +++ b/apps/device-manager/src/styles.css @@ -127,6 +127,231 @@ padding-right: 3px; } +.host-telemetry-workspace { + display: grid; + gap: 18px; + min-width: 0; +} + +.host-telemetry-header { + display: grid; + grid-template-columns: auto minmax(0, 1fr) auto; + align-items: start; + gap: 14px; +} + +.host-telemetry-header__copy { + display: grid; + gap: 3px; +} + +.host-telemetry-header__copy small, +.host-telemetry-section__heading small, +.host-telemetry-evidence small { + color: var(--nodedc-text-tertiary); + font-size: 0.62rem; + font-weight: 780; + letter-spacing: 0.12em; + text-transform: uppercase; +} + +.host-telemetry-header__copy h2, +.host-telemetry-header__copy p, +.host-telemetry-section__heading h3, +.host-telemetry-section__heading p { + margin: 0; +} + +.host-telemetry-header__copy h2 { + font-size: 1.4rem; + letter-spacing: -0.035em; +} + +.host-telemetry-header__copy p, +.host-telemetry-section__heading p { + color: var(--nodedc-text-tertiary); + font-size: 0.76rem; + line-height: 1.5; +} + +.host-telemetry-header__actions { + display: flex; + align-items: center; + gap: 8px; +} + +.host-telemetry-metric-grid { + display: grid; + grid-template-columns: repeat(4, minmax(0, 1fr)); + gap: 10px; +} + +.host-telemetry-metric, +.host-telemetry-section { + border: 1px solid var(--nodedc-glass-outline); + border-radius: var(--nodedc-radius-lg); + background: color-mix(in srgb, var(--nodedc-surface) 82%, transparent); +} + +.host-telemetry-metric { + display: grid; + grid-template-columns: minmax(0, 1fr) auto; + gap: 5px 12px; + min-height: 112px; + padding: 14px; + overflow: hidden; +} + +.host-telemetry-metric > span, +.host-telemetry-metric > small, +.host-telemetry-facts small, +.host-telemetry-list__row small, +.host-telemetry-service small { + color: var(--nodedc-text-tertiary); + font-size: 0.66rem; +} + +.host-telemetry-metric > strong { + font-size: 1.35rem; + font-weight: 620; +} + +.host-telemetry-metric > small { + grid-column: 1 / -1; +} + +.host-telemetry-sparkline { + grid-column: 1 / -1; + width: 100%; + height: 32px; + overflow: visible; +} + +.host-telemetry-sparkline polyline { + fill: none; + stroke: rgb(var(--nodedc-accent-rgb)); + stroke-width: 1.5; + vector-effect: non-scaling-stroke; +} + +.host-telemetry-sparkline--empty { + border-bottom: 1px solid var(--nodedc-glass-outline); +} + +.host-telemetry-section { + display: grid; + gap: 14px; + padding: 18px; +} + +.host-telemetry-section__heading { + display: flex; + align-items: flex-start; + justify-content: space-between; + gap: 16px; +} + +.host-telemetry-section__heading > div { + display: grid; + gap: 4px; +} + +.host-telemetry-section__heading h3 { + font-size: 1rem; +} + +.host-telemetry-facts { + display: grid; + grid-template-columns: repeat(4, minmax(0, 1fr)); + gap: 1px; + overflow: hidden; + border-radius: var(--nodedc-radius-md); + background: var(--nodedc-glass-outline); +} + +.host-telemetry-facts > div { + display: grid; + gap: 5px; + min-width: 0; + padding: 12px; + background: var(--nodedc-surface-soft); +} + +.host-telemetry-facts strong { + overflow: hidden; + font-size: 0.78rem; + font-weight: 590; + text-overflow: ellipsis; + white-space: nowrap; +} + +.host-telemetry-split { + display: grid; + grid-template-columns: repeat(2, minmax(0, 1fr)); + gap: 12px; +} + +.host-telemetry-list, +.host-telemetry-service-grid { + display: grid; + gap: 7px; +} + +.host-telemetry-list__row, +.host-telemetry-service { + display: flex; + align-items: center; + justify-content: space-between; + gap: 16px; + min-width: 0; + padding: 10px 12px; + border-radius: var(--nodedc-radius-md); + background: var(--nodedc-surface-soft); +} + +.host-telemetry-list__row > span, +.host-telemetry-service > span { + display: grid; + gap: 3px; + min-width: 0; +} + +.host-telemetry-list__row > span:last-child, +.host-telemetry-service > span:last-child { + justify-items: end; + text-align: right; +} + +.host-telemetry-list__row strong, +.host-telemetry-service strong { + overflow: hidden; + font-size: 0.76rem; + font-weight: 590; + text-overflow: ellipsis; + white-space: nowrap; +} + +.host-telemetry-service-grid { + grid-template-columns: repeat(3, minmax(0, 1fr)); +} + +.host-telemetry-evidence { + grid-template-columns: repeat(4, minmax(0, 1fr)); +} + +.host-telemetry-evidence > div { + display: grid; + gap: 5px; + min-width: 0; +} + +.host-telemetry-evidence strong { + overflow: hidden; + font-size: 0.75rem; + text-overflow: ellipsis; + white-space: nowrap; +} + @media (max-width: 760px) { .device-control-resource-grid, .device-control-policy-grid { @@ -168,6 +393,23 @@ grid-column: 2; justify-content: flex-start; } + + .host-telemetry-header { + grid-template-columns: auto minmax(0, 1fr); + } + + .host-telemetry-header__actions { + grid-column: 2; + justify-content: flex-start; + } + + .host-telemetry-metric-grid, + .host-telemetry-facts, + .host-telemetry-service-grid, + .host-telemetry-evidence, + .host-telemetry-split { + grid-template-columns: 1fr; + } } html, diff --git a/apps/device-manager/src/types.ts b/apps/device-manager/src/types.ts index c20e5b5..d9ce22d 100644 --- a/apps/device-manager/src/types.ts +++ b/apps/device-manager/src/types.ts @@ -358,9 +358,63 @@ export interface InfrastructureHostView { managementCredentialConfigured: boolean; lifecycleState: string; health: HealthProjectionView; + telemetry: InfrastructureHostTelemetryView; ontology: OntologyRefView; } +export interface InfrastructureHostTelemetrySnapshot { + schemaVersion: "nodedc.infrastructure.host-telemetry.v1"; + profile: "linux-host-telegraf-v1"; + hostKey: string; + observedAt: string; + source: { agent: string; agentVersion: string; collectorRef: string }; + hardware: { + hostname: string | null; + architecture: string | null; + platform: string | null; + kernelRelease: string | null; + cpuModel: string | null; + logicalProcessors: number | null; + }; + cpu: { usagePercent: number | null; load1: number | null; load5: number | null; load15: number | null }; + memory: { totalBytes: number | null; availableBytes: number | null; freeBytes: number | null; usedBytes: number | null; usedPercent: number | null }; + swap: { totalBytes: number | null; availableBytes: number | null; freeBytes: number | null; usedBytes: number | null; usedPercent: number | null }; + system: { + uptimeSeconds: number | null; + users: number | null; + processes: { total: number | null; running: number | null; sleeping: number | null; blocked: number | null; zombies: number | null }; + }; + disks: Array<{ device: string | null; mount: string | null; filesystem: string | null; totalBytes: number | null; freeBytes: number | null; usedBytes: number | null; usedPercent: number | null }>; + network: Array<{ interface: string | null; bytesReceived: number | null; bytesSent: number | null; packetsReceived: number | null; packetsSent: number | null; errorsReceived: number | null; errorsSent: number | null; droppedReceived: number | null; droppedSent: number | null }>; + services: Array<{ name: string | null; loadState: string | null; activeState: string | null; subState: string | null; memoryBytes: number | null; restarts: number | null; pid: number | null }>; +} + +export interface InfrastructureHostTelemetryView { + state: string; + freshness: "fresh" | "stale" | "missing"; + observedAt: string | null; + receivedAt: string | null; + expiresAt: string | null; + current: InfrastructureHostTelemetrySnapshot | null; + history: Array<{ + observedAt: string; + cpuUsagePercent: number | null; + memoryUsedPercent: number | null; + network: Array<{ interface: string | null; bytesReceived: number | null; bytesSent: number | null }>; + }>; + observation: null | { + observationRef: string; + entityId: string; + catalogHash: string; + targetRef: string; + serviceInstanceRef: string; + edgeRef: string; + profileRef: string; + source: { agent: string; agentVersion: string; collectorRef: string; provenanceRef: string }; + observedProperties: string[]; + }; +} + export interface InfrastructureEndpointView { endpointRef: string; hostRef: string; diff --git a/deployment/device-control-core-release-v3.json b/deployment/device-control-core-release-v3.json new file mode 100644 index 0000000..427af9e --- /dev/null +++ b/deployment/device-control-core-release-v3.json @@ -0,0 +1,40 @@ +{ + "schemaVersion": "nodedc.device-plane.device-control-core-release.v3", + "releaseId": "__PATCH_ID__", + "action": "upgrade", + "predecessor": { + "kind": "release", + "patchId": "device-control-core-release-v2-20260822-038", + "artifactSha256": "e2d062b82b022dba662522b5d6e192026ac964d78950d903295ca3cbbc95ab28" + }, + "service": "device-control-core", + "composeActivation": "preserve-active-v4-topology", + "identity": "reuse-existing-runner-managed-host-local-private-key-public-certificate-export", + "identityRecovery": "forbidden-valid-existing-identity-required", + "tlsPurpose": "clientAuth", + "direction": "core-initiated", + "endpointPolicy": "public-ipv4-standard-https-tcp-443-only", + "coreNetworks": [ + "device-plane-private", + "device-plane-egress" + ], + "publicIngress": "none-on-synology", + "edgeRegistrations": "preserved", + "commandTransport": "typed-service-ping-v1", + "commandCatalog": "allowlisted-adapter-typed-commands-only", + "credentialBoundary": "transient-core-memory-then-single-pinned-mtls-command-envelope-to-edge-never-persisted-never-logged-never-returned", + "telemetryTransport": "edge-channel-host-telemetry-observed-v1", + "telemetryContract": "nodedc.infrastructure.host-telemetry.v1", + "telemetryStorage": "device-control-core-postgres-seven-day-retention", + "ontologyProjection": "observation-observed-property-provenance-freshness-v1", + "telemetryFreshness": "fifteen-seconds-missing-stale-not-unhealthy", + "gelios": "untouched-legacy-only", + "preservedServices": [ + "device-manager", + "device-gateway", + "device-postgres", + "device-backhaul-target" + ], + "healthGate": "bounded-container-grace+core-edge-contract+exact-private-egress-network-boundary", + "rollback": "restore-preapply-source-and-core-runtime" +} diff --git a/deployment/device-edge-vps-host-telemetry-v1.json b/deployment/device-edge-vps-host-telemetry-v1.json new file mode 100644 index 0000000..024b5b1 --- /dev/null +++ b/deployment/device-edge-vps-host-telemetry-v1.json @@ -0,0 +1,77 @@ +{ + "schemaVersion": "nodedc.device-edge-vps.host-telemetry.v1", + "mode": "provider-neutral-host-observation-over-accepted-core-channel", + "status": "active-host-telemetry", + "authority": "DCPLATFORM-21/DCPLATFORM-76/DCPLATFORM-77/ADR-0001", + "component": "device-edge-vps", + "phase": "host-telemetry", + "runtimeHost": "koffyvngij", + "predecessorPatch": "device-edge-vps-command-transport-20260812-013", + "predecessorArtifactSha256": "c7486ec879681ddd706f229628c8556ca8c9ccc4f152a85debb409c302759ef", + "agent": "telegraf", + "agentVersion": "1.38.4", + "agentRuntimeUser": "nodedc-telemetry", + "agentService": "nodedc-host-telemetry-agent.service", + "collector": "127.0.0.1:18223/internal/v1/host-telemetry", + "collectorExposure": "loopback-only", + "transport": "existing-core-initiated-pinned-mtls-channel", + "messageKind": "host.telemetry.observed", + "observationEntity": "observation.observation", + "observationTarget": "infrastructure.host", + "observedProperties": [ + "host.cpu.utilization", + "host.memory.utilization", + "host.swap.utilization", + "host.disk.utilization", + "host.network.counters", + "host.process.counts", + "host.systemd.unit-state" + ], + "commandTransport": "typed-service-ping-v1", + "publicIngress": "preserved:tcp/443-mtls-core-channel+tcp/9921-bidirectional-tracker-session", + "mqtt": "disabled-no-public-broker-no-wan-plaintext", + "database": "none-on-vps", + "credentials": "none-added", + "gelios": "untouched-legacy-only", + "resourceCeilings": { + "agentMemory": "96M", + "agentSwap": "0", + "agentCpu": "15%", + "agentTasks": 64, + "agentOpenFiles": 512, + "collectorBodyBytes": 524288, + "channelEnvelopeBytes": 1048576 + }, + "preserved": [ + "management-ssh-key", + "accepted-node-runtime", + "accepted-core-channel-trust-and-registration", + "accepted-tracker-ingress", + "accepted-typed-command-transport", + "retired-tailnet-boundary", + "gelios-production-path" + ], + "forbidden": [ + "public-mqtt", + "plaintext-wan-telemetry", + "vps-initiated-synology-connection", + "public-health", + "vps-database", + "browser-secrets", + "generic-shell", + "ontology-core-telemetry-storage" + ], + "acceptance": [ + "exact-command-transport-013-predecessor", + "telegraf-1.38.4-exact-archive-and-binary", + "dedicated-non-root-agent", + "loopback-only-http-output", + "core-channel-remains-accepted", + "host-observation-reaches-device-control-core", + "public-port-set-unchanged", + "typed-command-transport-preserved", + "tailscale-remains-absent", + "gelios-untouched" + ], + "rollback": "restore-exact-command-transport-013-source-runtime-and-remove-agent-runtime" +} diff --git a/deployment/device-manager-release-v9.json b/deployment/device-manager-release-v9.json new file mode 100644 index 0000000..24bf061 --- /dev/null +++ b/deployment/device-manager-release-v9.json @@ -0,0 +1,56 @@ +{ + "schemaVersion": "nodedc.device-plane.device-manager-release.v9", + "releaseId": "__PATCH_ID__", + "action": "upgrade", + "predecessor": { + "kind": "release", + "patchId": "device-manager-release-v8-20260822-039", + "artifactSha256": "30a83d4c6b5c029c96c4af19fa558e0ae75e4b60bc897881457304bd17e0e465" + }, + "controlCorePredecessor": { + "patchId": "device-control-core-release-v3-20260822-040", + "artifactSha256": "08448a56cdf391076f92c5242e368fd0033b420167874645eebfa1f22184ee92" + }, + "edgeChannelPredecessor": { + "patchId": "device-edge-core-channel-upgrade-v4-20260812-023", + "artifactSha256": "c10d5b6b7d55ab239f85b6c8130e34ce9f84985e3b46e6e5534733156c7982fc" + }, + "service": "device-manager", + "publicIngress": "reverse-proxy-only", + "deviceCoreManagementApi": "file-token-authenticated", + "launcherTrust": "file-token-scoped-to-device-core-handoff", + "edgeChannel": "preserve-active-v4-core-initiated-pinned-mtls", + "edgeChannelIdentity": "reuse-runner-managed-host-local-private-key-public-certificate-export", + "edgeChannelEgress": "preserve-dedicated-core-only-bridge-no-host-ingress-public-ipv4-tcp-443-only", + "healthGate": "bounded-container-grace+core-contract+persistent-data", + "commandTransport": "typed-service-ping-v1", + "commandCatalog": "allowlisted-adapter-typed-commands-only", + "credentialBoundary": "transient-core-memory-then-single-pinned-mtls-command-envelope-to-edge-never-persisted-never-logged-never-returned", + "presentationPersistence": "runner-managed-host-data-bind", + "presentationDataHostPath": "/volume1/docker/nodedc-device-plane/data/device-manager", + "presentationDataContainerPath": "/var/lib/nodedc-device-manager", + "presentationDataOwnership": "uid-1000-gid-1000-mode-0750", + "presentationDataLifecycle": "preserve-across-manager-recreate-and-source-rollback", + "presentationPath": "/var/lib/nodedc-device-manager/device-manager-presentation.json", + "mediaRoot": "/var/lib/nodedc-device-manager/media", + "defaultAccentHex": "#f5f5f5", + "overviewLayout": "mission-core-landing-stage-v1", + "faviconSet": "nodedc-adaptive-v1", + "commandFormLayout": "aligned-control-row-v1", + "secondaryEmptyTypography": "help-text-sm-v1", + "infrastructureHostProjection": "ontology-backed-host-runtime-v1", + "ontologyFoundation": "ontology-core-device-foundation-20260822-001", + "ontologyCatalogHash": "229c61c02a790906", + "assetBinding": "temporal-device-asset-binding-v1", + "infrastructureRuntime": "host-endpoint-deployment-service-instance-v1", + "healthEvidence": "ttl-observation-missing-not-unhealthy-v1", + "telemetryWorkspace": "mission-core-compute-module-parity-v1", + "telemetryNavigation": "full-workspace-back-navigation-v1", + "telemetryPollInterval": "three-seconds", + "telemetryFreshness": "fifteen-seconds-missing-stale-not-unhealthy", + "telemetryOntologyProjection": "observation-observed-property-provenance-freshness-v1", + "telemetryAgent": "telegraf-host-observer-v1", + "interactiveShell": "disabled-pending-managed-session-boundary", + "gelios": "untouched-legacy-only", + "rollback": "restore-preapply-snapshot-preserve-manager-data" +} diff --git a/infra/deploy-runner/build-device-control-core-release-artifact.mjs b/infra/deploy-runner/build-device-control-core-release-artifact.mjs index b24ac7d..cd23fa1 100644 --- a/infra/deploy-runner/build-device-control-core-release-artifact.mjs +++ b/infra/deploy-runner/build-device-control-core-release-artifact.mjs @@ -35,20 +35,24 @@ if ( ); } +const isV3 = patchId.startsWith("device-control-core-release-v3-"); const isV2 = patchId.startsWith("device-control-core-release-v2-"); const expectedV2Predecessor = Object.freeze({ patchId: predecessorPatchId ?? "device-control-core-release-20260812-024", artifactSha256: predecessorSha256 ?? "a289e909283109642e6bba3d9822a31f63423cfe0bbcd52705979681bd2bc793", }); -const descriptorPath = isV2 - ? "deployment/device-control-core-release-v2.json" - : "deployment/device-control-core-release-v1.json"; +const descriptorPath = isV3 + ? "deployment/device-control-core-release-v3.json" + : isV2 + ? "deployment/device-control-core-release-v2.json" + : "deployment/device-control-core-release-v1.json"; const entries = [ ".dockerignore", "package.json", "package-lock.json", "packages/device-protocol-contract", "packages/device-edge-channel-contract", + ...(isV3 ? ["packages/infrastructure-telemetry-contract"] : []), "services/device-control-core", descriptorPath, ]; @@ -88,6 +92,7 @@ try { "services/device-control-core/src/device-gateway-core-runtime.mjs", "packages/device-protocol-contract/src/index.mjs", "packages/device-edge-channel-contract/src/index.mjs", + ...(isV3 ? ["packages/infrastructure-telemetry-contract/src/index.mjs"] : []), ]) { const imported = spawnSync( process.execPath, @@ -101,7 +106,7 @@ try { const descriptor = JSON.parse(await readFile(join(payload, descriptorPath), "utf8")); if ( - descriptor.schemaVersion !== `nodedc.device-plane.device-control-core-release.${isV2 ? "v2" : "v1"}` + descriptor.schemaVersion !== `nodedc.device-plane.device-control-core-release.${isV3 ? "v3" : isV2 ? "v2" : "v1"}` || descriptor.releaseId !== patchId || descriptor.action !== "upgrade" || descriptor.service !== "device-control-core" @@ -113,11 +118,11 @@ try { || JSON.stringify(descriptor.coreNetworks) !== JSON.stringify(["device-plane-private", "device-plane-egress"]) || descriptor.publicIngress !== "none-on-synology" || descriptor.edgeRegistrations !== "preserved" - || descriptor.commandTransport !== (isV2 ? "typed-service-ping-v1" : "disabled") - || descriptor.gelios !== (isV2 ? "untouched-legacy-only" : "untouched") + || descriptor.commandTransport !== ((isV2 || isV3) ? "typed-service-ping-v1" : "disabled") + || descriptor.gelios !== ((isV2 || isV3) ? "untouched-legacy-only" : "untouched") || descriptor.rollback !== "restore-preapply-source-and-core-runtime" || ( - isV2 + (isV2 || isV3) && ( descriptor.commandCatalog !== "allowlisted-adapter-typed-commands-only" || descriptor.credentialBoundary !== "transient-core-memory-then-single-pinned-mtls-command-envelope-to-edge-never-persisted-never-logged-never-returned" @@ -125,6 +130,16 @@ try { || descriptor.predecessor?.artifactSha256 !== expectedV2Predecessor.artifactSha256 ) ) + || ( + isV3 + && ( + descriptor.telemetryTransport !== "edge-channel-host-telemetry-observed-v1" + || descriptor.telemetryContract !== "nodedc.infrastructure.host-telemetry.v1" + || descriptor.telemetryStorage !== "device-control-core-postgres-seven-day-retention" + || descriptor.ontologyProjection !== "observation-observed-property-provenance-freshness-v1" + || descriptor.telemetryFreshness !== "fifteen-seconds-missing-stale-not-unhealthy" + ) + ) ) { throw new Error("device_control_core_release_contract_mismatch"); } diff --git a/infra/deploy-runner/build-device-edge-vps-artifact.mjs b/infra/deploy-runner/build-device-edge-vps-artifact.mjs index bec472e..da8a1bf 100644 --- a/infra/deploy-runner/build-device-edge-vps-artifact.mjs +++ b/infra/deploy-runner/build-device-edge-vps-artifact.mjs @@ -38,11 +38,12 @@ if ( "tailscale-retirement", "tracker-ingress", "command-transport", + "host-telemetry", ].includes(phase) || !/^[A-Za-z0-9._-]{1,96}$/.test(patchId || "") ) { throw new Error( - "usage: build-device-edge-vps-artifact.mjs ", + "usage: build-device-edge-vps-artifact.mjs ", ); } @@ -53,16 +54,22 @@ if ( ) { throw new Error("vps_initiated_transport_frozen:ADR-0001"); } -const acceptedSharedSourcePhases = new Set(["core-channel", "tracker-ingress"]); +const acceptedSharedSourcePhases = new Set([ + "core-channel", + "tracker-ingress", + "command-transport", +]); if (acceptedSharedSourcePhases.has(phase)) { throw new Error(`accepted_vps_phase_rebuild_frozen:${phase}:ADR-0001`); } const nodeArchive = "node-v22.23.2-linux-x64.tar.xz"; const tailscaleArchive = "tailscale_1.102.2_amd64.tgz"; +const telegrafArchive = "telegraf-1.38.4_linux_amd64.tar.gz"; const runtimeDigests = new Map([ [nodeArchive, "d60acfe00a2932254bb0ad20e01b0d74397a0875595de719654b214f4b03f307"], [tailscaleArchive, "ad2cde12f8de95f7b93a1e0401e652291c603d42b9d60a33fb1741eb38ab04d8"], + [telegrafArchive, "81857e9745ebf26e058b6fdc27b9b2c210fd1fe61e57d7fad3d4bb9131f60041"], ]); const entriesByPhase = { @@ -130,6 +137,21 @@ const entriesByPhase = { "vps/edge-process/device-edge-runtime.mjs", "deployment/device-edge-vps-command-transport-v1.json", ], + "host-telemetry": [ + "packages/device-edge-channel-contract/package.json", + "packages/device-edge-channel-contract/src", + "packages/infrastructure-telemetry-contract/package.json", + "packages/infrastructure-telemetry-contract/src", + "services/device-edge-channel/package.json", + "services/device-edge-channel/src", + "vps/edge-process/device-edge-runtime.mjs", + "vps/edge-process/host-telemetry-runtime.mjs", + "vps/config/nodedc-host-telemetry-telegraf.conf", + "vps/systemd/nodedc-device-edge-runtime.service", + "vps/systemd/nodedc-host-telemetry-agent.service", + "deployment/device-edge-vps-host-telemetry-v1.json", + `vendor/${telegrafArchive}`, + ], }; const entries = entriesByPhase[phase]; const ignoredBasenames = new Set([".DS_Store", ".git", "node_modules"]); @@ -191,10 +213,10 @@ try { ? "tcp/9921" : ["core-channel", "tailscale-retirement"].includes(phase) ? "tcp/443-mtls-only" - : ["tracker-ingress", "command-transport"].includes(phase) + : ["tracker-ingress", "command-transport", "host-telemetry"].includes(phase) ? "tcp/443-mtls+tcp/9921-telemetry" : "disabled", - commandTransport: phase === "command-transport" + commandTransport: ["command-transport", "host-telemetry"].includes(phase) ? "typed-service-ping-v1" : "disabled", gelios: "untouched", @@ -212,7 +234,7 @@ async function assertBoundary() { if ( descriptor.component !== "device-edge-vps" || descriptor.runtimeHost !== "koffyvngij" - || descriptor.commandTransport !== (phase === "command-transport" + || descriptor.commandTransport !== (["command-transport", "host-telemetry"].includes(phase) ? "typed-service-ping-v1" : "disabled") || !String(descriptor.gelios || "").startsWith("untouched") @@ -419,6 +441,40 @@ async function assertBoundary() { } } } + if (phase === "host-telemetry") { + for (const required of [ + '"predecessorPatch": "device-edge-vps-command-transport-20260812-013"', + '"agent": "telegraf"', + '"agentVersion": "1.38.4"', + '"transport": "existing-core-initiated-pinned-mtls-channel"', + '"mqtt": "disabled-no-public-broker-no-wan-plaintext"', + "User=nodedc-telemetry", + "IPAddressDeny=any", + "IPAddressAllow=localhost", + "MemoryMax=96M", + "CPUQuota=15%", + 'url = "http://127.0.0.1:18223/internal/v1/host-telemetry"', + 'data_format = "json"', + "submitHostTelemetry", + "createHostTelemetryCollector", + ]) { + if (!combined.includes(required)) { + throw new Error(`host_telemetry_boundary_missing:${required}`); + } + } + for (const forbidden of [ + "mqtt://", + "tcp://", + "outputs.mqtt", + "PRIVATE KEY", + "TS_AUTHKEY", + "device.dc.ru", + ]) { + if (combined.includes(forbidden)) { + throw new Error(`host_telemetry_boundary_violation:${forbidden}`); + } + } + } } function canonicalTarScript() { diff --git a/infra/deploy-runner/build-device-manager-control-plane-artifact.mjs b/infra/deploy-runner/build-device-manager-control-plane-artifact.mjs index 89b75db..688d5c0 100644 --- a/infra/deploy-runner/build-device-manager-control-plane-artifact.mjs +++ b/infra/deploy-runner/build-device-manager-control-plane-artifact.mjs @@ -11,11 +11,13 @@ const platformRoot = resolve(scriptDir, "../.."); const devicePlaneRoot = platformRoot; const managerRoot = resolve(platformRoot, "apps/device-manager"); const artifactDir = resolve(process.env.NODEDC_DEPLOY_ARTIFACT_DIR || resolve(scriptDir, "../deploy-artifacts")); -const [patchId = "device-manager-release-v8-20260822-039", ...extra] = process.argv.slice(2); +const [patchId = "device-manager-release-v9-20260822-041", ...extra] = process.argv.slice(2); if (extra.length || !/^[A-Za-z0-9._-]{1,96}$/.test(patchId)) throw new Error("usage: build-device-manager-control-plane-artifact.mjs [patch-id]"); -const descriptorPath = patchId.startsWith("device-manager-release-v8-") - ? "deployment/device-manager-release-v8.json" +const descriptorPath = patchId.startsWith("device-manager-release-v9-") + ? "deployment/device-manager-release-v9.json" + : patchId.startsWith("device-manager-release-v8-") + ? "deployment/device-manager-release-v8.json" : patchId.startsWith("device-manager-release-v7-") ? "deployment/device-manager-release-v7.json" : patchId.startsWith("device-manager-release-v6-") @@ -34,7 +36,8 @@ const isV5 = descriptorPath.endsWith("release-v5.json"); const isV6 = descriptorPath.endsWith("release-v6.json"); const isV7 = descriptorPath.endsWith("release-v7.json"); const isV8 = descriptorPath.endsWith("release-v8.json"); -const isPersistent = isV4 || isV5 || isV6 || isV7 || isV8; +const isV9 = descriptorPath.endsWith("release-v9.json"); +const isPersistent = isV4 || isV5 || isV6 || isV7 || isV8 || isV9; const isManagerOnly = isV3 || isPersistent; const composeSource = resolve(devicePlaneRoot, "docker-compose.device-manager.yml"); const composeSourceSha256 = createHash("sha256").update(await readFile(composeSource)).digest("hex"); @@ -173,7 +176,28 @@ try { : "restore-preapply-snapshot") ); if (commonContractInvalid) throw new Error("device_manager_activation_successor_contract_mismatch"); - if (descriptorPath.endsWith("release-v8.json")) { + if (descriptorPath.endsWith("release-v9.json")) { + if ( + descriptor.schemaVersion !== "nodedc.device-plane.device-manager-release.v9" + || descriptor.predecessor?.kind !== "release" + || descriptor.predecessor?.patchId !== "device-manager-release-v8-20260822-039" + || descriptor.predecessor?.artifactSha256 !== "30a83d4c6b5c029c96c4af19fa558e0ae75e4b60bc897881457304bd17e0e465" + || descriptor.controlCorePredecessor?.patchId !== "device-control-core-release-v3-20260822-040" + || descriptor.controlCorePredecessor?.artifactSha256 !== "08448a56cdf391076f92c5242e368fd0033b420167874645eebfa1f22184ee92" + || descriptor.edgeChannelPredecessor?.patchId !== "device-edge-core-channel-upgrade-v4-20260812-023" + || descriptor.edgeChannelPredecessor?.artifactSha256 !== "c10d5b6b7d55ab239f85b6c8130e34ce9f84985e3b46e6e5534733156c7982fc" + || descriptor.telemetryWorkspace !== "mission-core-compute-module-parity-v1" + || descriptor.telemetryNavigation !== "full-workspace-back-navigation-v1" + || descriptor.telemetryPollInterval !== "three-seconds" + || descriptor.telemetryFreshness !== "fifteen-seconds-missing-stale-not-unhealthy" + || descriptor.telemetryOntologyProjection !== "observation-observed-property-provenance-freshness-v1" + || descriptor.telemetryAgent !== "telegraf-host-observer-v1" + || descriptor.interactiveShell !== "disabled-pending-managed-session-boundary" + || descriptor.rollback !== "restore-preapply-snapshot-preserve-manager-data" + || descriptor.gelios !== "untouched-legacy-only" + ) throw new Error("device_manager_v9_host_telemetry_workspace_contract_mismatch"); + await validateFaviconBundle(payload, "device_manager_v9"); + } else if (descriptorPath.endsWith("release-v8.json")) { if ( descriptor.schemaVersion !== "nodedc.device-plane.device-manager-release.v8" || descriptor.predecessor?.kind !== "release" diff --git a/infra/deploy-runner/nodedc-b2-vps-deploy b/infra/deploy-runner/nodedc-b2-vps-deploy index d7eb0ed..9dcc27f 100755 --- a/infra/deploy-runner/nodedc-b2-vps-deploy +++ b/infra/deploy-runner/nodedc-b2-vps-deploy @@ -46,6 +46,8 @@ RELAY_USER = "nodedc-relay" RELAY_GROUP = "nodedc-relay" CHANNEL_USER = "nodedc-channel" CHANNEL_GROUP = "nodedc-channel" +TELEMETRY_USER = "nodedc-telemetry" +TELEMETRY_GROUP = "nodedc-telemetry" TAILSCALE_REQUIRED_TAG = "tag:device-edge-vps" MANAGEMENT_KEY_FINGERPRINT = ( "SHA256:DYYy1E3DaxIQGC0jnsW6SP7gXdBHUy3A1zn4pvgVUEw" @@ -67,10 +69,19 @@ TAILSCALE_ARCHIVE_SHA256 = ( NODE_BIN_SHA256 = "3517c2df0b2f8cd7f422b4b8450ef81c6889f08eb03e281d6de9079b15e6a327" TAILSCALE_BIN_SHA256 = "58b0fa0907677ea6afe0d3022cc3e99b1a03f39a7ed60144843ed38252e00c80" TAILSCALED_BIN_SHA256 = "5f17b092bac92326325f6c4ffd9991fad3c073975abe412d02ee68721a500394" +TELEGRAF_VERSION = "1.38.4" +TELEGRAF_ARCHIVE = "telegraf-1.38.4_linux_amd64.tar.gz" +TELEGRAF_ARCHIVE_SHA256 = ( + "81857e9745ebf26e058b6fdc27b9b2c210fd1fe61e57d7fad3d4bb9131f60041" +) +TELEGRAF_BIN_SHA256 = ( + "0643b582546eb9c70d99a9646b3e49e25ccdea4f78ab06fd9096ed71dba1babb" +) NODE_BIN = LIVE_ROOT / "runtime/node/bin/node" TAILSCALE_BIN = LIVE_ROOT / "runtime/tailscale/tailscale" TAILSCALED_BIN = LIVE_ROOT / "runtime/tailscale/tailscaled" +TELEGRAF_BIN = LIVE_ROOT / "runtime/telegraf/usr/bin/telegraf" TAILSCALE_SOCKET = Path("/run/nodedc-b2-vps/tailscaled.sock") TAILSCALE_STATE = Path("/var/lib/nodedc-b2-vps/tailscale/tailscaled.state") TRUST_ROOT = Path("/var/lib/nodedc-b2-vps/trust") @@ -102,6 +113,10 @@ CHANNEL_CORE_CERTIFICATE = CHANNEL_TRUST_ROOT / "core-certificate.pem" CHANNEL_RUNTIME_CONFIG = CHANNEL_TRUST_ROOT / "runtime.json" CHANNEL_HEALTH_PORT = 18222 CHANNEL_PUBLIC_PORT = 443 +HOST_TELEMETRY_PORT = 18223 +HOST_TELEMETRY_UNIT = Path( + "/etc/systemd/system/nodedc-host-telemetry-agent.service" +) CORE_CHANNEL_ACCEPTED_PATCH = "device-edge-vps-core-channel-20260812-010" CORE_CHANNEL_ACCEPTED_SHA256 = ( "c8ef3c4bb45850cad32e881eba081bc4c891c2886e5500d02cb94616d82353f3" @@ -116,6 +131,12 @@ TRACKER_INGRESS_ACCEPTED_PATCH = "device-edge-vps-tracker-ingress-20260812-012" TRACKER_INGRESS_ACCEPTED_SHA256 = ( "290acef118839c6b0c31aac864c47da1832a289537366af9322d4624a1dd81ec" ) +COMMAND_TRANSPORT_ACCEPTED_PATCH = ( + "device-edge-vps-command-transport-20260812-013" +) +COMMAND_TRANSPORT_ACCEPTED_SHA256 = ( + "c7486ec879681ddd706f229628c8556ca8c9ccc4f152a85debb409c302759ef" +) FOUNDATION_ENTRIES = ( "vps/config/00-nodedc-b2-vps.conf", @@ -181,6 +202,21 @@ COMMAND_TRANSPORT_ENTRIES = ( "vps/edge-process/device-edge-runtime.mjs", "deployment/device-edge-vps-command-transport-v1.json", ) +HOST_TELEMETRY_ENTRIES = ( + "packages/device-edge-channel-contract/package.json", + "packages/device-edge-channel-contract/src", + "packages/infrastructure-telemetry-contract/package.json", + "packages/infrastructure-telemetry-contract/src", + "services/device-edge-channel/package.json", + "services/device-edge-channel/src", + "vps/edge-process/device-edge-runtime.mjs", + "vps/edge-process/host-telemetry-runtime.mjs", + "vps/config/nodedc-host-telemetry-telegraf.conf", + "vps/systemd/nodedc-device-edge-runtime.service", + "vps/systemd/nodedc-host-telemetry-agent.service", + "deployment/device-edge-vps-host-telemetry-v1.json", + f"vendor/{TELEGRAF_ARCHIVE}", +) PHASE_ENTRIES = { "foundation": FOUNDATION_ENTRIES, @@ -191,6 +227,7 @@ PHASE_ENTRIES = { "tailscale-retirement": TAILSCALE_RETIREMENT_ENTRIES, "tracker-ingress": TRACKER_INGRESS_ENTRIES, "command-transport": COMMAND_TRANSPORT_ENTRIES, + "host-telemetry": HOST_TELEMETRY_ENTRIES, } SUPERSEDED_TRANSPORT_PHASES = frozenset({"backhaul", "relay"}) @@ -312,6 +349,35 @@ PHASE_FILE_SHA256 = { "deployment/device-edge-vps-command-transport-v1.json": "971166143fe954b9c5043cce9a464d17efbc87933da1405b2517a4693a7bed09", }, + "host-telemetry": { + "deployment/device-edge-vps-host-telemetry-v1.json": + "69d4ed7c7462d982e53fa2f0688d7a129952811b96dcb1bf4ae3721f235fc406", + "packages/device-edge-channel-contract/package.json": + "57d5349b5dcef2cacd4f3e4fad010359a65d59f5f903eff07d89f67c497f97c0", + "packages/device-edge-channel-contract/src/index.mjs": + "58a53836495dc891de191c6022cf7661a2198deb9fff7055a5cc23a36ddf49d2", + "packages/infrastructure-telemetry-contract/package.json": + "5ef70204acc9a2bee68be959347487dc2e8fb7fbe8bd88731033e7ab204acf34", + "packages/infrastructure-telemetry-contract/src/index.mjs": + "6d4b60b79e131380fcec403cf8842a3612614b5540c7052020e84d4fc8a9360f", + "services/device-edge-channel/package.json": + "bdf502be43b62bdd6db05b022a532d93ba954277ac5143d6058d2f27f6a2e9d2", + "services/device-edge-channel/src/runtime.mjs": + "4c0e874b2f1161910d3abde9a07f4a7744ffeece325303cc9985521a0eafb47b", + "services/device-edge-channel/src/server.mjs": + "a82057218bb368ab926404f90a19cc17c0359b57dc890f38a1324ab8c497c17b", + "vps/config/nodedc-host-telemetry-telegraf.conf": + "596e386d1e37b8178ccc660760567f5a8914ae7bef43b3603b31846c73154972", + "vps/edge-process/device-edge-runtime.mjs": + "0fc32e8c71a028777b0945ea6a0dbab22b0244caab1d7f616702c8a4a143c597", + "vps/edge-process/host-telemetry-runtime.mjs": + "ce1c6f368199d6c91e8a7496e0e8388e3c390018f2695107bc2877eced5e566d", + "vps/systemd/nodedc-device-edge-runtime.service": + "88cd8d34df254f8daa1d82f6bcd175fb061376b65491ee6ac90b296fce675dfb", + "vps/systemd/nodedc-host-telemetry-agent.service": + "0d3aa1644af528ebd4cca7a95fb1bb89fe14544a64fb9729fd469296c4343506", + f"vendor/{TELEGRAF_ARCHIVE}": TELEGRAF_ARCHIVE_SHA256, + }, } # Exact immutable baselines from terminally accepted predecessor artifacts. @@ -472,7 +538,11 @@ def validate_payload(payload: Path, phase: str): descriptor.get("component") != COMPONENT or descriptor.get("runtimeHost") != RUNTIME_HOST or descriptor.get("commandTransport") - != ("typed-service-ping-v1" if phase == "command-transport" else "disabled") + != ( + "typed-service-ping-v1" + if phase in {"command-transport", "host-telemetry"} + else "disabled" + ) or not str(descriptor.get("gelios", "")).startswith("untouched") or not descriptor.get("rollback") ): @@ -737,6 +807,25 @@ def current_phase_preflight(phase: str): if (LIVE_ROOT / COMMAND_TRANSPORT_ENTRIES[-1]).exists(): die("VPS command transport target path already exists") return {"predecessor": "accepted-tracker-ingress-012"} + if phase == "host-telemetry": + command_record = applied_phase_record("command-transport") + if ( + command_record.get("patch") != COMMAND_TRANSPORT_ACCEPTED_PATCH + or command_record.get("sha256") + != COMMAND_TRANSPORT_ACCEPTED_SHA256 + ): + die("VPS host telemetry command transport predecessor mismatch") + source_file_state("command-transport") + validate_command_transport_runtime() + if ( + HOST_TELEMETRY_UNIT.exists() + or TELEGRAF_BIN.exists() + or user_exists(TELEMETRY_USER) + or (LIVE_ROOT / HOST_TELEMETRY_ENTRIES[-2]).exists() + ): + die("VPS host telemetry target boundary already exists") + assert_port_closed(HOST_TELEMETRY_PORT) + return {"predecessor": "accepted-command-transport-013"} validate_foundation_runtime( require_running_tailnet=phase in {"backhaul", "relay"}, expected_key_user=BACKHAUL_USER if phase == "relay" else SERVICE_USER, @@ -848,6 +937,12 @@ def backup_targets_for_phase(phase: str): return common + [CHANNEL_UNIT, NFTABLES_CONFIG] if phase == "command-transport": return common + [CHANNEL_UNIT, NFTABLES_CONFIG] + if phase == "host-telemetry": + return common + [ + CHANNEL_UNIT, + HOST_TELEMETRY_UNIT, + TELEGRAF_BIN.parent, + ] return common + [RELAY_UNIT, NFTABLES_CONFIG] @@ -887,7 +982,13 @@ def create_backup(patch_id: str, phase: str): "serviceUserExisted": user_exists(), "serviceUsersExisted": { name: user_exists(name) - for name in (SERVICE_USER, BACKHAUL_USER, RELAY_USER, CHANNEL_USER) + for name in ( + SERVICE_USER, + BACKHAUL_USER, + RELAY_USER, + CHANNEL_USER, + TELEMETRY_USER, + ) }, "services": { name: { @@ -902,6 +1003,7 @@ def create_backup(patch_id: str, phase: str): "nodedc-b2-backhaul.service", "nodedc-b2-relay.service", "nodedc-device-edge-channel.service", + "nodedc-host-telemetry-agent.service", ) }, } @@ -1336,6 +1438,32 @@ def apply_command_transport(_payload: Path): validate_command_transport_runtime() +def apply_host_telemetry(_payload: Path): + ensure_service_user( + TELEMETRY_USER, + "/var/lib/nodedc-b2-vps/telemetry", + ) + extract_vendor_binary( + LIVE_ROOT / f"vendor/{TELEGRAF_ARCHIVE}", + f"telegraf-{TELEGRAF_VERSION}/usr/bin/telegraf", + TELEGRAF_BIN, + ) + install_file( + LIVE_ROOT / "vps/systemd/nodedc-device-edge-runtime.service", + CHANNEL_UNIT, + 0o644, + ) + install_file( + LIVE_ROOT / "vps/systemd/nodedc-host-telemetry-agent.service", + HOST_TELEMETRY_UNIT, + 0o644, + ) + systemctl("daemon-reload") + systemctl("restart", "nodedc-device-edge-channel.service") + systemctl("enable", "--now", "nodedc-host-telemetry-agent.service") + validate_host_telemetry_runtime() + + def sshd_effective(): return run(["/usr/sbin/sshd", "-T"]).stdout.lower() @@ -1777,6 +1905,126 @@ def validate_command_transport_runtime(): return health +def validate_host_telemetry_runtime(): + source_file_state("foundation") + source_file_state("tailscale-retirement") + source_file_state("host-telemetry") + command_record = applied_phase_record("command-transport") + if ( + command_record.get("patch") != COMMAND_TRANSPORT_ACCEPTED_PATCH + or command_record.get("sha256") != COMMAND_TRANSPORT_ACCEPTED_SHA256 + ): + die("host telemetry command transport identity mismatch") + binary = assert_regular_nonsymlink( + TELEGRAF_BIN, + "VPS Telegraf runtime", + ) + if ( + binary.st_uid != 0 + or binary.st_gid != 0 + or (binary.st_mode & 0o777) != 0o755 + or sha256_file(TELEGRAF_BIN) != TELEGRAF_BIN_SHA256 + ): + die("VPS Telegraf runtime identity mismatch") + version = run([str(TELEGRAF_BIN), "version"]).stdout.strip() + if not version.startswith(f"Telegraf {TELEGRAF_VERSION}"): + die("VPS Telegraf version mismatch") + telemetry_account = pwd.getpwnam(TELEMETRY_USER) + channel_account = pwd.getpwnam(CHANNEL_USER) + if ( + telemetry_account.pw_shell != "/usr/sbin/nologin" + or telemetry_account.pw_uid == channel_account.pw_uid + ): + die("VPS telemetry runtime identity is not isolated") + if not service_active("nodedc-device-edge-channel.service"): + die("VPS Device Edge runtime is not active") + if not service_active("nodedc-host-telemetry-agent.service"): + die("VPS host telemetry agent is not active") + if ( + systemctl( + "is-enabled", + "nodedc-host-telemetry-agent.service", + check=False, + ).returncode != 0 + ): + die("VPS host telemetry agent is not enabled") + health = None + last_error = "no host telemetry acceptance" + for _attempt in range(60): + candidate = core_channel_health(require_accepted=True) + host_telemetry = candidate.get("hostTelemetry") or {} + if ( + host_telemetry.get("listening") is True + and host_telemetry.get("host") == "127.0.0.1" + and host_telemetry.get("port") == HOST_TELEMETRY_PORT + and host_telemetry.get("profile") == "linux-host-telegraf-v1" + and int(host_telemetry.get("accepted") or 0) >= 1 + and host_telemetry.get("lastErrorCode") is None + ): + health = candidate + break + last_error = json.dumps(host_telemetry, sort_keys=True) + time.sleep(2) + if health is None: + die(f"VPS host telemetry acceptance timeout: {last_error}") + expected = { + "ok": True, + "service": "nodedc-device-edge-runtime", + "channel": "accepted", + "trackerIngress": "telemetry-ingest", + "commandTransport": "typed-service-ping-v1", + } + for key, value in expected.items(): + if health.get(key) != value: + die(f"VPS host telemetry preserved contract mismatch: {key}") + if not port_is_open("127.0.0.1", HOST_TELEMETRY_PORT, timeout=5): + die("VPS host telemetry collector is unavailable") + if port_is_open(PUBLIC_IPV4, HOST_TELEMETRY_PORT, timeout=2): + die("VPS host telemetry collector became public") + for port in (1883, 8883): + if port_is_open("127.0.0.1", port) or port_is_open(PUBLIC_IPV4, port): + die(f"VPS forbidden MQTT listener became available: {port}") + for port in (22, CHANNEL_PUBLIC_PORT, 9921): + if not port_is_open(PUBLIC_IPV4, port, timeout=5): + die(f"VPS host telemetry preserved listener unavailable: {port}") + unit = run([ + "/usr/bin/systemctl", + "show", + "nodedc-host-telemetry-agent.service", + "--property=User,Group,NoNewPrivileges,CapabilityBoundingSet,MemoryMax,MemorySwapMax,TasksMax,LimitNOFILE", + ]).stdout + for required in ( + "User=nodedc-telemetry", + "Group=nodedc-telemetry", + "NoNewPrivileges=yes", + "CapabilityBoundingSet=", + "MemoryMax=100663296", + "MemorySwapMax=0", + "TasksMax=64", + "LimitNOFILE=512", + ): + if required not in unit: + die(f"VPS host telemetry resource boundary mismatch: {required}") + nft = run([ + "/usr/sbin/nft", + "list", + "table", + "inet", + "nodedc_b2_vps", + ]).stdout + for required in ( + "policy drop", + "tcp dport 22", + "tcp dport 443", + "tcp dport 9921", + ): + if required not in nft: + die(f"VPS host telemetry firewall contract mismatch: {required}") + if f"tcp dport {HOST_TELEMETRY_PORT}" in nft: + die("VPS host telemetry firewall exposed the collector") + return health + + def validate_relay_runtime(): validate_backhaul_runtime() source_file_state("relay") @@ -1835,6 +2083,7 @@ def restore_service_enablement(metadata): def rollback(backup: Path, phase: str): for service in ( + "nodedc-host-telemetry-agent.service", "nodedc-b2-relay.service", "nodedc-b2-backhaul.service", "nodedc-device-edge-channel.service", @@ -1866,6 +2115,9 @@ def rollback(backup: Path, phase: str): if phase == "core-channel": if not users_before.get(CHANNEL_USER, False) and user_exists(CHANNEL_USER): run(["/usr/sbin/userdel", CHANNEL_USER], check=False) + if phase == "host-telemetry": + if not users_before.get(TELEMETRY_USER, False) and user_exists(TELEMETRY_USER): + run(["/usr/sbin/userdel", TELEMETRY_USER], check=False) if phase == "foundation" and not metadata.get("serviceUserExisted"): runtime_state_root = Path("/var/lib/nodedc-b2-vps") if LIVE_ROOT.exists() and not LIVE_ROOT.is_symlink(): @@ -1955,7 +2207,7 @@ def plan_artifact(artifact_argument: str): print("runtime_composition=single-non-root-process:core-channel+universal-gateway") print("tailscale=preserved:absent") print("services=recreate:nodedc-device-edge-channel") - else: + elif phase == "command-transport": print("public_core_channel=preserved:155.212.211.15:443/tcp:tls13-mtls-h2") print(f"health=127.0.0.1:{CHANNEL_HEALTH_PORT}:combined-edge-runtime") print("public_b2_ingress=155.212.211.15:9921/tcp:bidirectional-session") @@ -1964,16 +2216,27 @@ def plan_artifact(artifact_argument: str): print("runtime_composition=single-non-root-process:core-channel+universal-gateway") print("tailscale=preserved:absent") print("services=recreate:nodedc-device-edge-channel") + else: + print("public_core_channel=preserved:155.212.211.15:443/tcp:tls13-mtls-h2") + print(f"health=127.0.0.1:{CHANNEL_HEALTH_PORT}:combined-edge-runtime") + print(f"host_telemetry_collector=127.0.0.1:{HOST_TELEMETRY_PORT}:loopback-only") + print(f"host_telemetry_agent=telegraf:{TELEGRAF_VERSION}:sha256:{TELEGRAF_ARCHIVE_SHA256}") + print("host_telemetry_interval=2s") + print("host_telemetry_transport=existing-pinned-mtls-core-channel") + print("host_telemetry_ontology=observation.observation=>infrastructure.host") + print("mqtt=disabled") + print("public_port_set=preserved:22,443,9921") + print("services=recreate:nodedc-device-edge-channel+create:nodedc-host-telemetry-agent") print( "command_transport=typed-service-ping-v1:allowlisted-adapter-only" - if phase == "command-transport" + if phase in {"command-transport", "host-telemetry"} else "command_transport=disabled" ) - if phase == "command-transport": + if phase in {"command-transport", "host-telemetry"}: print("command_catalog=allowlisted-adapter-typed-commands-only") print( "gelios=untouched-legacy-only" - if phase == "command-transport" + if phase in {"command-transport", "host-telemetry"} else "gelios=untouched" ) print("dns=unchanged") @@ -2013,8 +2276,10 @@ def apply_artifact(artifact_argument: str): apply_tailscale_retirement(loaded["payload"]) elif loaded["phase"] == "tracker-ingress": apply_tracker_ingress(loaded["payload"]) - else: + elif loaded["phase"] == "command-transport": apply_command_transport(loaded["payload"]) + else: + apply_host_telemetry(loaded["payload"]) archived = archive_artifact(loaded["artifact"], APPLIED_ROOT) record = { diff --git a/infra/deploy-runner/test_device_edge_vps_artifact.py b/infra/deploy-runner/test_device_edge_vps_artifact.py index 76a4484..1ab982c 100644 --- a/infra/deploy-runner/test_device_edge_vps_artifact.py +++ b/infra/deploy-runner/test_device_edge_vps_artifact.py @@ -82,7 +82,7 @@ class DeviceEdgeVpsArtifactTest(unittest.TestCase): RUNNER.preflight({"phase": phase}) def test_accepted_shared_source_phases_cannot_be_rebuilt(self): - for phase in ("core-channel", "tracker-ingress"): + for phase in ("core-channel", "tracker-ingress", "command-transport"): with self.subTest(phase=phase), tempfile.TemporaryDirectory( prefix=f"nodedc-vps-frozen-{phase}-" ) as directory: @@ -102,6 +102,7 @@ class DeviceEdgeVpsArtifactTest(unittest.TestCase): for name, digest in ( (RUNNER.NODE_ARCHIVE, RUNNER.NODE_ARCHIVE_SHA256), (RUNNER.TAILSCALE_ARCHIVE, RUNNER.TAILSCALE_ARCHIVE_SHA256), + (RUNNER.TELEGRAF_ARCHIVE, RUNNER.TELEGRAF_ARCHIVE_SHA256), ): path = DEFAULT_RUNTIME_CACHE / name if not path.is_file(): @@ -114,10 +115,19 @@ class DeviceEdgeVpsArtifactTest(unittest.TestCase): + ", ".join(missing) ) + def require_telegraf_cache(self): + archive = DEFAULT_RUNTIME_CACHE / RUNNER.TELEGRAF_ARCHIVE + if not archive.is_file(): + self.skipTest(f"immutable Telegraf runtime is not available: {archive}") + self.assertEqual( + hashlib.sha256(archive.read_bytes()).hexdigest(), + RUNNER.TELEGRAF_ARCHIVE_SHA256, + ) + def test_builders_are_deterministic_narrow_and_secret_free(self): - self.require_runtime_cache() + self.require_telegraf_cache() for phase in ( - "command-transport", + "host-telemetry", ): with self.subTest(phase=phase), tempfile.TemporaryDirectory( prefix=f"nodedc-vps-{phase}-" @@ -183,7 +193,7 @@ class DeviceEdgeVpsArtifactTest(unittest.TestCase): self.assertIn("runtime_digest_mismatch", result.stderr) def test_runner_loads_each_exact_phase(self): - self.require_runtime_cache() + self.require_telegraf_cache() with tempfile.TemporaryDirectory(prefix="nodedc-vps-load-") as directory: inbox = Path(directory) / "inbox" inbox.mkdir() @@ -191,7 +201,7 @@ class DeviceEdgeVpsArtifactTest(unittest.TestCase): RUNNER.INBOX_ROOT = inbox try: for phase in ( - "command-transport", + "host-telemetry", ): result = self.build( inbox, @@ -416,14 +426,15 @@ class DeviceEdgeVpsArtifactTest(unittest.TestCase): self.assertIn("command_transport=disabled", rendered) self.assertIn("gelios=untouched", rendered) - def test_command_transport_plan_is_typed_single_process_and_bounded(self): - with tempfile.TemporaryDirectory(prefix="nodedc-vps-command-plan-") as directory: + def test_host_telemetry_plan_is_loopback_agent_over_existing_mtls(self): + self.require_telegraf_cache() + with tempfile.TemporaryDirectory(prefix="nodedc-vps-host-telemetry-plan-") as directory: inbox = Path(directory) / "inbox" inbox.mkdir() result = self.build( inbox, - "command-transport", - "device-edge-vps-command-transport-plan-001", + "host-telemetry", + "device-edge-vps-host-telemetry-plan-001", ) self.assertEqual(result.returncode, 0, result.stderr) artifact = Path(json.loads(result.stdout)["artifact"]) @@ -433,7 +444,7 @@ class DeviceEdgeVpsArtifactTest(unittest.TestCase): with patch.object(RUNNER, "assert_root"), patch.object( RUNNER, "preflight", - return_value={"predecessor": "accepted-tracker-ingress-012"}, + return_value={"predecessor": "accepted-command-transport-013"}, ), patch("builtins.print") as output: RUNNER.plan_artifact(str(artifact)) finally: @@ -442,12 +453,13 @@ class DeviceEdgeVpsArtifactTest(unittest.TestCase): " ".join(str(arg) for arg in call.args) for call in output.call_args_list ) - self.assertIn("phase=command-transport", rendered) - self.assertIn("predecessor=accepted-tracker-ingress-012", rendered) + self.assertIn("phase=host-telemetry", rendered) + self.assertIn("predecessor=accepted-command-transport-013", rendered) + self.assertIn("host_telemetry_agent=telegraf:1.38.4", rendered) + self.assertIn("host_telemetry_collector=127.0.0.1:18223:loopback-only", rendered) + self.assertIn("host_telemetry_transport=existing-pinned-mtls-core-channel", rendered) + self.assertIn("mqtt=disabled", rendered) self.assertIn("command_transport=typed-service-ping-v1", rendered) - self.assertIn("command_catalog=allowlisted-adapter-typed-commands-only", rendered) - self.assertIn("runtime_composition=single-non-root-process", rendered) - self.assertIn("public_b2_ingress=155.212.211.15:9921/tcp:bidirectional-session", rendered) self.assertIn("gelios=untouched-legacy-only", rendered) def test_publish_payload_preserves_unselected_executable_modes(self): diff --git a/packages/device-edge-channel-contract/src/index.mjs b/packages/device-edge-channel-contract/src/index.mjs index 5a855cf..cdcaea6 100644 --- a/packages/device-edge-channel-contract/src/index.mjs +++ b/packages/device-edge-channel-contract/src/index.mjs @@ -22,6 +22,7 @@ export const EDGE_TO_CORE_MESSAGE_KINDS = Object.freeze([ "adapter.message", "delivery.acknowledged", "command.status", + "host.telemetry.observed", "channel.counters", ]); diff --git a/packages/infrastructure-telemetry-contract/package.json b/packages/infrastructure-telemetry-contract/package.json new file mode 100644 index 0000000..68e424d --- /dev/null +++ b/packages/infrastructure-telemetry-contract/package.json @@ -0,0 +1,10 @@ +{ + "name": "@nodedc/infrastructure-telemetry-contract", + "version": "0.1.0", + "private": true, + "type": "module", + "exports": "./src/index.mjs", + "engines": { + "node": ">=20" + } +} diff --git a/packages/infrastructure-telemetry-contract/src/index.mjs b/packages/infrastructure-telemetry-contract/src/index.mjs new file mode 100644 index 0000000..a192c51 --- /dev/null +++ b/packages/infrastructure-telemetry-contract/src/index.mjs @@ -0,0 +1,380 @@ +export const HOST_TELEMETRY_SCHEMA = + "nodedc.infrastructure.host-telemetry.v1"; + +export const HOST_TELEMETRY_PROFILE = "linux-host-telegraf-v1"; + +const REF_RE = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/; +const ISO_TIMESTAMP_RE = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{3})?Z$/; + +export function telegrafBatchToHostTelemetry(input, context = {}) { + const metrics = telegrafMetrics(input); + const observedAt = latestMetricTimestamp(metrics) ?? normalizeTimestamp( + context.observedAt ?? new Date().toISOString(), + "observed_at", + ); + const byName = new Map(); + for (const metric of metrics) { + const values = byName.get(metric.name) ?? []; + values.push(metric); + byName.set(metric.name, values); + } + + const cpu = metricWithTag(byName.get("cpu"), "cpu", "cpu-total") + ?? firstMetric(byName.get("cpu")); + const memory = firstMetric(byName.get("mem")); + const swap = firstMetric(byName.get("swap")); + const system = firstMetric(byName.get("system")); + const processes = firstMetric(byName.get("processes")); + const systemCpu = firstMetric(byName.get("system_cpu")); + + return normalizeHostTelemetrySnapshot({ + schemaVersion: HOST_TELEMETRY_SCHEMA, + profile: HOST_TELEMETRY_PROFILE, + hostKey: context.hostKey, + observedAt, + source: { + agent: "telegraf", + agentVersion: context.agentVersion ?? "1.38.4", + collectorRef: context.collectorRef ?? "service:nodedc-host-telemetry-agent", + }, + hardware: { + hostname: context.hostname ?? metricHost(metrics), + architecture: context.architecture ?? null, + platform: context.platform ?? "linux", + kernelRelease: context.kernelRelease ?? null, + cpuModel: context.cpuModel ?? null, + logicalProcessors: finiteInteger( + context.logicalProcessors ?? systemCpu?.fields.cpu_count, + ), + }, + cpu: { + usagePercent: finiteNumber(cpu?.fields.usage_active) + ?? percentFromIdle(cpu?.fields.usage_idle), + load1: finiteNumber(system?.fields.load1), + load5: finiteNumber(system?.fields.load5), + load15: finiteNumber(system?.fields.load15), + }, + memory: { + totalBytes: finiteInteger(memory?.fields.total), + availableBytes: finiteInteger(memory?.fields.available), + usedBytes: finiteInteger(memory?.fields.used), + usedPercent: finiteNumber(memory?.fields.used_percent), + }, + swap: { + totalBytes: finiteInteger(swap?.fields.total), + freeBytes: finiteInteger(swap?.fields.free), + usedBytes: finiteInteger(swap?.fields.used), + usedPercent: finiteNumber(swap?.fields.used_percent), + }, + system: { + uptimeSeconds: finiteInteger(system?.fields.uptime), + users: finiteInteger(system?.fields.n_users), + processes: { + total: finiteInteger(processes?.fields.total), + running: finiteInteger(processes?.fields.running), + sleeping: finiteInteger(processes?.fields.sleeping), + blocked: finiteInteger(processes?.fields.blocked), + zombies: finiteInteger(processes?.fields.zombies), + }, + }, + disks: (byName.get("disk") ?? []).map((metric) => ({ + device: textOrNull(metric.tags.device), + mount: textOrNull(metric.tags.path), + filesystem: textOrNull(metric.tags.fstype), + totalBytes: finiteInteger(metric.fields.total), + freeBytes: finiteInteger(metric.fields.free), + usedBytes: finiteInteger(metric.fields.used), + usedPercent: finiteNumber(metric.fields.used_percent), + })), + network: (byName.get("net") ?? []).map((metric) => ({ + interface: textOrNull(metric.tags.interface), + bytesReceived: finiteInteger(metric.fields.bytes_recv), + bytesSent: finiteInteger(metric.fields.bytes_sent), + packetsReceived: finiteInteger(metric.fields.packets_recv), + packetsSent: finiteInteger(metric.fields.packets_sent), + errorsReceived: finiteInteger(metric.fields.err_in), + errorsSent: finiteInteger(metric.fields.err_out), + droppedReceived: finiteInteger(metric.fields.drop_in), + droppedSent: finiteInteger(metric.fields.drop_out), + })), + services: (byName.get("systemd_units") ?? []).map((metric) => ({ + name: textOrNull(metric.tags.name), + loadState: textOrNull(metric.tags.load), + activeState: textOrNull(metric.tags.active), + subState: textOrNull(metric.tags.sub), + memoryBytes: finiteInteger(metric.fields.mem_current), + restarts: finiteInteger(metric.fields.restarts), + pid: finiteInteger(metric.fields.pid), + })), + }); +} + +export function normalizeHostTelemetrySnapshot(input) { + assertPlainObject(input, "host_telemetry"); + if (input.schemaVersion !== HOST_TELEMETRY_SCHEMA) { + throw new TypeError("host_telemetry_schema_invalid"); + } + if (input.profile !== HOST_TELEMETRY_PROFILE) { + throw new TypeError("host_telemetry_profile_invalid"); + } + const snapshot = { + schemaVersion: HOST_TELEMETRY_SCHEMA, + profile: HOST_TELEMETRY_PROFILE, + hostKey: normalizeRef(input.hostKey, "host_key"), + observedAt: normalizeTimestamp(input.observedAt, "observed_at"), + source: normalizeSource(input.source), + hardware: normalizeHardware(input.hardware), + cpu: normalizeCpu(input.cpu), + memory: normalizeMemory(input.memory, "memory"), + swap: normalizeMemory(input.swap, "swap"), + system: normalizeSystem(input.system), + disks: normalizeArray(input.disks, normalizeDisk, 32), + network: normalizeArray(input.network, normalizeNetwork, 64), + services: normalizeArray(input.services, normalizeService, 64), + }; + return deepFreeze(snapshot); +} + +function telegrafMetrics(input) { + const candidate = Array.isArray(input) + ? input + : input && typeof input === "object" && Array.isArray(input.metrics) + ? input.metrics + : input && typeof input === "object" + ? [input] + : null; + if (!candidate || candidate.length < 1 || candidate.length > 512) { + throw new TypeError("host_telemetry_telegraf_batch_invalid"); + } + return candidate.map((metric) => { + assertPlainObject(metric, "host_telemetry_telegraf_metric"); + assertPlainObject(metric.fields, "host_telemetry_telegraf_fields"); + const name = text(metric.name, 1, 80, "host_telemetry_telegraf_name_invalid"); + const tags = metric.tags == null ? {} : metric.tags; + assertPlainObject(tags, "host_telemetry_telegraf_tags"); + return { name, fields: { ...metric.fields }, tags: { ...tags }, timestamp: metric.timestamp }; + }); +} + +function latestMetricTimestamp(metrics) { + let latest = null; + for (const metric of metrics) { + const raw = Number(metric.timestamp); + if (!Number.isFinite(raw) || raw <= 0) continue; + const milliseconds = raw > 10_000_000_000 ? raw / 1_000_000 : raw * 1000; + if (!Number.isFinite(milliseconds)) continue; + const value = new Date(milliseconds).toISOString(); + if (!latest || value > latest) latest = value; + } + return latest; +} + +function metricWithTag(metrics = [], key, value) { + return metrics.find((metric) => metric.tags[key] === value) ?? null; +} + +function firstMetric(metrics = []) { + return metrics[0] ?? null; +} + +function metricHost(metrics) { + for (const metric of metrics) { + if (typeof metric.tags.host === "string" && metric.tags.host.trim()) { + return metric.tags.host.trim().slice(0, 160); + } + } + return null; +} + +function normalizeSource(input) { + assertPlainObject(input, "host_telemetry_source"); + return { + agent: text(input.agent, 1, 64, "host_telemetry_agent_invalid"), + agentVersion: text(input.agentVersion, 1, 64, "host_telemetry_agent_version_invalid"), + collectorRef: normalizeRef(input.collectorRef, "collector_ref"), + }; +} + +function normalizeHardware(input) { + assertPlainObject(input, "host_telemetry_hardware"); + return { + hostname: optionalText(input.hostname, 160), + architecture: optionalText(input.architecture, 64), + platform: optionalText(input.platform, 64), + kernelRelease: optionalText(input.kernelRelease, 160), + cpuModel: optionalText(input.cpuModel, 256), + logicalProcessors: optionalInteger(input.logicalProcessors, 1_024), + }; +} + +function normalizeCpu(input) { + assertPlainObject(input, "host_telemetry_cpu"); + return { + usagePercent: optionalPercent(input.usagePercent), + load1: optionalNumber(input.load1, 0, 100_000), + load5: optionalNumber(input.load5, 0, 100_000), + load15: optionalNumber(input.load15, 0, 100_000), + }; +} + +function normalizeMemory(input, field) { + assertPlainObject(input, `host_telemetry_${field}`); + return { + totalBytes: optionalInteger(input.totalBytes, Number.MAX_SAFE_INTEGER), + availableBytes: optionalInteger(input.availableBytes, Number.MAX_SAFE_INTEGER), + freeBytes: optionalInteger(input.freeBytes, Number.MAX_SAFE_INTEGER), + usedBytes: optionalInteger(input.usedBytes, Number.MAX_SAFE_INTEGER), + usedPercent: optionalPercent(input.usedPercent), + }; +} + +function normalizeSystem(input) { + assertPlainObject(input, "host_telemetry_system"); + assertPlainObject(input.processes, "host_telemetry_processes"); + return { + uptimeSeconds: optionalInteger(input.uptimeSeconds, Number.MAX_SAFE_INTEGER), + users: optionalInteger(input.users, 1_000_000), + processes: { + total: optionalInteger(input.processes.total, 1_000_000), + running: optionalInteger(input.processes.running, 1_000_000), + sleeping: optionalInteger(input.processes.sleeping, 1_000_000), + blocked: optionalInteger(input.processes.blocked, 1_000_000), + zombies: optionalInteger(input.processes.zombies, 1_000_000), + }, + }; +} + +function normalizeDisk(input) { + assertPlainObject(input, "host_telemetry_disk"); + return { + device: optionalText(input.device, 256), + mount: optionalText(input.mount, 512), + filesystem: optionalText(input.filesystem, 64), + totalBytes: optionalInteger(input.totalBytes, Number.MAX_SAFE_INTEGER), + freeBytes: optionalInteger(input.freeBytes, Number.MAX_SAFE_INTEGER), + usedBytes: optionalInteger(input.usedBytes, Number.MAX_SAFE_INTEGER), + usedPercent: optionalPercent(input.usedPercent), + }; +} + +function normalizeNetwork(input) { + assertPlainObject(input, "host_telemetry_network"); + return { + interface: optionalText(input.interface, 64), + bytesReceived: optionalInteger(input.bytesReceived, Number.MAX_SAFE_INTEGER), + bytesSent: optionalInteger(input.bytesSent, Number.MAX_SAFE_INTEGER), + packetsReceived: optionalInteger(input.packetsReceived, Number.MAX_SAFE_INTEGER), + packetsSent: optionalInteger(input.packetsSent, Number.MAX_SAFE_INTEGER), + errorsReceived: optionalInteger(input.errorsReceived, Number.MAX_SAFE_INTEGER), + errorsSent: optionalInteger(input.errorsSent, Number.MAX_SAFE_INTEGER), + droppedReceived: optionalInteger(input.droppedReceived, Number.MAX_SAFE_INTEGER), + droppedSent: optionalInteger(input.droppedSent, Number.MAX_SAFE_INTEGER), + }; +} + +function normalizeService(input) { + assertPlainObject(input, "host_telemetry_service"); + return { + name: optionalText(input.name, 160), + loadState: optionalText(input.loadState, 64), + activeState: optionalText(input.activeState, 64), + subState: optionalText(input.subState, 64), + memoryBytes: optionalInteger(input.memoryBytes, Number.MAX_SAFE_INTEGER), + restarts: optionalInteger(input.restarts, Number.MAX_SAFE_INTEGER), + pid: optionalInteger(input.pid, Number.MAX_SAFE_INTEGER), + }; +} + +function normalizeArray(value, mapper, maximum) { + if (!Array.isArray(value) || value.length > maximum) { + throw new TypeError("host_telemetry_collection_invalid"); + } + return value.map(mapper); +} + +function normalizeRef(value, field) { + if (typeof value !== "string" || !REF_RE.test(value)) { + throw new TypeError(`host_telemetry_${field}_invalid`); + } + return value; +} + +function normalizeTimestamp(value, field) { + if (typeof value !== "string" || !ISO_TIMESTAMP_RE.test(value)) { + throw new TypeError(`host_telemetry_${field}_invalid`); + } + const timestamp = new Date(value); + if (!Number.isFinite(timestamp.valueOf())) { + throw new TypeError(`host_telemetry_${field}_invalid`); + } + return timestamp.toISOString(); +} + +function text(value, minimum, maximum, errorCode) { + if (typeof value !== "string") throw new TypeError(errorCode); + const normalized = value.trim(); + if (normalized.length < minimum || normalized.length > maximum) { + throw new TypeError(errorCode); + } + return normalized; +} + +function optionalText(value, maximum) { + if (value == null || value === "") return null; + return text(value, 1, maximum, "host_telemetry_text_invalid"); +} + +function textOrNull(value) { + return typeof value === "string" && value.trim() ? value.trim() : null; +} + +function optionalInteger(value, maximum) { + if (value == null) return null; + const normalized = Number(value); + if (!Number.isSafeInteger(normalized) || normalized < 0 || normalized > maximum) { + throw new TypeError("host_telemetry_integer_invalid"); + } + return normalized; +} + +function optionalNumber(value, minimum, maximum) { + if (value == null) return null; + const normalized = Number(value); + if (!Number.isFinite(normalized) || normalized < minimum || normalized > maximum) { + throw new TypeError("host_telemetry_number_invalid"); + } + return normalized; +} + +function optionalPercent(value) { + return optionalNumber(value, 0, 100); +} + +function finiteInteger(value) { + const normalized = Number(value); + return Number.isSafeInteger(normalized) && normalized >= 0 ? normalized : null; +} + +function finiteNumber(value) { + const normalized = Number(value); + return Number.isFinite(normalized) && normalized >= 0 ? normalized : null; +} + +function percentFromIdle(value) { + const idle = finiteNumber(value); + return idle == null ? null : Math.max(0, Math.min(100, 100 - idle)); +} + +function assertPlainObject(value, field) { + if (!value || typeof value !== "object" || Array.isArray(value)) { + throw new TypeError(`${field}_invalid`); + } +} + +function deepFreeze(value) { + if (value && typeof value === "object" && !Object.isFrozen(value)) { + Object.freeze(value); + for (const nested of Object.values(value)) deepFreeze(nested); + } + return value; +} diff --git a/packages/infrastructure-telemetry-contract/test/contract.test.mjs b/packages/infrastructure-telemetry-contract/test/contract.test.mjs new file mode 100644 index 0000000..6a89d02 --- /dev/null +++ b/packages/infrastructure-telemetry-contract/test/contract.test.mjs @@ -0,0 +1,37 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + HOST_TELEMETRY_SCHEMA, + telegrafBatchToHostTelemetry, +} from "../src/index.mjs"; + +test("normalizes a bounded Telegraf Linux batch", () => { + const value = telegrafBatchToHostTelemetry({ metrics: [ + { name: "cpu", tags: { cpu: "cpu-total", host: "edge-01" }, fields: { usage_active: 21.5 }, timestamp: 1_777_000_000 }, + { name: "mem", tags: { host: "edge-01" }, fields: { total: 1024, available: 700, used: 324, used_percent: 31.64 }, timestamp: 1_777_000_000 }, + { name: "system", tags: { host: "edge-01" }, fields: { load1: 0.2, load5: 0.1, load15: 0.05, uptime: 120, n_users: 1 }, timestamp: 1_777_000_000 }, + { name: "net", tags: { interface: "eth0", host: "edge-01" }, fields: { bytes_recv: 100, bytes_sent: 200 }, timestamp: 1_777_000_000 }, + { name: "systemd_units", tags: { name: "nodedc-device-edge-channel.service", load: "loaded", active: "active", sub: "running" }, fields: { mem_current: 2048, restarts: 0, pid: 42 }, timestamp: 1_777_000_000 }, + ] }, { + hostKey: "robot2b-b2-edge-vps", + architecture: "x64", + kernelRelease: "6.8.0", + cpuModel: "KVM CPU", + logicalProcessors: 1, + }); + + assert.equal(value.schemaVersion, HOST_TELEMETRY_SCHEMA); + assert.equal(value.cpu.usagePercent, 21.5); + assert.equal(value.memory.totalBytes, 1024); + assert.equal(value.network[0].interface, "eth0"); + assert.equal(value.services[0].activeState, "active"); + assert.equal(Object.isFrozen(value), true); +}); + +test("rejects an oversized Telegraf batch", () => { + assert.throws( + () => telegrafBatchToHostTelemetry({ metrics: Array.from({ length: 513 }, () => ({})) }, { hostKey: "host-01" }), + /host_telemetry_telegraf_batch_invalid/, + ); +}); diff --git a/services/device-control-core/Dockerfile b/services/device-control-core/Dockerfile index b194456..b42feb2 100644 --- a/services/device-control-core/Dockerfile +++ b/services/device-control-core/Dockerfile @@ -10,6 +10,7 @@ WORKDIR /app COPY packages/device-protocol-contract ./packages/device-protocol-contract COPY packages/device-edge-channel-contract ./packages/device-edge-channel-contract +COPY packages/infrastructure-telemetry-contract ./packages/infrastructure-telemetry-contract COPY services/device-control-core ./services/device-control-core USER node diff --git a/services/device-control-core/migrations/017_infrastructure_host_telemetry.sql b/services/device-control-core/migrations/017_infrastructure_host_telemetry.sql new file mode 100644 index 0000000..6a2b6f6 --- /dev/null +++ b/services/device-control-core/migrations/017_infrastructure_host_telemetry.sql @@ -0,0 +1,52 @@ +begin; + +create table if not exists device_infrastructure_host_telemetry_samples ( + id uuid primary key, + owner_scope_id uuid not null, + project_id uuid not null, + host_id uuid not null, + service_instance_id uuid not null, + edge_id uuid not null, + observed_at timestamptz not null, + received_at timestamptz not null default now(), + expires_at timestamptz not null, + schema_version text not null + check (schema_version = 'nodedc.infrastructure.host-telemetry.v1'), + profile_ref text not null + check (profile_ref = 'linux-host-telegraf-v1'), + agent_name text not null + check (agent_name = 'telegraf'), + agent_version text not null + check (length(btrim(agent_version)) between 1 and 64), + collector_ref text not null + check (length(btrim(collector_ref)) between 3 and 160), + provenance_ref text not null + check (length(btrim(provenance_ref)) between 3 and 256), + snapshot jsonb not null, + ontology_entity_id text not null default 'observation.observation' + check (ontology_entity_id = 'observation.observation'), + ontology_catalog_hash text not null default '229c61c02a790906' + check (ontology_catalog_hash = '229c61c02a790906'), + created_at timestamptz not null default now(), + unique (edge_id, observed_at), + foreign key (host_id, project_id, owner_scope_id) + references device_infrastructure_hosts(id, project_id, owner_scope_id), + foreign key (service_instance_id, project_id, owner_scope_id) + references device_infrastructure_service_instances(id, project_id, owner_scope_id), + foreign key (edge_id) references device_edges(id), + check (expires_at > observed_at), + check (received_at >= observed_at - interval '5 minutes'), + check (jsonb_typeof(snapshot) = 'object') +); + +create index if not exists device_host_telemetry_project_host_time_idx + on device_infrastructure_host_telemetry_samples ( + project_id, + host_id, + observed_at desc + ); + +create index if not exists device_host_telemetry_retention_idx + on device_infrastructure_host_telemetry_samples (observed_at); + +commit; diff --git a/services/device-control-core/src/device-gateway-core-runtime.mjs b/services/device-control-core/src/device-gateway-core-runtime.mjs index 0c10bf6..54ed916 100644 --- a/services/device-control-core/src/device-gateway-core-runtime.mjs +++ b/services/device-control-core/src/device-gateway-core-runtime.mjs @@ -18,6 +18,9 @@ import { normalizeAdapterMessage, normalizeDiscoverySignal, } from "../../../packages/device-protocol-contract/src/index.mjs"; +import { + normalizeHostTelemetrySnapshot, +} from "../../../packages/infrastructure-telemetry-contract/src/index.mjs"; // Runtime-owned transport implementation; kept inside the deployable Core context. const CHANNEL_TRACKER_SESSION_ID = "channel:control"; @@ -277,7 +280,7 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { return; } if (envelope.messageKind === "channel.heartbeat") return; - if (["discovery.observed", "adapter.message", "command.status"].includes(envelope.messageKind)) { + if (["discovery.observed", "adapter.message", "command.status", "host.telemetry.observed"].includes(envelope.messageKind)) { scheduleTrackerEvent(connection, envelope); return; } @@ -297,7 +300,9 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { ? acceptDiscovery(connection, envelope) : envelope.messageKind === "adapter.message" ? acceptAdapterMessage(connection, envelope) - : acceptCommandStatus(connection, envelope)) + : envelope.messageKind === "command.status" + ? acceptCommandStatus(connection, envelope) + : acceptHostTelemetry(connection, envelope)) .catch((error) => failConnection(connection, error)) .finally(() => { if (connection.sessionChains.get(envelope.trackerSessionId) === work) { @@ -371,6 +376,23 @@ export function createDeviceGatewayCoreChannelClient(options = {}) { } } + async function acceptHostTelemetry(connection, envelope) { + try { + const snapshot = normalizeHostTelemetrySnapshot(envelope.payload?.snapshot); + const receipt = await config.recordHostTelemetry(snapshot, { + authenticatedEdgeRef: connection.registration.edgeRegistrationId, + }); + if (receipt?.status !== "recorded") { + throw new Error("device_gateway_core_host_telemetry_receipt_invalid"); + } + sendEventResult(connection, envelope, { status: "recorded" }); + totalEventsAccepted += 1; + } catch (error) { + sendEventRejection(connection, envelope, error); + totalEventsRejected += 1; + } + } + async function acceptCommandStatus(connection, envelope) { try { await config.recordCommandStatus(envelope.payload?.status); @@ -580,7 +602,13 @@ function normalizeConfig(options) { } const offerCommand = options.offerCommand ?? (async () => null); const recordCommandStatus = options.recordCommandStatus ?? (async () => undefined); - if (typeof offerCommand !== "function" || typeof recordCommandStatus !== "function") { + const recordHostTelemetry = options.recordHostTelemetry + ?? (async () => { throw new Error("device_host_telemetry_repository_unavailable"); }); + if ( + typeof offerCommand !== "function" + || typeof recordCommandStatus !== "function" + || typeof recordHostTelemetry !== "function" + ) { throw new TypeError("device_gateway_core_command_runtime_invalid"); } const registrationProvider = typeof options.registrationProvider === "function" @@ -627,6 +655,7 @@ function normalizeConfig(options) { commandTransport, offerCommand, recordCommandStatus, + recordHostTelemetry, keepaliveMs, deadPeerMs, connectTimeoutMs: normalizeInteger( diff --git a/services/device-control-core/src/edge-channel-supervisor.mjs b/services/device-control-core/src/edge-channel-supervisor.mjs index 38bc08e..4f8417c 100644 --- a/services/device-control-core/src/edge-channel-supervisor.mjs +++ b/services/device-control-core/src/edge-channel-supervisor.mjs @@ -116,6 +116,15 @@ export function createDeviceEdgeChannelSupervisor(options = {}) { message, { authenticatedEdgeRef: registration.edgeRegistrationId }, ), + recordHostTelemetry: (snapshot, context) => + typeof config.repository.recordInfrastructureHostTelemetry === "function" + ? config.repository.recordInfrastructureHostTelemetry({ + snapshot, + authenticatedEdgeRef: context.authenticatedEdgeRef, + }) + : Promise.reject(new Error( + "device_host_telemetry_repository_unavailable", + )), commandTransport: config.typedCommandRuntime ? "typed-service-ping-v1" : "disabled", diff --git a/services/device-control-core/src/host-telemetry-repository.mjs b/services/device-control-core/src/host-telemetry-repository.mjs new file mode 100644 index 0000000..98b0bac --- /dev/null +++ b/services/device-control-core/src/host-telemetry-repository.mjs @@ -0,0 +1,105 @@ +import { randomUUID } from "node:crypto"; + +import { + normalizeHostTelemetrySnapshot, +} from "../../../packages/infrastructure-telemetry-contract/src/index.mjs"; + +const FRESHNESS_SECONDS = 15; +const RETENTION_DAYS = 7; + +export async function recordInfrastructureHostTelemetry(client, input) { + const snapshot = normalizeHostTelemetrySnapshot(input?.snapshot); + const edgeId = entityId(input?.authenticatedEdgeRef, "edge"); + const relation = await client.query( + `select disi.id as service_instance_id, disi.host_id, + disi.project_id, disi.owner_scope_id, dih.host_key + from device_infrastructure_service_instances disi + join device_infrastructure_hosts dih on dih.id = disi.host_id + where disi.edge_id = $1 + and disi.lifecycle_state in ('active', 'degraded') + and dih.lifecycle_state = 'active' + order by disi.updated_at desc, disi.id + limit 2`, + [edgeId], + ); + if (relation.rows.length !== 1) { + throw domainError("device_host_telemetry_edge_host_binding_invalid", 409); + } + const target = relation.rows[0]; + if (target.host_key !== snapshot.hostKey) { + throw domainError("device_host_telemetry_host_key_mismatch", 409); + } + const observedAt = new Date(snapshot.observedAt); + const clockSkewMs = Math.abs(Date.now() - observedAt.valueOf()); + if (!Number.isFinite(observedAt.valueOf()) || clockSkewMs > 5 * 60 * 1000) { + throw domainError("device_host_telemetry_clock_skew_invalid", 409); + } + const id = randomUUID(); + const provenanceRef = `${input.authenticatedEdgeRef}:${snapshot.source.collectorRef}`; + const inserted = await client.query( + `with inserted as ( + insert into device_infrastructure_host_telemetry_samples ( + id, owner_scope_id, project_id, host_id, service_instance_id, edge_id, + observed_at, expires_at, schema_version, profile_ref, + agent_name, agent_version, collector_ref, provenance_ref, snapshot + ) values ( + $1, $2, $3, $4, $5, $6, + $7, $7::timestamptz + ($8 * interval '1 second'), $9, $10, + $11, $12, $13, $14, $15::jsonb + ) + on conflict (edge_id, observed_at) do nothing + returning id, received_at, expires_at, false as replayed + ) + select id, received_at, expires_at, replayed from inserted + union all + select id, received_at, expires_at, true as replayed + from device_infrastructure_host_telemetry_samples + where edge_id = $6 and observed_at = $7::timestamptz + and not exists (select 1 from inserted) + limit 1`, + [ + id, + target.owner_scope_id, + target.project_id, + target.host_id, + target.service_instance_id, + edgeId, + snapshot.observedAt, + FRESHNESS_SECONDS, + snapshot.schemaVersion, + snapshot.profile, + snapshot.source.agent, + snapshot.source.agentVersion, + snapshot.source.collectorRef, + provenanceRef, + JSON.stringify(snapshot), + ], + ); + await client.query( + `delete from device_infrastructure_host_telemetry_samples + where observed_at < now() - ($1 * interval '1 day')`, + [RETENTION_DAYS], + ); + const row = inserted.rows[0]; + return Object.freeze({ + status: "recorded", + replayed: row.replayed, + observationRef: `observation:${row.id}`, + hostRef: `host:${target.host_id}`, + observedAt: snapshot.observedAt, + receivedAt: new Date(row.received_at).toISOString(), + expiresAt: new Date(row.expires_at).toISOString(), + }); +} + +function entityId(value, kind) { + const match = new RegExp(`^${kind}:([0-9a-f-]{36})$`, "i").exec(String(value || "")); + if (!match) throw domainError(`device_${kind}_ref_invalid`, 400); + return match[1]; +} + +function domainError(code, statusCode) { + const error = new Error(code); + error.statusCode = statusCode; + return error; +} diff --git a/services/device-control-core/src/ontology-query-repository.mjs b/services/device-control-core/src/ontology-query-repository.mjs index 908f653..4f8b792 100644 --- a/services/device-control-core/src/ontology-query-repository.mjs +++ b/services/device-control-core/src/ontology-query-repository.mjs @@ -9,7 +9,7 @@ export async function getDeviceProjectOntologyProjection(client, actor, projectI { lock: false }, ); - const [assets, assetBindings, hosts, endpoints, deployments, services] = + const [assets, assetBindings, hosts, endpoints, deployments, services, telemetry] = await Promise.all([ client.query( `select * from device_assets @@ -82,8 +82,33 @@ export async function getDeviceProjectOntologyProjection(client, actor, projectI order by disi.display_name, disi.id`, [projectId], ), + client.query( + `select * from ( + select dihts.id, dihts.host_id, dihts.service_instance_id, + dihts.edge_id, dihts.observed_at, dihts.received_at, + dihts.expires_at, dihts.profile_ref, dihts.agent_name, + dihts.agent_version, dihts.collector_ref, + dihts.provenance_ref, dihts.snapshot, + dihts.ontology_entity_id, dihts.ontology_catalog_hash, + row_number() over ( + partition by dihts.host_id order by dihts.observed_at desc, dihts.id desc + ) as sample_rank + from device_infrastructure_host_telemetry_samples dihts + where dihts.project_id = $1 + ) ranked + where sample_rank <= 120 + order by host_id, observed_at desc, id desc`, + [projectId], + ), ]); + const telemetryByHost = telemetry.rows.reduce((result, row) => { + const rows = result.get(row.host_id) ?? []; + rows.push(row); + result.set(row.host_id, rows); + return result; + }, new Map()); + return { ontology: { catalogHash: "229c61c02a790906", @@ -91,7 +116,7 @@ export async function getDeviceProjectOntologyProjection(client, actor, projectI }, assets: assets.rows.map(assetView), assetBindings: assetBindings.rows.map(assetBindingView), - hosts: hosts.rows.map(hostView), + hosts: hosts.rows.map((row) => hostView(row, telemetryByHost.get(row.id) ?? [])), endpoints: endpoints.rows.map(endpointView), deployments: deployments.rows.map(deploymentView), serviceInstances: services.rows.map(serviceInstanceView), @@ -131,7 +156,7 @@ function assetBindingView(row) { }; } -function hostView(row) { +function hostView(row, telemetryRows) { return { hostRef: `host:${row.id}`, hostKey: row.host_key, @@ -141,10 +166,72 @@ function hostView(row) { managementCredentialConfigured: row.management_credential_configured === true, lifecycleState: row.lifecycle_state, health: healthView(row), + telemetry: hostTelemetryView(telemetryRows), ontology: ontologyView(row), }; } +function hostTelemetryView(rows) { + const latest = rows[0]; + if (!latest) { + return { + state: "unobserved", + freshness: "missing", + observedAt: null, + receivedAt: null, + expiresAt: null, + current: null, + history: [], + observation: null, + }; + } + const fresh = new Date(latest.expires_at).valueOf() > Date.now(); + return { + state: fresh ? "online" : "unobserved", + freshness: fresh ? "fresh" : "stale", + observedAt: toIso(latest.observed_at), + receivedAt: toIso(latest.received_at), + expiresAt: toIso(latest.expires_at), + current: latest.snapshot, + history: [...rows].reverse().map((row) => ({ + observedAt: toIso(row.observed_at), + cpuUsagePercent: numberOrNull(row.snapshot?.cpu?.usagePercent), + memoryUsedPercent: numberOrNull(row.snapshot?.memory?.usedPercent), + network: Array.isArray(row.snapshot?.network) + ? row.snapshot.network.map((item) => ({ + interface: item.interface ?? null, + bytesReceived: numberOrNull(item.bytesReceived), + bytesSent: numberOrNull(item.bytesSent), + })) + : [], + })), + observation: { + observationRef: `observation:${latest.id}`, + entityId: latest.ontology_entity_id, + catalogHash: latest.ontology_catalog_hash, + targetRef: `host:${latest.host_id}`, + serviceInstanceRef: `service-instance:${latest.service_instance_id}`, + edgeRef: `edge:${latest.edge_id}`, + profileRef: latest.profile_ref, + source: { + agent: latest.agent_name, + agentVersion: latest.agent_version, + collectorRef: latest.collector_ref, + provenanceRef: latest.provenance_ref, + }, + observedProperties: [ + "host.cpu.utilization", + "host.memory.utilization", + "host.swap.utilization", + "host.disk.utilization", + "host.network.counters", + "host.process.counts", + "host.systemd.unit-state", + ], + }, + }; +} + function endpointView(row) { return { endpointRef: `endpoint:${row.id}`, @@ -211,3 +298,8 @@ function ontologyView(row) { function toIso(value) { return value == null ? null : new Date(value).toISOString(); } + +function numberOrNull(value) { + const normalized = Number(value); + return Number.isFinite(normalized) ? normalized : null; +} diff --git a/services/device-control-core/src/postgres-repository.mjs b/services/device-control-core/src/postgres-repository.mjs index 3d51ff3..c711d27 100644 --- a/services/device-control-core/src/postgres-repository.mjs +++ b/services/device-control-core/src/postgres-repository.mjs @@ -55,6 +55,9 @@ import { planTypedServicePing, recordTypedCommandStatus, } from "./typed-command-repository.mjs"; +import { + recordInfrastructureHostTelemetry, +} from "./host-telemetry-repository.mjs"; const { Pool } = pg; const serviceRoot = resolve(dirname(fileURLToPath(import.meta.url)), ".."); @@ -75,6 +78,7 @@ const migrationFiles = [ "014_device_registry_profile_commands.sql", "015_device_integration_identity.sql", "016_device_asset_infrastructure_ontology.sql", + "017_infrastructure_host_telemetry.sql", ]; export class PostgresDeviceRepository { @@ -238,6 +242,12 @@ export class PostgresDeviceRepository { return this.#executeWrite((client) => recordTypedCommandStatus(client, input)); } + async recordInfrastructureHostTelemetry(input) { + return this.#executeWrite((client) => + recordInfrastructureHostTelemetry(client, input) + ); + } + async #executeWrite(operation) { const client = await this.pool.connect(); try { diff --git a/services/device-control-core/test/host-telemetry-repository.test.mjs b/services/device-control-core/test/host-telemetry-repository.test.mjs new file mode 100644 index 0000000..deeba39 --- /dev/null +++ b/services/device-control-core/test/host-telemetry-repository.test.mjs @@ -0,0 +1,102 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { readFile } from "node:fs/promises"; + +import { + recordInfrastructureHostTelemetry, +} from "../src/host-telemetry-repository.mjs"; + +const edgeId = "73da0c42-a641-4559-b8f7-23509b60bfe9"; +const hostId = "adf2a5b6-3c0b-4a39-998c-07dfb7818ad1"; +const serviceId = "01f14736-f5c2-4867-9cbc-2d268996a871"; +const projectId = "ad7b357c-c7ac-4bf8-a638-c7f956e9aa71"; +const ownerId = "78da71d5-f48f-4de0-8e47-729f6d644151"; + +test("records a normalized host observation only through the Edge-to-host graph", async () => { + const queries = []; + const now = new Date(); + const client = { + async query(sql, parameters) { + queries.push({ sql, parameters }); + if (queries.length === 1) return { rows: [{ + service_instance_id: serviceId, + host_id: hostId, + project_id: projectId, + owner_scope_id: ownerId, + host_key: "robot2b-b2-edge-vps", + }] }; + if (queries.length === 2) return { rows: [{ + id: "7ea94f66-6eed-4a27-8f04-e67060fa7e94", + received_at: now, + expires_at: new Date(now.valueOf() + 15_000), + replayed: false, + }] }; + return { rows: [] }; + }, + }; + + const result = await recordInfrastructureHostTelemetry(client, { + authenticatedEdgeRef: `edge:${edgeId}`, + snapshot: snapshot(now.toISOString()), + }); + + assert.equal(result.status, "recorded"); + assert.equal(result.replayed, false); + assert.equal(result.hostRef, `host:${hostId}`); + assert.match(queries[0].sql, /device_infrastructure_service_instances/); + assert.equal(queries[0].parameters[0], edgeId); + assert.match(queries[1].sql, /device_infrastructure_host_telemetry_samples/); + assert.equal(queries[1].parameters[5], edgeId); + assert.equal(JSON.stringify(queries).includes("password"), false); + assert.match(queries[2].sql, /delete from device_infrastructure_host_telemetry_samples/); +}); + +test("rejects telemetry whose host key disagrees with the canonical graph", async () => { + const client = { + async query() { + return { rows: [{ + service_instance_id: serviceId, + host_id: hostId, + project_id: projectId, + owner_scope_id: ownerId, + host_key: "canonical-host", + }] }; + }, + }; + await assert.rejects( + () => recordInfrastructureHostTelemetry(client, { + authenticatedEdgeRef: `edge:${edgeId}`, + snapshot: snapshot(new Date().toISOString()), + }), + /device_host_telemetry_host_key_mismatch/, + ); +}); + +test("host telemetry migration is additive, observation-backed and secret-free", async () => { + const sql = await readFile(new URL("../migrations/017_infrastructure_host_telemetry.sql", import.meta.url), "utf8"); + assert.match(sql, /create table if not exists device_infrastructure_host_telemetry_samples/); + assert.match(sql, /observation\.observation/); + assert.match(sql, /references device_infrastructure_hosts/); + assert.match(sql, /references device_infrastructure_service_instances/); + assert.doesNotMatch(sql, /insert into device_infrastructure_hosts/i); + assert.doesNotMatch(sql, /password|private.key|credential/i); +}); + +function snapshot(observedAt) { + return { + schemaVersion: "nodedc.infrastructure.host-telemetry.v1", + profile: "linux-host-telegraf-v1", + hostKey: "robot2b-b2-edge-vps", + observedAt, + source: { agent: "telegraf", agentVersion: "1.38.4", collectorRef: "service:nodedc-host-telemetry-agent" }, + hardware: { hostname: "koffyvngij", architecture: "x64", platform: "linux", kernelRelease: "6.8.0", cpuModel: "KVM CPU", logicalProcessors: 1 }, + cpu: { usagePercent: 20, load1: 0.2, load5: 0.1, load15: 0.05 }, + memory: { totalBytes: 1024, availableBytes: 700, freeBytes: null, usedBytes: 324, usedPercent: 31.6 }, + swap: { totalBytes: 0, availableBytes: null, freeBytes: 0, usedBytes: 0, usedPercent: 0 }, + system: { uptimeSeconds: 100, users: 1, processes: { total: 10, running: 1, sleeping: 9, blocked: 0, zombies: 0 } }, + disks: [], + network: [], + services: [], + }; +} diff --git a/services/device-edge-channel/src/runtime.mjs b/services/device-edge-channel/src/runtime.mjs index e5cd861..09bcf26 100644 --- a/services/device-edge-channel/src/runtime.mjs +++ b/services/device-edge-channel/src/runtime.mjs @@ -14,6 +14,9 @@ import { normalizeAdapterMessage, normalizeDiscoverySignal, } from "../../../packages/device-protocol-contract/src/index.mjs"; +import { + normalizeHostTelemetrySnapshot, +} from "../../../packages/infrastructure-telemetry-contract/src/index.mjs"; const CHANNEL_TRACKER_SESSION_ID = "channel:control"; const CHANNEL_PROFILE_REF = "channel.control.v1"; @@ -195,6 +198,20 @@ export function createDeviceEdgeChannelServer(options = {}) { } return Object.freeze({ status: "recorded" }); }, + async submitHostTelemetry(snapshot) { + const normalized = normalizeHostTelemetrySnapshot(snapshot); + const result = await submitEvent("host.telemetry.observed", { + snapshot: normalized, + }, { + trackerSessionId: `host-telemetry:${normalized.hostKey}`, + adapterProfileRef: normalized.profile, + eventAt: normalized.observedAt, + }); + if (result?.status !== "recorded") { + throw new Error("device_edge_channel_host_telemetry_invalid"); + } + return Object.freeze({ status: "recorded" }); + }, status() { return Object.freeze({ listening: started, diff --git a/services/device-edge-channel/test/channel-integration.test.mjs b/services/device-edge-channel/test/channel-integration.test.mjs index c15b55e..1e5b3ce 100644 --- a/services/device-edge-channel/test/channel-integration.test.mjs +++ b/services/device-edge-channel/test/channel-integration.test.mjs @@ -68,6 +68,33 @@ test("accepts synthetic discovery and durable message results over Core-initiate } }); +test("delivers a normalized host observation over the existing pinned mTLS channel", async () => { + const recorded = []; + const pair = await startPair({ + recordHostTelemetry: async (snapshot, context) => { + recorded.push({ snapshot, context }); + return { status: "recorded" }; + }, + }); + + try { + const observation = hostTelemetrySnapshot(); + assert.deepEqual(await pair.edge.submitHostTelemetry(observation), { + status: "recorded", + }); + assert.equal(recorded.length, 1); + assert.equal(recorded[0].snapshot.hostKey, "robot2b-b2-edge-vps"); + assert.equal( + recorded[0].context.authenticatedEdgeRef, + "edge:pilot-1", + ); + assert.equal(pair.edge.status().eventsAccepted, 1); + assert.equal(pair.core.status().eventsAccepted, 1); + } finally { + await stopPair(pair); + } +}); + test("recovers the claimed device from telemetry after a Core restart and completes a typed command", async () => { const commandRef = "command:11111111-1111-4111-8111-111111111111"; const deviceRef = "device:22222222-2222-4222-8222-222222222222"; @@ -547,9 +574,61 @@ function createCoreClient(options) { commandTransport: options.commandTransport, offerCommand: options.offerCommand, recordCommandStatus: options.recordCommandStatus, + recordHostTelemetry: options.recordHostTelemetry, }); } +function hostTelemetrySnapshot() { + return { + schemaVersion: "nodedc.infrastructure.host-telemetry.v1", + profile: "linux-host-telegraf-v1", + hostKey: "robot2b-b2-edge-vps", + observedAt: new Date().toISOString(), + source: { + agent: "telegraf", + agentVersion: "1.38.4", + collectorRef: "service:nodedc-host-telemetry-agent", + }, + hardware: { + hostname: "koffyvngij", + architecture: "x64", + platform: "linux", + kernelRelease: "6.8.0", + cpuModel: "KVM CPU", + logicalProcessors: 1, + }, + cpu: { usagePercent: 20, load1: 0.2, load5: 0.1, load15: 0.05 }, + memory: { + totalBytes: 1024, + availableBytes: 700, + freeBytes: null, + usedBytes: 324, + usedPercent: 31.6, + }, + swap: { + totalBytes: 0, + availableBytes: null, + freeBytes: 0, + usedBytes: 0, + usedPercent: 0, + }, + system: { + uptimeSeconds: 100, + users: 1, + processes: { + total: 10, + running: 1, + sleeping: 9, + blocked: 0, + zombies: 0, + }, + }, + disks: [], + network: [], + services: [], + }; +} + function productionDiscoveryObserver() { return createDeviceGatewayIngest({ identifierPepper: "test-only-device-edge-channel-identifier-pepper", diff --git a/vps/config/nodedc-host-telemetry-telegraf.conf b/vps/config/nodedc-host-telemetry-telegraf.conf new file mode 100644 index 0000000..4190b47 --- /dev/null +++ b/vps/config/nodedc-host-telemetry-telegraf.conf @@ -0,0 +1,52 @@ +[agent] + interval = "2s" + round_interval = true + metric_batch_size = 1000 + metric_buffer_limit = 10000 + collection_jitter = "0s" + flush_interval = "2s" + flush_jitter = "0s" + precision = "1s" + debug = false + quiet = false + hostname = "koffyvngij" + omit_hostname = false + +[global_tags] + nodedc_host_key = "robot2b-b2-edge-vps" + nodedc_profile = "linux-host-telegraf-v1" + +[[inputs.cpu]] + percpu = false + totalcpu = true + collect_cpu_time = false + report_active = true + +[[inputs.mem]] + +[[inputs.swap]] + +[[inputs.system]] + +[[inputs.processes]] + +[[inputs.disk]] + ignore_fs = ["tmpfs", "devtmpfs", "devfs", "iso9660", "overlay", "aufs", "squashfs", "nsfs"] + +[[inputs.diskio]] + +[[inputs.net]] + interfaces = ["*"] + ignore_protocol_stats = true + +[[inputs.systemd_units]] + pattern = "nodedc-*.service ssh.service systemd-networkd.service" + details = true + +[[outputs.http]] + url = "http://127.0.0.1:18223/internal/v1/host-telemetry" + method = "POST" + timeout = "10s" + data_format = "json" + use_batch_format = true + content_encoding = "identity" diff --git a/vps/edge-process/device-edge-runtime.mjs b/vps/edge-process/device-edge-runtime.mjs index 4376e3b..1f5f951 100644 --- a/vps/edge-process/device-edge-runtime.mjs +++ b/vps/edge-process/device-edge-runtime.mjs @@ -13,6 +13,7 @@ import { import { createDeviceGatewayRuntime, } from "../../services/device-gateway/src/runtime.mjs"; +import { createHostTelemetryCollector } from "./host-telemetry-runtime.mjs"; if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { await main(); @@ -51,16 +52,26 @@ export async function main(environment = process.env) { onMessage: (message) => channel.submitAdapterMessage(message), onCommandStatus: (status) => channel.submitCommandStatus(status), }); - const health = createCombinedHealthServer(channel, gateway, base.health); + const telemetry = createHostTelemetryCollector({ + submit: (snapshot) => channel.submitHostTelemetry(snapshot), + hostKey: environment.NODEDC_INFRASTRUCTURE_HOST_KEY + ?? "robot2b-b2-edge-vps", + host: environment.NODEDC_HOST_TELEMETRY_HOST ?? "127.0.0.1", + port: environment.NODEDC_HOST_TELEMETRY_PORT ?? 18223, + agentVersion: environment.NODEDC_HOST_TELEMETRY_AGENT_VERSION ?? "1.38.4", + }); + const health = createCombinedHealthServer(channel, gateway, telemetry, base.health); let stopping = false; try { await channel.start(); await gateway.start(); + await telemetry.start(); await listen(health, base.health.port, base.health.host); } catch (error) { await Promise.allSettled([ gateway.stop(), + telemetry.stop(), channel.stop(), closeServer(health), ]); @@ -72,6 +83,7 @@ export async function main(environment = process.env) { channel: `${base.channel.host}:${base.channel.port}`, health: `${base.health.host}:${base.health.port}`, trackerIngress: `${tracker.tcpHost}:${tracker.tcpPort}`, + hostTelemetry: `${telemetry.status().host}:${telemetry.status().port}`, adapterProfile: tracker.protocolProfileRef, edgeRegistrationId: base.channel.edgeRegistrationId, channelGeneration: base.channel.channelGeneration, @@ -87,6 +99,7 @@ export async function main(environment = process.env) { stopping = true; await Promise.allSettled([ gateway.stop(), + telemetry.stop(), channel.stop(), closeServer(health), ]); @@ -161,7 +174,7 @@ export function normalizeTrackerIngressConfiguration(environment = {}, edgeRef) }); } -function createCombinedHealthServer(channel, gateway, healthConfig) { +function createCombinedHealthServer(channel, gateway, telemetry, healthConfig) { return createServer((request, response) => { response.setHeader("Content-Type", "application/json; charset=utf-8"); response.setHeader("Cache-Control", "no-store"); @@ -179,6 +192,7 @@ function createCombinedHealthServer(channel, gateway, healthConfig) { ...channel.status(), trackerIngress: "telemetry-ingest", tracker: gateway.status(), + hostTelemetry: telemetry.status(), commandTransport: "typed-service-ping-v1", })); }); diff --git a/vps/edge-process/host-telemetry-runtime.mjs b/vps/edge-process/host-telemetry-runtime.mjs new file mode 100644 index 0000000..59416b5 --- /dev/null +++ b/vps/edge-process/host-telemetry-runtime.mjs @@ -0,0 +1,161 @@ +import { createServer } from "node:http"; +import { arch, cpus, hostname, platform, release } from "node:os"; + +import { + telegrafBatchToHostTelemetry, +} from "../../packages/infrastructure-telemetry-contract/src/index.mjs"; + +const MAX_BODY_BYTES = 512 * 1024; + +export function createHostTelemetryCollector(options = {}) { + const config = normalizeConfig(options); + let accepted = 0; + let rejected = 0; + let lastObservedAt = null; + let lastAcceptedAt = null; + let lastErrorCode = null; + let pending = false; + + const server = createServer(async (request, response) => { + response.setHeader("Content-Type", "application/json; charset=utf-8"); + response.setHeader("Cache-Control", "no-store"); + response.setHeader("X-Content-Type-Options", "nosniff"); + if (request.method !== "POST" || request.url !== "/internal/v1/host-telemetry") { + response.statusCode = 404; + response.end(JSON.stringify({ ok: false, error: "not_found" })); + return; + } + if (pending) { + rejected += 1; + response.statusCode = 429; + response.end(JSON.stringify({ ok: false, error: "host_telemetry_busy" })); + return; + } + pending = true; + try { + const body = await readJsonBody(request, MAX_BODY_BYTES); + const processors = cpus(); + const snapshot = telegrafBatchToHostTelemetry(body, { + hostKey: config.hostKey, + hostname: hostname(), + architecture: arch(), + platform: platform(), + kernelRelease: release(), + cpuModel: processors[0]?.model ?? null, + logicalProcessors: processors.length || null, + agentVersion: config.agentVersion, + }); + lastObservedAt = snapshot.observedAt; + await config.submit(snapshot); + accepted += 1; + lastAcceptedAt = new Date().toISOString(); + lastErrorCode = null; + response.statusCode = 202; + response.end(JSON.stringify({ ok: true, accepted: true })); + } catch (error) { + rejected += 1; + lastErrorCode = safeErrorCode(error); + response.statusCode = lastErrorCode.includes("too_large") ? 413 : 503; + response.end(JSON.stringify({ ok: false, error: lastErrorCode })); + } finally { + pending = false; + } + }); + + return Object.freeze({ + async start() { + await listen(server, config.port, config.host); + return server.address(); + }, + async stop() { + await closeServer(server); + }, + status() { + return Object.freeze({ + listening: server.listening, + host: config.host, + port: config.port, + profile: "linux-host-telegraf-v1", + accepted, + rejected, + pending, + lastObservedAt, + lastAcceptedAt, + lastErrorCode, + }); + }, + }); +} + +function normalizeConfig(options) { + if (typeof options.submit !== "function") { + throw new TypeError("host_telemetry_submit_required"); + } + return Object.freeze({ + submit: options.submit, + hostKey: normalizeRef(options.hostKey, "host_telemetry_host_key_invalid"), + agentVersion: String(options.agentVersion ?? "1.38.4"), + host: normalizeLoopbackHost(options.host ?? "127.0.0.1"), + port: normalizePort(options.port ?? 18223), + }); +} + +async function readJsonBody(request, maximumBytes) { + const chunks = []; + let total = 0; + for await (const chunk of request) { + total += chunk.length; + if (total > maximumBytes) throw new Error("host_telemetry_body_too_large"); + chunks.push(chunk); + } + if (total < 2) throw new Error("host_telemetry_body_invalid"); + try { + return JSON.parse(Buffer.concat(chunks).toString("utf8")); + } catch { + throw new Error("host_telemetry_body_invalid"); + } +} + +function normalizeRef(value, errorCode) { + if (typeof value !== "string" || !/^[a-z][a-z0-9-]{1,62}$/.test(value)) { + throw new TypeError(errorCode); + } + return value; +} + +function normalizeLoopbackHost(value) { + if (!["127.0.0.1", "::1"].includes(value)) { + throw new TypeError("host_telemetry_host_invalid"); + } + return value; +} + +function normalizePort(value) { + const normalized = Number(value); + if (!Number.isSafeInteger(normalized) || normalized < 0 || normalized > 65_535) { + throw new TypeError("host_telemetry_port_invalid"); + } + return normalized; +} + +function safeErrorCode(error) { + const value = String(error?.message || "host_telemetry_internal_error"); + return /^[a-z0-9_.:-]{3,160}$/.test(value) + ? value + : "host_telemetry_internal_error"; +} + +function listen(server, port, host) { + return new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(port, host, () => { + server.off("error", reject); + resolve(); + }); + }); +} + +function closeServer(server) { + if (!server.listening) return Promise.resolve(); + return new Promise((resolve) => server.close(() => resolve())); +} diff --git a/vps/edge-process/host-telemetry-runtime.test.mjs b/vps/edge-process/host-telemetry-runtime.test.mjs new file mode 100644 index 0000000..cf8a4d0 --- /dev/null +++ b/vps/edge-process/host-telemetry-runtime.test.mjs @@ -0,0 +1,37 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { createHostTelemetryCollector } from "./host-telemetry-runtime.mjs"; + +test("accepts Telegraf only on the bounded local collector", async () => { + let submitted = null; + const collector = createHostTelemetryCollector({ + hostKey: "robot2b-b2-edge-vps", + host: "127.0.0.1", + port: 0, + submit: async (value) => { submitted = value; }, + }); + const address = await collector.start(); + try { + const response = await fetch(`http://127.0.0.1:${address.port}/internal/v1/host-telemetry`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ metrics: [ + { name: "cpu", tags: { cpu: "cpu-total", host: "edge" }, fields: { usage_active: 9 }, timestamp: Math.floor(Date.now() / 1000) }, + ] }), + }); + assert.equal(response.status, 202); + assert.equal(submitted.hostKey, "robot2b-b2-edge-vps"); + assert.equal(collector.status().accepted, 1); + } finally { + await collector.stop(); + } +}); + +test("rejects non-loopback collector binds", () => { + assert.throws(() => createHostTelemetryCollector({ + hostKey: "host-01", + host: "0.0.0.0", + submit: async () => {}, + }), /host_telemetry_host_invalid/); +}); diff --git a/vps/systemd/nodedc-device-edge-runtime.service b/vps/systemd/nodedc-device-edge-runtime.service index 27d284c..2e51976 100644 --- a/vps/systemd/nodedc-device-edge-runtime.service +++ b/vps/systemd/nodedc-device-edge-runtime.service @@ -28,6 +28,10 @@ Environment=DEVICE_GATEWAY_MAX_SESSIONS_PER_ADDRESS=16 Environment=DEVICE_GATEWAY_MAX_CONNECTIONS_PER_MINUTE_PER_ADDRESS=60 Environment=DEVICE_GATEWAY_MAX_TRACKED_SOURCE_ADDRESSES=2048 Environment=DEVICE_GATEWAY_SESSION_TIMEOUT_MS=10000 +Environment=NODEDC_INFRASTRUCTURE_HOST_KEY=robot2b-b2-edge-vps +Environment=NODEDC_HOST_TELEMETRY_HOST=127.0.0.1 +Environment=NODEDC_HOST_TELEMETRY_PORT=18223 +Environment=NODEDC_HOST_TELEMETRY_AGENT_VERSION=1.38.4 Restart=always RestartSec=2 TimeoutStartSec=20 diff --git a/vps/systemd/nodedc-host-telemetry-agent.service b/vps/systemd/nodedc-host-telemetry-agent.service new file mode 100644 index 0000000..451a8f8 --- /dev/null +++ b/vps/systemd/nodedc-host-telemetry-agent.service @@ -0,0 +1,47 @@ +[Unit] +Description=NODE.DC infrastructure host telemetry agent +Documentation=https://docs.influxdata.com/telegraf/v1/ +After=network-online.target nodedc-device-edge-channel.service +Wants=network-online.target +Requires=nodedc-device-edge-channel.service + +[Service] +Type=notify +NotifyAccess=all +User=nodedc-telemetry +Group=nodedc-telemetry +ExecStart=/opt/nodedc-b2-vps/runtime/telegraf/usr/bin/telegraf --config /opt/nodedc-b2-vps/vps/config/nodedc-host-telemetry-telegraf.conf +Restart=always +RestartSec=3 +TimeoutStartSec=30 +TimeoutStopSec=15 +NoNewPrivileges=yes +PrivateTmp=yes +PrivateDevices=yes +ProtectSystem=strict +ProtectHome=yes +ProtectKernelTunables=yes +ProtectKernelModules=yes +ProtectKernelLogs=yes +ProtectControlGroups=yes +ProtectClock=yes +ProtectHostname=yes +RestrictSUIDSGID=yes +RestrictRealtime=yes +LockPersonality=yes +MemoryDenyWriteExecute=no +SystemCallArchitectures=native +RestrictAddressFamilies=AF_UNIX AF_INET +IPAddressDeny=any +IPAddressAllow=localhost +CapabilityBoundingSet= +AmbientCapabilities= +UMask=0077 +MemoryMax=96M +MemorySwapMax=0 +CPUQuota=15% +TasksMax=64 +LimitNOFILE=512 + +[Install] +WantedBy=multi-user.target