openvibers/openvibe.blog Commits

Store request telemetry in Postgres with daily retention (#42)

OP OpenVibers committed 30 days ago · b716782783 · 8 files, +117 −9

* Store request telemetry in Postgres with daily retention

Asked: T1 step 10b, dossier /home/workstation/openvibe/agents/ds/prep/T1.md lines 162-188, with the exact wiring in /home/workstation/openvibe/agents/jobs/sdk-t1-step-10-first-half-011009.out.md section 'Wiring for step 10b (OpenVibe.Blog)' (line 37 to the end). ONE pull request in OpenVibe.Blog from curr…
Harness job blog-t1-step-10b-dossier-home-workstatio-014215.

* Blog: telemetry retention through openvibe-sdk 0.46 startTelemetryRetention (first prune a minute after start)

Changes

ddiff --git a/migrations/0003_telemetry.sql b/migrations/0003_telemetry.sql
nnew file mode 100644
iindex 0000000..23dbf54
---- /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 by the worker.
++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/package-lock.json b/package-lock.json
iindex bcda84c..74913c1 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -17,7 +17,7 @@
"iovalkey": "^0.4.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",
"openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.3.0",
-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.46.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"pg": "^8.23.0"
},
@@@ -907,9 +907,9 @@
}
},
"node_modules/openvibe-sdk": {
-- "version": "0.44.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
-- "integrity": "sha512-8OjkIXVPUrGaOriBeHGK5GDoz4IRpjpKt7knZ5HSZnUpi0ND1Z3XD1YET4HiRydweT/rQy9/1h3oKod+tAP7Hg==",
++ "version": "0.46.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.46.0",
++ "integrity": "sha512-heZ06pqRNd8MKdSNZHGYe7FzHsZJxYz+IeWnTePWHshmqRI3qiPeRu7Xg/aCWiYOdIgLn3DUXw+vXhN/aw+LuA==",
"license": "MIT",
"engines": {
"node": ">=20"
ddiff --git a/package.json b/package.json
iindex 65722af..ddd3af1 100644
---- a/package.json
++++ b/package.json
@@@ -29,7 +29,7 @@
"iovalkey": "^0.4.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.143.0",
"openvibe-publishing": "https://codeload.github.com/OpenVibers/OpenVibe.Publishing/tar.gz/refs/tags/v1.3.0",
-- "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.46.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.3",
"pg": "^8.23.0"
},
ddiff --git a/server/app.js b/server/app.js
iindex cd16085..a5a554d 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -106,6 +106,17 @@ async function createApp(opts = {}) {
require('./render/layout').setRelease(release.release);
const metrics = require('openvibe-shared/metrics').instrument(app, { service: 'blog', release: release.release });
app.locals.metrics = metrics.registry;
++ // Request telemetry (openvibe-sdk/service): one aggregated sample per route template, status class and flush
++ // window, stored in Blog's own telemetry_samples table; static assets and /metrics produce none.
++ const { createHttpTelemetry } = require('openvibe-sdk/service');
++ const { createPgTelemetrySink } = require('openvibe-sdk/telemetry');
++ const telemetry = createHttpTelemetry({
++ service: 'blog',
++ sink: opts.telemetrySink || createPgTelemetrySink({ db: store.db, log }),
++ skipPrefixes: ['/shared'],
++ });
++ app.use(telemetry.middleware());
++ ctx.telemetry = telemetry;
app.locals.ctx = ctx;
// Per-actor limits (http/actor-limits.js) for the API, the editor and comment posts, counted once each
// router resolved req.viewer; the per-address limits below stay.
ddiff --git a/server/index.js b/server/index.js
iindex 1c7b63f..55359f2 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -17,7 +17,7 @@ const { startSubscriptions } = require('openvibe-sdk/account-data');
function createLifecycle({ server, ctx, exit, signals, extra = [] }) {
return gracefulStop({
name: 'Blog', server, deadlineExitCode: 0, exit, signals,
-- stop: [() => ctx.worker.stop(), () => ctx.changelog.stop(), ...extra],
++ stop: [() => ctx.telemetry.stop(), () => ctx.worker.stop(), () => ctx.changelog.stop(), ...extra],
close: [() => ctx.outbox.stop(), () => ctx.store.close()],
});
}
ddiff --git a/server/worker.js b/server/worker.js
iindex 8f2150d..c89cb1a 100644
---- a/server/worker.js
++++ b/server/worker.js
@@@ -1,6 +1,6 @@
'use strict';
/**
-- * Background work in the Blog process (openvibe-shared has no job runner; this is two timers):
++ * Background work in the Blog process (openvibe-shared has no job runner; this is three timers):
*
* scheduled publication every BLOG_SCHEDULE_INTERVAL_MS: claim due jobs (leased, so two
* processes never run the same job at once) and apply them; the effect
@@@ -8,10 +8,15 @@
* media verification every BLOG_MEDIA_VERIFY_INTERVAL_MS: ask OpenVibe.Media whether the
* objects attached to published posts still exist; gone → the explicit
* broken-asset state; an outage changes nothing (check_failed)
++ * telemetry retention openvibe-sdk/telemetry startTelemetryRetention(): a minute after start, then once a day,
++ * telemetry_samples rows older than 30 days are deleted (a redeploy never pushes it a day out)
*/
++const { pruneTelemetry, startTelemetryRetention } = require('openvibe-sdk/telemetry');
++
function createWorker({ config, store, posts, media, outbox, log = console }) {
let scheduleTimer = null;
let mediaTimer = null;
++ let retention = null;
let running = false;
async function scheduleTick() {
@@@ -40,8 +45,13 @@ function createWorker({ config, store, posts, media, outbox, log = console }) {
return out;
}
++ async function pruneSamples() {
++ try { return await pruneTelemetry({ db: store.db }); } catch (err) { log.warn('[Blog] telemetry retention failed:', err.message); return null; }
++ }
++
return {
scheduleTick,
++ pruneSamples,
verifyMedia,
start() {
if (!config.worker.enabled) return;
@@@ -51,9 +61,10 @@ function createWorker({ config, store, posts, media, outbox, log = console }) {
mediaTimer = setInterval(() => verifyMedia().catch((err) => log.warn('[Blog] media verification failed:', err.message)), config.media.verifyIntervalMs);
mediaTimer.unref();
}
++ retention = startTelemetryRetention({ db: store.db, log });
setTimeout(scheduleTick, 1000).unref();
},
-- stop() { clearInterval(scheduleTimer); clearInterval(mediaTimer); },
++ async stop() { clearInterval(scheduleTimer); clearInterval(mediaTimer); if (retention) await retention.stop(); },
};
}
ddiff --git a/test/helpers/boot.js b/test/helpers/boot.js
iindex 6283224..fe80c99 100644
---- a/test/helpers/boot.js
++++ b/test/helpers/boot.js
@@@ -45,7 +45,7 @@ async function boot(opts = {}) {
let built = null;
async function start() {
const config = configLib.load(env);
-- built = await createApp({ config, store: createStore(testdb.db, { now: clock.now }), now: clock.now, log: opts.log || quiet, entitlementCheck: opts.entitlementCheck, limitsNow: opts.limitsNow, indexnow: opts.indexnow, accountSend: opts.accountSend });
++ built = await createApp({ config, store: createStore(testdb.db, { now: clock.now }), now: clock.now, log: opts.log || quiet, entitlementCheck: opts.entitlementCheck, limitsNow: opts.limitsNow, indexnow: opts.indexnow, accountSend: opts.accountSend, telemetrySink: opts.telemetrySink });
// Warm the shared JWKS cache so an immediately-verified token does not race the first fetch.
await require('openvibe-sdk/auth').jwksClient(`${network.url}/api/.well-known/jwks`).keysForKid(null);
server = await new Promise((resolve) => { const s = http.createServer(built.app); s.listen(0, '127.0.0.1', () => resolve(s)); });
ddiff --git a/test/telemetry.test.js b/test/telemetry.test.js
nnew file mode 100644
iindex 0000000..33aafb3
---- /dev/null
++++ b/test/telemetry.test.js
@@@ -0,0 +1,58 @@
++'use strict';
++/**
++ * Request telemetry (openvibe-sdk/service, plan T1 step 10b): app.js builds one collector for service 'blog' and
++ * mounts its middleware, index.js stops it first, and a route hit five times flushes as ONE aggregated
++ * http.request sample labelled with the route template and status class. /metrics and static assets make none.
++ */
++const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
++const { boot, check, done } = require('./helpers/boot');
++
++const read = (p) => fs.readFileSync(path.join(__dirname, '..', p), 'utf8');
++
++(async () => {
++ await check('app.js creates the blog collector and mounts its middleware; index.js stops it first', () => {
++ const app = read('server/app.js');
++ assert.match(app, /createHttpTelemetry\(\{[^}]*service: 'blog'/s);
++ assert.match(app, /app\.use\(telemetry\.middleware\(\)\)/);
++ assert.match(app, /ctx\.telemetry = telemetry/);
++ assert.match(read('server/index.js'), /stop: \[\(\) => ctx\.telemetry\.stop\(\),/);
++ });
++
++ const samples = [];
++ const t = await boot({ telemetrySink: async (batch) => { samples.push(...batch); } });
++ const requests = () => samples.filter((s) => s.operation === 'http.request');
++
++ await check('five hits on one API route flush as one http.request sample: template, 2xx, count 5', async () => {
++ await Promise.all([1, 2, 3, 4, 5].map(async () => assert.strictEqual((await t.get('/api/v1/changelog')).status, 200)));
++ await t.ctx.telemetry.flush();
++ const mine = requests().filter((s) => /^\/api\/v1\/changelog\b/.test(s.resource || '') && s.status === '2xx');
++ assert.strictEqual(mine.length, 1, JSON.stringify(requests()));
++ assert.strictEqual(mine[0].service, 'blog');
++ assert.strictEqual(mine[0].resource, '/api/v1/changelog');
++ assert.strictEqual(mine[0].extra.count, 5);
++ });
++
++ await check('/metrics and a static asset produce no sample', async () => {
++ samples.length = 0;
++ await t.get('/metrics');
++ const css = await t.get('/css/blog.css');
++ assert.strictEqual(css.status, 200);
++ await t.ctx.telemetry.flush();
++ assert.deepStrictEqual(requests(), []);
++ });
++
++ await check('the worker retention drops telemetry_samples rows older than 30 days and keeps newer ones', async () => {
++ const db = t.ctx.store.db;
++ const insert = (days) => db.query(`INSERT INTO telemetry_samples (at, service, operation) VALUES (now() - make_interval(days => $1), 'blog', 'http.request')`, [days]);
++ await insert(45);
++ await insert(1);
++ assert.strictEqual(await t.ctx.worker.pruneSamples(), 1);
++ const left = await db.query('SELECT count(*)::int AS n FROM telemetry_samples');
++ assert.strictEqual(left.rows[0].n, 1);
++ });
++
++ await t.close();
++ done();
++})().catch((err) => { console.error(err); process.exit(1); });