feat(telemetry): add canonical VPS host monitoring

2026-08-22 17:29:30 +03:00
parent a3e7a015de
commit 23ea96fbb2
33 changed files with 2541 additions and 51 deletions
@@ -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
@@ -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;
@@ -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(
@@ -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",
@@ -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;
+}
@@ -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;
+}
@@ -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 {
@@ -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: [],
+  };
+}