openvibers/openvibe.play Commits
Play: request telemetry in its own table, 30-day retention (#12)
Changes
nnew file mode 100644@@@ -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);@@@ -38,6 +38,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,@@@ -68,6 +70,11 @@ async function createApp(opts = {}) {require('./render/layout').setRelease(release.release);const metrics = require('openvibe-shared/metrics').instrument(app, { service: 'play', 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: 'play', 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;@@@ -6,6 +6,7 @@*/const { createApp } = require('./app');const { gracefulStop } = require('openvibe-sdk/service');++const { startTelemetryRetention } = require('openvibe-sdk/telemetry');const { startSubscriptions, TOPICS } = require('openvibe-sdk/account-data');const subjectMerge = require('./play/subject-merge');@@@ -17,7 +18,7 @@ const subjectMerge = require('./play/subject-merge');function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {return gracefulStop({name: 'OpenVibe.Play', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,-- close: [() => { for (const t of timers) clearInterval(t); }, ...extra, () => ctx.keys.client.stop(), () => ctx.s.close()],++ close: [() => { for (const t of timers) clearInterval(t); }, ...extra, () => ctx.keys.client.stop(), async () => { if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); }, () => ctx.s.close()],});}@@@ -29,6 +30,8 @@ async function start() {console.log(`[OpenVibe.Play] ${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();// The account subscriptions at OpenVibe.Events (ADR-033) and account merges (ADR-029), created when missing; off// without EVENTS_URL, PLAY_EVENTS_SECRET or the client secret.@@@ -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, limitsNow: opts.limitsNow, callerLimits: opts.callerLimits === true, accountSend: opts.accountSend });++ const built = await createApp({ config, store, log, telemetrySink: opts.telemetrySink, limitsNow: opts.limitsNow, callerLimits: opts.callerLimits === true, accountSend: opts.accountSend });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}`;@@@ -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);});nnew file mode 100644@@@ -0,0 +1,44 @@++'use strict';++/**++ * Request telemetry (openvibe-sdk/service, plan T1 step 10): app.js builds one collector for service 'play' 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, 'play');++ 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), 'play', '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); });
new file mode 100644
new file mode 100644
@@ -0,0 +1,28 @@
@@ -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);
@@ -38,6 +38,8 @@ const { html } = require('./render/html');
@@ -38,6 +38,8 @@ const { html } = require('./render/html');
const PUBLIC_DIR = path.join(__dirname, '..', 'public');
const PUBLIC_DIR = path.join(__dirname, '..', 'public');
const VERSION = require('../package.json').version;
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,
* opts: config, store, now (clock), fetchImpl (Network), log, limitsNow, callerLimits (false: count nobody,
@@ -68,6 +70,11 @@ async function createApp(opts = {}) {
@@ -68,6 +70,11 @@ async function createApp(opts = {}) {
require('./render/layout').setRelease(release.release);
require('./render/layout').setRelease(release.release);
const metrics = require('openvibe-shared/metrics').instrument(app, { service: 'play', release: release.release });
const metrics = require('openvibe-shared/metrics').instrument(app, { service: 'play', release: release.release });
app.locals.metrics = metrics.registry;
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: 'play', sink: opts.telemetrySink || createPgTelemetrySink({ db: s.db, log }), skipPrefixes: ['/shared'] });
+ app.use(ctx.telemetry.middleware());
app.locals.ctx = ctx;
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);
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;
ctx.valkey = valkey;
@@ -6,6 +6,7 @@
@@ -6,6 +6,7 @@
*/
*/
const { createApp } = require('./app');
const { createApp } = require('./app');
const { gracefulStop } = require('openvibe-sdk/service');
const { gracefulStop } = require('openvibe-sdk/service');
+const { startTelemetryRetention } = require('openvibe-sdk/telemetry');
const { startSubscriptions, TOPICS } = require('openvibe-sdk/account-data');
const { startSubscriptions, TOPICS } = require('openvibe-sdk/account-data');
const subjectMerge = require('./play/subject-merge');
const subjectMerge = require('./play/subject-merge');
@@ -17,7 +18,7 @@ const subjectMerge = require('./play/subject-merge');
@@ -17,7 +18,7 @@ const subjectMerge = require('./play/subject-merge');
function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {
function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {
return gracefulStop({
return gracefulStop({
name: 'OpenVibe.Play', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,
name: 'OpenVibe.Play', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,
- close: [() => { for (const t of timers) clearInterval(t); }, ...extra, () => ctx.keys.client.stop(), () => ctx.s.close()],
+ close: [() => { for (const t of timers) clearInterval(t); }, ...extra, () => ctx.keys.client.stop(), async () => { if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); }, () => ctx.s.close()],
});
});
}
}
@@ -29,6 +30,8 @@ async function start() {
@@ -29,6 +30,8 @@ async function start() {
console.log(`[OpenVibe.Play] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl} (db ${ctx.s.db.store})`);
console.log(`[OpenVibe.Play] ${config.nodeEnv} on http://${config.host}:${config.port} → ${config.baseUrl} (db ${ctx.s.db.store})`);
});
});
server.keepAliveTimeout = 65_000;
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.keys.client.start();
// The account subscriptions at OpenVibe.Events (ADR-033) and account merges (ADR-029), created when missing; off
// The account subscriptions at OpenVibe.Events (ADR-033) and account merges (ADR-029), created when missing; off
// without EVENTS_URL, PLAY_EVENTS_SECRET or the client secret.
// without EVENTS_URL, PLAY_EVENTS_SECRET or the client secret.
@@ -38,7 +38,7 @@ async function boot(opts = {}) {
@@ -38,7 +38,7 @@ async function boot(opts = {}) {
const { createStore } = require('../../server/db');
const { createStore } = require('../../server/db');
const testdb = await require('./db').testDb();
const testdb = await require('./db').testDb();
const store = createStore(testdb.db, { now: opts.now });
const store = createStore(testdb.db, { now: opts.now });
- const built = await createApp({ config, store, log, limitsNow: opts.limitsNow, callerLimits: opts.callerLimits === true, accountSend: opts.accountSend });
+ const built = await createApp({ config, store, log, telemetrySink: opts.telemetrySink, limitsNow: opts.limitsNow, callerLimits: opts.callerLimits === true, accountSend: opts.accountSend });
await built.ctx.keys.ensure();
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 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}`;
const base = `http://127.0.0.1:${server.address().port}`;
@@ -16,6 +16,7 @@ const listen = (server) => new Promise((resolve) => server.listen(0, '127.0.0.1'
@@ -16,6 +16,7 @@ const listen = (server) => new Promise((resolve) => server.listen(0, '127.0.0.1'
const steps = [];
const steps = [];
const ctx = {
const ctx = {
keys: { client: { stop: () => { steps.push('keys.stop'); } } },
keys: { client: { stop: () => { steps.push('keys.stop'); } } },
+ telemetry: { stop: async () => { steps.push('telemetry.stop'); } },
s: { close: async () => { steps.push('store.close'); } },
s: { close: async () => { steps.push('store.close'); } },
};
};
const server = http.createServer((req, res) => res.end('ok'));
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'
@@ -28,7 +29,7 @@ const listen = (server) => new Promise((resolve) => server.listen(0, '127.0.0.1'
assert.strictEqual(code, 0);
assert.strictEqual(code, 0);
assert.deepStrictEqual(exits, [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);
assert.strictEqual(server.listening, false);
});
});
new file mode 100644
new file mode 100644
@@ -0,0 +1,44 @@
@@ -0,0 +1,44 @@
+'use strict';
+/**
+ * Request telemetry (openvibe-sdk/service, plan T1 step 10): app.js builds one collector for service 'play' 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, 'play');
+ 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), 'play', '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); });