openvibers/openvibe.search Commits

Search: request telemetry in its own table, 30-day retention (#62)

OP OpenVibers committed 29 days ago · f4ec6d6df9 · 5 files, +87 −3

* Search: request telemetry in its own table, 30-day retention

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** `NNNN_telemetry.sql` (expand): the `telemetry_samples` table from `telemetrySchema()`.
- **`server/app.js`** mounts `telemetryMiddleware` before any route. It does nothing until `init()`, so tests and tools that build the app are unaffected.
- **`server/index.js`** (production start):
  - `telemetry.init({ service, sink: createPgTelemetrySink(...) })` aggregates requests per route template, method and status class. Every 15 s (the singleton's own timer) 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 30 days, a minute after start and then daily.
  - 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 30 days are kept).

* Telemetry: flush each minute, keep 14 days (about 15 MB a service instead of 125 MB)

Changes

ddiff --git a/migrations/0003_telemetry.sql b/migrations/0003_telemetry.sql
nnew file mode 100644
iindex 0000000..14260fa
---- /dev/null
++++ b/migrations/0003_telemetry.sql
@@@ -0,0 +1,28 @@
++-- phase: expand
++-- Request telemetry (openvibe-sdk/telemetry telemetrySchema()): one row per aggregated sample, pruned after 30 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/app.js b/server/app.js
iindex 62aa36b..0f244f7 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -25,6 +25,7 @@ function createApp({ config, db, store, engine, auth, outbox, purges, purger, sa
// HTTP golden signals by route template, process metrics, release_info and the Search gauges;
// GET /metrics answers direct loopback callers only (Track O).
const metrics = instrument(app, { service: 'search', release: release.release });
++ app.use(require('openvibe-sdk/service').telemetryMiddleware); // request telemetry: a no-op until server/index.js calls telemetry.init()
registerSearchGauges(metrics.registry, { db, engine, outbox, purges });
app.use(http.middleware());
// One W3C trace across services (openvibe-shared/trace): calls made while serving a request carry its traceparent.
ddiff --git a/server/index.js b/server/index.js
iindex 79009c1..ca9f08b 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -14,7 +14,8 @@ const { createStore } = require('./store');
const { createServiceOutbox } = require('openvibe-sdk/events');
const contracts = require('openvibe-contracts');
const { jwksClient } = require('openvibe-sdk/auth');
--const { gracefulStop } = require('openvibe-sdk/service');
++const { gracefulStop, telemetry } = require('openvibe-sdk/service');
++const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');
const { createAuth } = require('./auth');
const { createPurgeQueue, createPurger } = require('./purge');
const { createSavedSearches } = require('./saved');
@@@ -96,7 +97,11 @@ if (require.main === module) {
// SIGTERM/SIGINT (openvibe-sdk/service, docs/service.md's handles family): requests in flight get 8 s,
// then handles.close() (the prune timer and JWKS refresher stopped, the relay, purger and saved-search
// notifier stopped, the server closed, the database closed; a rejection exits 1); past 10 s the process exits 1.
-- gracefulStop({ name: 'search', server: handles.server, handles, drainMs: 8000, deadlineMs: 10000 });
++ // Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status
++ // class each minute, stored in this service's telemetry_samples table; rows older than 14 days go daily.
++ telemetry.init({ service: 'search', sink: createPgTelemetrySink({ db: handles.db, log: console }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
++ const telemetryRetention = startTelemetryRetention({ db: handles.db, log: console, olderThanDays: 14 });
++ gracefulStop({ name: 'search', server: handles.server, handles, drainMs: 8000, deadlineMs: 10000, stop: [async () => { if (telemetryRetention) await telemetryRetention.stop(); await telemetry.stop(); }] });
}).catch((err) => {
console.error(`[search] failed to start: ${err.stack || err}`);
process.exit(1);
ddiff --git a/test/service-kit.test.js b/test/service-kit.test.js
iindex 2b0bbaa..b14ee88 100644
---- a/test/service-kit.test.js
++++ b/test/service-kit.test.js
@@@ -18,7 +18,7 @@ const quiet = { log() {}, warn() {}, error() {} };
const source = fs.readFileSync(path.join(__dirname, '../server/index.js'), 'utf8');
t("server/index.js imports gracefulStop from openvibe-sdk/service and owns no signal handler", () => {
-- assert.match(source, /const \{ gracefulStop \} = require\('openvibe-sdk\/service'\)/);
++ assert.match(source, /const \{ gracefulStop(, [A-Za-z]+)* \} = require\('openvibe-sdk\/service'\)/);
assert.doesNotMatch(source, /process\.on\('SIGTERM'/);
assert.doesNotMatch(source, /process\.on\('SIGINT'/);
assert.doesNotMatch(source, /setTimeout\(\(\) => process\.exit\(1\), 10000\)/, 'the old 10 s exit-1 timer is gone');
ddiff --git a/test/telemetry.test.js b/test/telemetry.test.js
nnew file mode 100644
iindex 0000000..2ad5ed0
---- /dev/null
++++ b/test/telemetry.test.js
@@@ -0,0 +1,50 @@
++'use strict';
++/**
++ * Request telemetry (openvibe-sdk/service, plan T1 step 10): app.js mounts the process-wide telemetryMiddleware (a no-op
++ * until init), server/index.js calls telemetry.init() for service 'search' with a PostgreSQL sink, starts the 30-day
++ * retention and flushes on stop. Five hits on one route flush as ONE aggregated http.request sample; /metrics and
++ * /shared make none; the retention deletes only rows older than 14 days.
++ */
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const { telemetry } = require('openvibe-sdk/service');
++const { pruneTelemetry } = require('openvibe-sdk/telemetry');
++const { boot, request } = require('./helpers');
++(async () => {
++ const read = (p) => fs.readFileSync(path.join(__dirname, '..', p), 'utf8');
++ assert.match(read('server/index.js'), /telemetry\.init\(\{ service: 'search', sink: createPgTelemetrySink\(/, 'index.js starts the collector');
++ assert.match(read('server/index.js'), /startTelemetryRetention\(/, 'index.js starts the retention');
++ assert.match(read('server/index.js'), /[Rr]etention\.stop\(\); await telemetry\.stop\(\);/, 'the stop flushes');
++
++ const t = await boot();
++ const samples = [];
++ telemetry.init({ service: 'search', sink: async (batch) => { samples.push(...batch); }, intervalMs: 3600e3 });
++ const requests = () => samples.filter((s) => s.operation === 'http.request');
++ try {
++ for (let i = 0; i < 5; i++) assert.strictEqual((await request(t.base, 'GET', '/')).status, 200);
++ await telemetry.flush();
++ const hit = requests().filter((s) => s.resource === '/' && s.status === '2xx');
++ assert.strictEqual(hit.length, 1, JSON.stringify(requests()));
++ assert.strictEqual(hit[0].service, 'search');
++ assert.strictEqual(hit[0].extra.count, 5);
++
++ samples.length = 0;
++ await await request(t.base, 'GET', '/metrics');
++ await await request(t.base, 'GET', '/shared/navbar.js');
++ await telemetry.flush();
++ assert.deepStrictEqual(requests(), [], '/metrics and /shared make no sample');
++
++ const db = t.db;
++ await db.query('DELETE FROM telemetry_samples');
++ const insert = (days) => db.query("INSERT INTO telemetry_samples (at, service, operation) VALUES (now() - make_interval(days => $1), 'search', 'http.request')", [days]);
++ await insert(20);
++ await insert(1);
++ assert.strictEqual(await pruneTelemetry({ db, olderThanDays: 14 }), 1);
++ assert.strictEqual((await db.query('SELECT count(*)::int AS n FROM telemetry_samples')).rows[0].n, 1);
++ console.log('telemetry ok');
++ } finally {
++ await telemetry.stop();
++ await (t.stop || t.close)();
++ }
++})().catch((err) => { console.error(err); process.exit(1); });