openvibers/openvibe.live Commits

Live: request telemetry in its own table, 14-day retention (#137)

OP OpenVibers committed 29 days ago · 9427d15eed · 4 files, +92 −2

Plan T1 step 10 (universal telemetry): this service now records its request telemetry using openvibe-sdk 0.46's process-wide collector, the same shape Network uses.

- **Migration** `0006_telemetry.sql` (expand; merges after Live#55, whose drop is 0005): the `telemetry_samples` table from `telemetrySchema()`.
- **`server/index.js`** mounts `telemetryMiddleware` before any route. It does nothing until `init()`, so tests and tools that build the app are unaffected.
- **`start()`** once the database is ready (a restore drill, `LIVE_DRILL`, returns before it and starts neither, as `test/drill-mode.test.js` requires):
  - `telemetry.init({ service, sink: createPgTelemetrySink(...) })` aggregates requests per route template, method and status class. Every minute (the singleton's own timer, `intervalMs: 60_000`) it writes one `http.request` sample per key plus the active-request, p95 and event-loop-lag gauges. `/shared`, health, readiness and `/metrics` produce no samples.
  - `startTelemetryRetention` deletes rows older than 14 days, a minute after start and then daily. At this interval and retention a service holds about 55k rows (about 15 MB with indexes); at 15 s and 30 days it would be about 125 MB, roughly 4.4 GB across the estate.
  - On stop, the retention stops and the collector flushes once before the database closes.
- **`test/telemetry.test.js`**:
  - checks that wiring in the source;
  - checks that five hits on one route flush as one sample (count 5) and that `/metrics` and `/shared` produce none;
  - checks that the migration's table works with the retention (only rows from the last 14 days are kept).

Changes

ddiff --git a/migrations/0006_telemetry.sql b/migrations/0006_telemetry.sql
nnew file mode 100644
iindex 00000000..1db3d21d
---- /dev/null
++++ b/migrations/0006_telemetry.sql
@@@ -0,0 +1,28 @@
++-- phase: expand
++-- Request telemetry (openvibe-sdk/telemetry telemetrySchema()): one row per aggregated sample, pruned after 14 days (server/index.js).
++CREATE TABLE IF NOT EXISTS telemetry_samples (
++ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
++ at timestamptz NOT NULL,
++ service text NOT NULL,
++ project text,
++ subject text,
++ resource text,
++ provider text,
++ node text,
++ cell text,
++ region text,
++ operation text NOT NULL,
++ latency_ms double precision,
++ queue_delay_ms double precision,
++ ttfb_ms double precision,
++ throughput_per_second double precision,
++ bytes bigint,
++ status text,
++ cache_status text,
++ cost_estimate double precision,
++ route_epoch bigint,
++ trace_id text,
++ extra jsonb NOT NULL DEFAULT '{}'::jsonb
++);
++CREATE INDEX IF NOT EXISTS telemetry_samples_at_idx ON telemetry_samples (at);
++CREATE INDEX IF NOT EXISTS telemetry_samples_service_idx ON telemetry_samples (service, operation, at);
ddiff --git a/server/index.js b/server/index.js
iindex 510cc709..822f69bb 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -146,7 +146,9 @@ const server = http.createServer(app);
// SIGTERM/SIGINT (non-drill) run this once: stop steps, beforeDrain (sockets and child processes stop while the
// HTTP server still takes connections), a 4 s drain, then the close steps; past 5 s the process exits 1.
// signals: false: shutdown() below installs the handlers, because a drill stops its own way.
--const { gracefulStop } = require('openvibe-sdk/service');
++const { gracefulStop, telemetry } = require('openvibe-sdk/service');
++const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');
++let telemetryRetention = null; // request telemetry's 14-day retention, started in start() (never under LIVE_DRILL)
const stopper = gracefulStop({
name: 'Live',
server,
@@@ -201,6 +203,7 @@ const stopper = gracefulStop({
},
close: [
async () => { try { await analyticsStore.close(); } catch (err) { console.error('[Analytics] Shutdown:', err); } },
++ async () => { if (telemetryRetention) await telemetryRetention.stop(); await telemetry.stop(); }, // request telemetry flushes while the database is open
async () => {
await db.close();
console.log('[Server] Goodbye — keep the vibe alive.');
@@@ -240,6 +243,7 @@ function releaseEnv() {
// answers direct loopback callers only; through nginx it is a 404 (server/web/observability.js).
const observability = require('./web/observability');
const { registry: metricsRegistry } = observability.mountMetrics(app, { release });
++app.use(telemetry.middleware); // request telemetry (openvibe-sdk/service): a no-op until start() calls telemetry.init()
if (!drill.enabled) require('./ai/ai-media-analyze').registerMetrics(metricsRegistry);
const vibeCodingPublishServer = new VibeCodingPublishServer(db);
@@@ -1168,6 +1172,10 @@ async function start() {
await db.initDb();
await analyticsStore.ready();
console.log('[Server] Database ready');
++ // Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status
++ // class each minute, stored in telemetry_samples; rows older than 14 days go daily. A drill never gets here.
++ telemetry.init({ service: 'live', sink: createPgTelemetrySink({ db: db.getDb(), log: console }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
++ telemetryRetention = startTelemetryRetention({ db: db.getDb(), log: console, olderThanDays: 14 });
// Seed/refresh built-in themes on every start (upserts by slug, so re-tuned
// palettes always take effect; IDs stay stable).
ddiff --git a/test/service-kit.test.js b/test/service-kit.test.js
iindex e6688769..43758440 100644
---- a/test/service-kit.test.js
++++ b/test/service-kit.test.js
@@@ -18,7 +18,7 @@ const source = fs.readFileSync(path.join(__dirname, '../server/index.js'), 'utf8
const quiet = { log() {}, warn() {}, error() {} };
// ── Static: server/index.js ──────────────────────────────────────────────────────
--assert.match(source, /const \{ gracefulStop \} = require\('openvibe-sdk\/service'\)/);
++assert.match(source, /const \{ gracefulStop(, [A-Za-z]+)* \} = require\('openvibe-sdk\/service'\)/);
assert.strictEqual(source.split('gracefulStop(').length - 1, 1, 'exactly one gracefulStop( call');
const at = source.indexOf('gracefulStop(');
const call = source.slice(at, source.indexOf('\n});', at));
ddiff --git a/test/telemetry.test.js b/test/telemetry.test.js
nnew file mode 100644
iindex 00000000..0e4be3a2
---- /dev/null
++++ b/test/telemetry.test.js
@@@ -0,0 +1,54 @@
++'use strict';
++// Request telemetry (openvibe-sdk/service, plan T1 step 10): server/index.js mounts the process-wide middleware right after
++// the metrics, and start() calls telemetry.init() for service 'live' with a PostgreSQL sink and starts the 14-day
++// retention once the database is ready (a drill returns before that, so LIVE_DRILL starts neither); the flush runs
++// before the database closes. Through express, five hits on one route flush as ONE aggregated sample; migration 0006's
++// table takes the retention.
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const express = require('express');
++const { telemetry } = require('openvibe-sdk/service');
++const { pruneTelemetry } = require('openvibe-sdk/telemetry');
++
++(async () => {
++ const src = fs.readFileSync(path.join(__dirname, '..', 'server', 'index.js'), 'utf8');
++ assert.ok(src.indexOf('app.use(telemetry.middleware)') > src.indexOf('observability.mountMetrics(app'), 'the middleware follows the metrics');
++ const start = src.indexOf('async function start()');
++ const drillReturn = src.indexOf('if (drill.enabled) return await startDrill();', start);
++ const init = src.indexOf("telemetry.init({ service: 'live', sink: createPgTelemetrySink(", start);
++ assert.ok(start > 0 && drillReturn > start && init > drillReturn, 'init runs in start(), after a drill has returned');
++ assert.ok(src.indexOf('telemetryRetention = startTelemetryRetention({ db: db.getDb(), log: console, olderThanDays: 14 })', start) > drillReturn);
++ const flush = src.indexOf('async () => { if (telemetryRetention) await telemetryRetention.stop(); await telemetry.stop(); }');
++ assert.ok(flush > 0 && flush < src.indexOf('await db.close();'), 'the flush runs before the database closes');
++
++ const app = express();
++ app.use(telemetry.middleware);
++ app.get('/api/streams/:id', (req, res) => res.json({ ok: true }));
++ const server = await new Promise((r) => { const s = app.listen(0, '127.0.0.1', () => r(s)); });
++ const base = `http://127.0.0.1:${server.address().port}`;
++ const samples = [];
++ telemetry.init({ service: 'live', sink: async (batch) => { samples.push(...batch); }, intervalMs: 3600e3 });
++ try {
++ for (let i = 0; i < 5; i++) assert.strictEqual((await fetch(`${base}/api/streams/${i}`)).status, 200);
++ await telemetry.flush();
++ const hit = samples.filter((s) => s.operation === 'http.request' && s.resource === '/api/streams/:id' && s.status === '2xx');
++ assert.strictEqual(hit.length, 1, JSON.stringify(samples));
++ assert.strictEqual(hit[0].extra.count, 5, 'one sample per route template, never per URL');
++
++ const db = require('../server/db/database');
++ await db.initDb();
++ const raw = db.getDb();
++ await raw.query('DELETE FROM telemetry_samples');
++ const insert = (days) => raw.query("INSERT INTO telemetry_samples (at, service, operation) VALUES (now() - make_interval(days => $1), 'live', 'http.request')", [days]);
++ await insert(20);
++ await insert(1);
++ assert.strictEqual(await pruneTelemetry({ db: raw, olderThanDays: 14 }), 1);
++ assert.strictEqual((await raw.query('SELECT count(*)::int AS n FROM telemetry_samples')).rows[0].n, 1);
++ console.log('telemetry ok');
++ } finally {
++ await telemetry.stop();
++ await new Promise((r) => server.close(r));
++ }
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });