Merge branch 'feature/monitoring-telemetry' into 'main'

feat: telemetry reporter for monitoring instances

See merge request postgres-ai/postgresai!251

Author: Sarumyan

.gitlab-ci.yml

@@ -331,6 +331,24 @@ cli:node:tests:
     - if: '$CI_PIPELINE_SOURCE == "merge_request_event"'
     - if: '$CI_COMMIT_BRANCH == "main"'
 
+telemetry:tests:
+  stage: test
+  image: oven/bun:1
+  variables:
+    GIT_STRATEGY: fetch
+  before_script:
+    - cd "$CI_PROJECT_DIR/telemetry" && bun install --frozen-lockfile
+  script:
+    - cd "$CI_PROJECT_DIR/telemetry" && bun audit --audit-level=high
+    - cd "$CI_PROJECT_DIR/telemetry" && bun test && bun run typecheck
+  rules:
+    - if: '$CI_PIPELINE_SOURCE == "merge_request_event"'
+      changes:
+        - telemetry/**/*
+    - if: '$CI_COMMIT_BRANCH == "main"'
+      changes:
+        - telemetry/**/*
+
 # Validate helm chart on merge requests and main branch
 validate-helm-chart:
   stage: validate

telemetry/Dockerfile

@@ -0,0 +1,25 @@
+FROM oven/bun:1 AS deps
+WORKDIR /app
+COPY package.json bun.lock ./
+RUN bun install --frozen-lockfile
+
+FROM oven/bun:1
+WORKDIR /app
+COPY --from=deps /app/node_modules ./node_modules
+COPY package.json tsconfig.json ./
+COPY bin ./bin
+COPY lib ./lib
+
+# /proc and /var/lib/docker (or /) need to be visible from inside the container.
+# Compose / k8s manifest must mount:
+#   - /proc            -> /host/proc    (readonly)
+#   - / (or data vol)  -> /host/disk    (readonly)
+#   - /var/run/docker.sock -> /var/run/docker.sock  (so `docker ps` works)
+#
+# Mounting the docker socket grants root-equivalent access to the host. Where
+# possible, prefer a docker-socket proxy restricted to GET /containers/json.
+# See telemetry/README.md for the deployment threat model.
+ENV PGAI_TELEMETRY_MEMINFO_PATH=/host/proc/meminfo \
+    PGAI_TELEMETRY_DISK_PATH=/host/disk
+
+CMD ["bun", "run", "./bin/telemetry.ts"]

telemetry/README.md

@@ -0,0 +1,120 @@
+# @postgresai/telemetry
+
+Telemetry reporter for PostgresAI monitoring instances. A small TS+Bun
+service that runs on each monitoring host and posts an hourly
+mini-healthcheck to the platform.
+
+## What it collects
+
+Each tick gathers four signals and POSTs them to the platform:
+
+| Signal | Source |
+|---|---|
+| OOM events in the lookback window | `journalctl -k --since "<lookback>"` |
+| Faulty containers (exited / dead / restarting / unhealthy) | `docker ps -a --format '{{json .}}'` |
+| Free RAM | `MemAvailable` from `/proc/meminfo` (falls back to `MemFree`) |
+| Free disk | `fs.statfs` on the configured mount |
+
+The companion platform-side hypertable, RPC, alert evaluator, and
+dispatcher live in `postgres-ai/platform-all!365`.
+
+## Configuration
+
+| Var | Required | Default |
+|---|---|---|
+| `PGAI_PLATFORM_API_URL` | yes | — |
+| `PGAI_API_TOKEN` | yes | — |
+| `PGAI_MONITORING_INSTANCE_ID` | yes | — |
+| `PGAI_TELEMETRY_DISK_PATH` | no | `/` |
+| `PGAI_TELEMETRY_MEMINFO_PATH` | no | `/proc/meminfo` |
+| `PGAI_TELEMETRY_OOM_LOOKBACK` | no | `24 hours ago` |
+| `PGAI_TELEMETRY_INTERVAL_SEC` | no | `3600` (min `60`) |
+
+`PGAI_API_TOKEN` is the existing PostgresAI checkup double-base64 token.
+
+## Build and run locally
+
+```sh
+cd telemetry
+bun install --frozen-lockfile
+bun test
+bun run typecheck
+bun run start          # actually starts reporting
+```
+
+## Run in a container
+
+```sh
+docker build -t postgresai-telemetry telemetry
+docker run --rm \
+  -e PGAI_PLATFORM_API_URL=https://postgres.ai/api/v1 \
+  -e PGAI_API_TOKEN=... \
+  -e PGAI_MONITORING_INSTANCE_ID=... \
+  --read-only \
+  -v /proc:/host/proc:ro \
+  -v /:/host/disk:ro \
+  -v /var/run/docker.sock:/var/run/docker.sock \
+  postgresai-telemetry
+```
+
+## Deployment requirements
+
+The agent must read host kernel logs, host memory, the host filesystem,
+and ask the local Docker daemon for its container list. Mount these:
+
+| Host path | Container path | Mode |
+|---|---|---|
+| `/proc` | `/host/proc` | read-only |
+| `/` (or data volume) | `/host/disk` | read-only |
+| `/var/run/docker.sock` | `/var/run/docker.sock` | read-write |
+
+## Threat model
+
+Mounting `/var/run/docker.sock` is **root-equivalent on the host**.
+Anyone who execs into this container — or compromises any of its
+dependencies — can launch privileged containers and take over the
+monitoring host.
+
+Recommended mitigations:
+
+- Prefer a docker-socket proxy (e.g.
+  [`tecnativa/docker-socket-proxy`](https://github.com/Tecnativa/docker-socket-proxy))
+  restricted to `CONTAINERS=1` so only `GET /containers/json` is exposed.
+- Drop all Linux capabilities the agent doesn't need
+  (`--cap-drop ALL --cap-add ...`).
+- Run as a non-root UID inside the container; the `oven/bun` base image
+  ships a `bun` user.
+- `PGAI_TELEMETRY_MEMINFO_PATH` and `PGAI_TELEMETRY_DISK_PATH` are
+  security-sensitive: an actor who can flip them can turn the heartbeat
+  into an arbitrary-file-read primitive. Keep them under config-management
+  control.
+
+## API contract
+
+`POST /rpc/monitoring_instance_telemetry_report` with a JSON body:
+
+```json
+{
+  "api_token": "<double-base64 token>",
+  "instance_id": "<uuid>",
+  "oom_count_24h": 0,
+  "faulty_containers": ["cadvisor"],
+  "free_ram_bytes": 8589934592,
+  "free_disk_bytes": 100000000000,
+  "metadata": { "collected_at": "2026-04-28T09:00:00.000Z" }
+}
+```
+
+All `*_bytes` fields are byte counts. `metadata` is an open-ended
+JSON object that today carries `collected_at` (ISO 8601 UTC).
+
+## Operational notes
+
+- **Startup tick**: the agent reports once on startup, then on each
+  `PGAI_TELEMETRY_INTERVAL_SEC` boundary.
+- **Graceful shutdown**: SIGTERM / SIGINT cancel the in-flight sleep
+  immediately. Shutdown latency is bounded by the current tick (not the
+  interval).
+- **Per-collector failure isolation**: each of the four collectors logs
+  a warning on failure and reports a safe default (`0` / `[]`) so a
+  single broken signal doesn't silence the heartbeat.

telemetry/bin/telemetry.ts

@@ -0,0 +1,105 @@
+#!/usr/bin/env bun
+/**
+ * Entry point for the PostgresAI monitoring instance telemetry agent.
+ *
+ * Runs forever, reporting once on startup and then every intervalSec.
+ * SIGTERM / SIGINT cancel the in-flight sleep and exit cleanly, so
+ * shutdown latency is bounded by the current tick (not by intervalSec).
+ */
+
+import { loadConfigFromEnv, ConfigError } from "../lib/config";
+import { collectSnapshot } from "../lib/collect";
+import { postReport } from "../lib/reporter";
+import type { ReporterConfig } from "../lib/types";
+
+const RESPONSE_LOG_CAP = 512;
+
+export async function tick(config: ReporterConfig): Promise<void> {
+  const snapshot = await collectSnapshot(config);
+  const result = await postReport(config, snapshot);
+  if (!result.ok) {
+    const raw = JSON.stringify(result.body ?? null);
+    const capped = raw.length > RESPONSE_LOG_CAP ? `${raw.slice(0, RESPONSE_LOG_CAP)}…` : raw;
+    const safe = capped.split(config.apiToken).join("[REDACTED]");
+    console.error(
+      `[telemetry] report failed status=${result.status} err=${result.error ?? ""} body=${safe}`
+    );
+    return;
+  }
+  console.log(
+    `[telemetry] reported snapshot oom=${snapshot.oomCount24h} faulty=${snapshot.faultyContainers.length} ram=${snapshot.freeRamBytes} disk=${snapshot.freeDiskBytes}`
+  );
+}
+
+export interface RunLoopDeps {
+  tickFn?: (config: ReporterConfig) => Promise<void>;
+  sleep?: (ms: number, signal: AbortSignal) => Promise<void>;
+  onSignal?: (handler: () => void) => void;
+}
+
+const defaultSleep = (ms: number, signal: AbortSignal): Promise<void> =>
+  new Promise<void>((resolve) => {
+    if (signal.aborted) return resolve();
+    const t = setTimeout(resolve, ms);
+    signal.addEventListener(
+      "abort",
+      () => {
+        clearTimeout(t);
+        resolve();
+      },
+      { once: true }
+    );
+  });
+
+const defaultOnSignal = (handler: () => void): void => {
+  process.once("SIGTERM", handler);
+  process.once("SIGINT", handler);
+};
+
+export async function runLoop(config: ReporterConfig, deps: RunLoopDeps = {}): Promise<void> {
+  const tickFn = deps.tickFn ?? tick;
+  const sleep = deps.sleep ?? defaultSleep;
+  const onSignal = deps.onSignal ?? defaultOnSignal;
+
+  const ac = new AbortController();
+  let stopped = false;
+  const stop = () => {
+    stopped = true;
+    ac.abort();
+  };
+  onSignal(stop);
+
+  await tickFn(config).catch((err) => {
+    console.error(`[telemetry] tick error: ${err instanceof Error ? err.message : String(err)}`);
+  });
+
+  while (!stopped) {
+    await sleep(config.intervalSec * 1000, ac.signal);
+    if (stopped) break;
+    await tickFn(config).catch((err) => {
+      console.error(`[telemetry] tick error: ${err instanceof Error ? err.message : String(err)}`);
+    });
+  }
+}
+
+async function main(): Promise<void> {
+  let config: ReporterConfig;
+  try {
+    config = loadConfigFromEnv();
+  } catch (err) {
+    if (err instanceof ConfigError) {
+      console.error(`[telemetry] config error: ${err.message}`);
+      process.exit(2);
+    }
+    throw err;
+  }
+
+  await runLoop(config);
+}
+
+if (import.meta.main) {
+  main().catch((err) => {
+    console.error(`[telemetry] fatal: ${err instanceof Error ? err.stack ?? err.message : String(err)}`);
+    process.exit(1);
+  });
+}

telemetry/bun.lock

@@ -0,0 +1,45 @@
+/**
+ * High-level "run all collectors" helper. The entry script calls this on
+ * each tick. Splitting this out keeps bin/telemetry.ts tiny and lets us
+ * unit-test the orchestration with stubbed collectors.
+ */
+
+import { collectOomCount } from "./collectors/oom";
+import { collectFaultyContainers } from "./collectors/containers";
+import { collectFreeRamBytes } from "./collectors/memory";
+import { collectFreeDiskBytes } from "./collectors/disk";
+import type { ReporterConfig, TelemetrySnapshot } from "./types";
+
+export interface CollectorOverrides {
+  oom?: () => Promise<number>;
+  containers?: () => Promise<string[]>;
+  memory?: () => Promise<number>;
+  disk?: () => Promise<number>;
+}
+
+export async function collectSnapshot(
+  config: ReporterConfig,
+  overrides: CollectorOverrides = {}
+): Promise<TelemetrySnapshot> {
+  const oomFn = overrides.oom ?? (() => collectOomCount({ lookback: config.oomLookback }));
+  const containersFn = overrides.containers ?? (() => collectFaultyContainers());
+  const memoryFn = overrides.memory ?? (() => collectFreeRamBytes({ meminfoPath: config.meminfoPath }));
+  const diskFn = overrides.disk ?? (() => collectFreeDiskBytes({ path: config.diskPath }));
+
+  const [oomCount24h, faultyContainers, freeRamBytes, freeDiskBytes] = await Promise.all([
+    oomFn(),
+    containersFn(),
+    memoryFn(),
+    diskFn(),
+  ]);
+
+  return {
+    oomCount24h,
+    faultyContainers,
+    freeRamBytes,
+    freeDiskBytes,
+    metadata: {
+      collected_at: new Date().toISOString(),
+    },
+  };
+}