openvibers/openvibe.rent Commits

Rent: request telemetry in its own table, 30-day retention (#47)

OP OpenVibers committed 29 days ago · bc90228197 · 6 files, +86 −3

Plan T1 step 10 (universal telemetry): this service now records its request telemetry the way Blog and Network do, using openvibe-sdk 0.46.

- **Migration** `NNNN_telemetry.sql` (expand): the `telemetry_samples` table from `telemetrySchema()`.
- **`server/app.js`**: one `createHttpTelemetry` collector, which aggregates requests per route template, method and status class per 15-second flush (one `http.request` sample per key, not one row per request). It writes through `createPgTelemetrySink` into this service's own table. `/shared`, health, readiness and `/metrics` produce no samples.
- **`server/index.js`**:
  - `startTelemetryRetention` deletes rows older than 30 days, a minute after start and then daily.
  - On stop, the retention timers stop and the collector flushes once before the store closes.
- **Tests**:
  - `test/telemetry.test.js`: five home-page hits flush as one sample (template `/`, 2xx, count 5); `/metrics` and `/shared` produce none; the retention keeps only rows from the last 30 days.
  - The lifecycle test's expected close order now includes the flush.

Changes

ddiff --git a/migrations/0007_telemetry.sql b/migrations/0007_telemetry.sql
nnew file mode 100644
iindex 0000000..14260fa
---- /dev/null
++++ b/migrations/0007_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 b450b27..eb6e31d 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -41,6 +41,8 @@ const { html } = require('./render/html');
const PUBLIC_DIR = path.join(__dirname, '..', 'public');
const VERSION = require('../package.json').version;
++const { createHttpTelemetry } = require('openvibe-sdk/service');
++const { createPgTelemetrySink } = require('openvibe-sdk/telemetry');
/**
* opts: config, store, now (clock), fetchImpl (Network), log, limitsNow, callerLimits (false: count nobody,
@@@ -98,6 +100,11 @@ async function createApp(opts = {}) {
require('./render/layout').setRelease(release.release);
const metrics = require('openvibe-shared/metrics').instrument(app, { service: 'rent', release: release.release });
app.locals.metrics = metrics.registry;
++ // Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status
++ // class each flush, stored in this service's telemetry_samples table; /shared, health, readiness and /metrics make
++ // none. server/index.js flushes it on stop and keeps the table to 30 days.
++ ctx.telemetry = createHttpTelemetry({ service: 'rent', sink: opts.telemetrySink || createPgTelemetrySink({ db: s.db, log }), skipPrefixes: ['/shared'] });
++ app.use(ctx.telemetry.middleware());
app.locals.ctx = ctx;
const valkey = opts.valkey !== undefined ? opts.valkey : (config.valkey.url ? require('openvibe-sdk/valkey').createValkey({ url: config.valkey.url, prefix: config.valkey.prefix, log }) : null);
ctx.valkey = valkey;
ddiff --git a/server/index.js b/server/index.js
iindex 91b1614..11d7972 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -9,6 +9,7 @@
*/
const { createApp } = require('./app');
const { gracefulStop } = require('openvibe-sdk/service');
++const { startTelemetryRetention } = require('openvibe-sdk/telemetry');
const { startExpiryTimer } = require('./listings/expiry');
const { startHpiTimer } = require('./housing/hpi');
@@@ -20,7 +21,7 @@ const { startHpiTimer } = require('./housing/hpi');
function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {
return gracefulStop({
name: 'OpenVibe.Rent', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,
-- close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.keys.client.stop(), () => ctx.search.stop(), () => ctx.s.close(), ...extra],
++ close: [() => { for (const t of timers) clearInterval(t); }, () => ctx.keys.client.stop(), () => ctx.search.stop(), async () => { if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); }, () => ctx.s.close(), ...extra],
});
}
@@@ -32,6 +33,8 @@ async function start() {
console.log(`[OpenVibe.Rent] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl} (db ${ctx.s.db.store})`);
});
server.keepAliveTimeout = 65_000;
++ // telemetry_samples rows older than 30 days go: a minute after start, then daily (openvibe-sdk/telemetry).
++ ctx.retention = startTelemetryRetention({ db: ctx.s.db, log: ctx.log || console });
ctx.keys.client.start();
ctx.search.start();
ddiff --git a/test/helpers/boot.js b/test/helpers/boot.js
iindex 9b48bea..fe50566 100644
---- a/test/helpers/boot.js
++++ b/test/helpers/boot.js
@@@ -38,7 +38,7 @@ async function boot(opts = {}) {
const { createStore } = require('../../server/db');
const testdb = await require('./db').testDb();
const store = createStore(testdb.db, { now: opts.now });
-- const built = await createApp({ config, store, log, indexnow: opts.indexnow, limitsNow: opts.limitsNow, callerLimits: opts.callerLimits === true, accountSend: opts.accountSend, housingFetchImpl: opts.housingFetchImpl });
++ const built = await createApp({ config, store, log, telemetrySink: opts.telemetrySink, indexnow: opts.indexnow, limitsNow: opts.limitsNow, callerLimits: opts.callerLimits === true, accountSend: opts.accountSend, housingFetchImpl: opts.housingFetchImpl });
await built.ctx.keys.ensure();
const server = await new Promise((resolve) => { const s = http.createServer(built.app); s.listen(0, '127.0.0.1', () => resolve(s)); });
const base = `http://127.0.0.1:${server.address().port}`;
ddiff --git a/test/service-kit.test.js b/test/service-kit.test.js
iindex 2a82b23..261eec8 100644
---- a/test/service-kit.test.js
++++ b/test/service-kit.test.js
@@@ -16,6 +16,7 @@ const listen = (server) => new Promise((resolve) => server.listen(0, '127.0.0.1'
const steps = [];
const ctx = {
keys: { client: { stop: () => { steps.push('keys.stop'); } } },
++ telemetry: { stop: async () => { steps.push('telemetry.stop'); } },
s: { close: async () => { steps.push('store.close'); } },
};
const server = http.createServer((req, res) => res.end('ok'));
@@@ -28,7 +29,7 @@ const listen = (server) => new Promise((resolve) => server.listen(0, '127.0.0.1'
assert.strictEqual(code, 0);
assert.deepStrictEqual(exits, [0]);
-- assert.deepStrictEqual(steps, ['keys.stop', 'store.close']);
++ assert.deepStrictEqual(steps, ['keys.stop', 'telemetry.stop', 'store.close']);
assert.strictEqual(server.listening, false);
});
ddiff --git a/test/telemetry.test.js b/test/telemetry.test.js
nnew file mode 100644
iindex 0000000..d6d1e69
---- /dev/null
++++ b/test/telemetry.test.js
@@@ -0,0 +1,44 @@
++'use strict';
++/**
++ * Request telemetry (openvibe-sdk/service, plan T1 step 10): app.js builds one collector for service 'rent' with a
++ * PostgreSQL sink, a page hit five times flushes as ONE aggregated http.request sample (route template, 2xx, count 5),
++ * /metrics and /shared files make none, and the 30-day retention deletes only older rows.
++ */
++const assert = require('assert');
++const { pruneTelemetry } = require('openvibe-sdk/telemetry');
++const { boot, check, done } = require('./helpers/boot');
++
++(async () => {
++ const samples = [];
++ const t = await boot({ telemetrySink: async (batch) => { samples.push(...batch); } });
++ const requests = () => samples.filter((s) => s.operation === 'http.request');
++ try {
++ await check('five hits on the home page flush as one http.request sample: template, 2xx, count 5', async () => {
++ for (let i = 0; i < 5; i++) assert.strictEqual((await t.get('/')).status, 200);
++ await t.ctx.telemetry.flush();
++ const home = requests().filter((s) => s.resource === '/' && s.status === '2xx');
++ assert.strictEqual(home.length, 1, JSON.stringify(requests()));
++ assert.strictEqual(home[0].service, 'rent');
++ assert.strictEqual(home[0].extra.count, 5);
++ });
++
++ await check('/metrics and a /shared file make no sample', async () => {
++ samples.length = 0;
++ await t.get('/metrics');
++ await t.get('/shared/navbar.js');
++ await t.ctx.telemetry.flush();
++ assert.deepStrictEqual(requests(), []);
++ });
++
++ await check('the table exists and the retention deletes rows older than 30 days, keeping newer ones', async () => {
++ const db = t.ctx.s.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), 'rent', 'http.request')", [days]);
++ await insert(45);
++ await insert(1);
++ assert.strictEqual(await pruneTelemetry({ db }), 1);
++ assert.strictEqual((await db.query('SELECT count(*)::int AS n FROM telemetry_samples')).rows[0].n, 1);
++ });
++ } finally { await t.close(); }
++ done();
++})().catch((e) => { console.error(e); process.exit(1); });