openvibers/openvibe.chat Commits
Chat: request telemetry in its own table, 30-day retention (#99)
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);@@@ -77,6 +77,7 @@ function createApp({ chatServer, relay, events = null, callServer = null }) {// request rates/latencies and process metrics, from openvibe-shared (before any route).const release = require('openvibe-shared/release').createRelease({ service: 'chat', root: require('path').join(__dirname, '..'), packages: ['openvibe-shared', 'openvibe-sdk', 'openvibe-contracts'] });const metrics = require('openvibe-shared/metrics').instrument(app, { service: 'chat', release: release.release });++ app.use(require('openvibe-sdk/service').telemetryMiddleware); // request telemetry: a no-op until server/index.js calls telemetry.init()release.mount(app, { registry: metrics.registry });app.locals.metrics = metrics.registry;// How sign-ins resolve (auth/network-session.js): here, through Live, or refused.@@@ -23,7 +23,8 @@ const subscriptions = require('./events/subscriptions');const chatAi = require('./ai/chat-ai');const { limits } = require('./net/actor-limits');const { createValkey } = require('openvibe-sdk/valkey');--const { gracefulStop } = require('openvibe-sdk/service');++const { gracefulStop, telemetry } = require('openvibe-sdk/service');++const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');const { createApp } = require('./app');let rejectionsLogged = false;@@@ -67,6 +68,10 @@ async function start() {await new Promise((resolve) => server.listen(config.port, config.host, resolve));console.log(`[Chat] listening on http://${config.host}:${server.address().port} (ws /ws/chat${config.calls.enabled ? ', /ws/call' : ''})`);const subs = subscriptions.startAtBoot({ config, port: server.address().port });++ // 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: 'chat', sink: createPgTelemetrySink({ db: db.getDb(), log: console }), skipPrefixes: ['/shared'], intervalMs: 60_000 });++ const telemetryRetention = startTelemetryRetention({ db: db.getDb(), log: console, olderThanDays: 14 });// systemd sends SIGTERM (SIGINT by hand); openvibe-sdk/service's gracefulStop takes the signal, runs the stop// steps in order (nothing new starts), drains the HTTP server, runs the close steps, then exits 0. drainMs@@@ -94,7 +99,7 @@ async function start() {() => chatServer.close(),() => callServer.close(),],-- close: [() => db.close()],++ close: [async () => { if (telemetryRetention) await telemetryRetention.stop(); await telemetry.stop(); }, () => db.close()],});return { server, relay, events, callServer, subscriptions: subs, shutdown };}@@@ -28,7 +28,7 @@ const TABLES = ['ctx_users', 'ctx_streams', 'ctx_managed_streams', 'ctx_channels', 'ctx_sync', 'events_outbox', 'service_outbox','chat_ingress_applied', 'audio_requests', 'import_hold', 'import_runs', 'chat_meta', 'deploy_releases','chat_event_inbox', 'calls', 'rooms', 'room_members', 'room_messages', 'room_attachments', 'token_revocations', 'chat_control_log',-- 'network_blocks', 'account_data_events', 'ticket_conversations', 'ticket_messages',++ 'network_blocks', 'account_data_events', 'ticket_conversations', 'ticket_messages', 'telemetry_samples',];const INDEXES = ['idx_chat_ts_deleted', 'idx_chat_page_live', 'idx_chat_channel_user_ts', 'idx_chat_stream_ts', 'idx_emotes_channel_code', 'idx_chat_tl_dedup',nnew file mode 100644@@@ -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 'chat' with a PostgreSQL sink, starts the 14-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 } = require('./helpers');++(async () => {++ const read = (p) => fs.readFileSync(path.join(__dirname, '..', p), 'utf8');++ assert.match(read('server/index.js'), /telemetry\.init\(\{ service: 'chat', 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: 'chat', 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 fetch(t.base + '/')).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, 'chat');++ assert.strictEqual(hit[0].extra.count, 5);++++ samples.length = 0;++ await await fetch(t.base + '/metrics');++ await await fetch(t.base + '/shared/navbar.js');++ await telemetry.flush();++ assert.deepStrictEqual(requests(), [], '/metrics and /shared make no sample');++++ const db = require('../server/db/database').getDb();++ 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), 'chat', '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);++ process.stdout.write('telemetry ok\n');++ } finally {++ await telemetry.stop();++ if (t && t.close) await t.close();++ }++})().then(() => process.exit(0), (err) => { process.stderr.write(`${(err && err.stack) || err}\n`); process.exit(1); }); // the helper keeps sockets open
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);
@@ -77,6 +77,7 @@ function createApp({ chatServer, relay, events = null, callServer = null }) {
@@ -77,6 +77,7 @@ function createApp({ chatServer, relay, events = null, callServer = null }) {
// request rates/latencies and process metrics, from openvibe-shared (before any route).
// request rates/latencies and process metrics, from openvibe-shared (before any route).
const release = require('openvibe-shared/release').createRelease({ service: 'chat', root: require('path').join(__dirname, '..'), packages: ['openvibe-shared', 'openvibe-sdk', 'openvibe-contracts'] });
const release = require('openvibe-shared/release').createRelease({ service: 'chat', root: require('path').join(__dirname, '..'), packages: ['openvibe-shared', 'openvibe-sdk', 'openvibe-contracts'] });
const metrics = require('openvibe-shared/metrics').instrument(app, { service: 'chat', release: release.release });
const metrics = require('openvibe-shared/metrics').instrument(app, { service: 'chat', release: release.release });
+ app.use(require('openvibe-sdk/service').telemetryMiddleware); // request telemetry: a no-op until server/index.js calls telemetry.init()
release.mount(app, { registry: metrics.registry });
release.mount(app, { registry: metrics.registry });
app.locals.metrics = metrics.registry;
app.locals.metrics = metrics.registry;
// How sign-ins resolve (auth/network-session.js): here, through Live, or refused.
// How sign-ins resolve (auth/network-session.js): here, through Live, or refused.
@@ -23,7 +23,8 @@ const subscriptions = require('./events/subscriptions');
@@ -23,7 +23,8 @@ const subscriptions = require('./events/subscriptions');
const chatAi = require('./ai/chat-ai');
const chatAi = require('./ai/chat-ai');
const { limits } = require('./net/actor-limits');
const { limits } = require('./net/actor-limits');
const { createValkey } = require('openvibe-sdk/valkey');
const { createValkey } = require('openvibe-sdk/valkey');
-const { gracefulStop } = require('openvibe-sdk/service');
+const { gracefulStop, telemetry } = require('openvibe-sdk/service');
+const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');
const { createApp } = require('./app');
const { createApp } = require('./app');
let rejectionsLogged = false;
let rejectionsLogged = false;
@@ -67,6 +68,10 @@ async function start() {
@@ -67,6 +68,10 @@ async function start() {
await new Promise((resolve) => server.listen(config.port, config.host, resolve));
await new Promise((resolve) => server.listen(config.port, config.host, resolve));
console.log(`[Chat] listening on http://${config.host}:${server.address().port} (ws /ws/chat${config.calls.enabled ? ', /ws/call' : ''})`);
console.log(`[Chat] listening on http://${config.host}:${server.address().port} (ws /ws/chat${config.calls.enabled ? ', /ws/call' : ''})`);
const subs = subscriptions.startAtBoot({ config, port: server.address().port });
const subs = subscriptions.startAtBoot({ config, port: server.address().port });
+ // 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: 'chat', sink: createPgTelemetrySink({ db: db.getDb(), log: console }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
+ const telemetryRetention = startTelemetryRetention({ db: db.getDb(), log: console, olderThanDays: 14 });
// systemd sends SIGTERM (SIGINT by hand); openvibe-sdk/service's gracefulStop takes the signal, runs the stop
// systemd sends SIGTERM (SIGINT by hand); openvibe-sdk/service's gracefulStop takes the signal, runs the stop
// steps in order (nothing new starts), drains the HTTP server, runs the close steps, then exits 0. drainMs
// steps in order (nothing new starts), drains the HTTP server, runs the close steps, then exits 0. drainMs
@@ -94,7 +99,7 @@ async function start() {
@@ -94,7 +99,7 @@ async function start() {
() => chatServer.close(),
() => chatServer.close(),
() => callServer.close(),
() => callServer.close(),
],
],
- close: [() => db.close()],
+ close: [async () => { if (telemetryRetention) await telemetryRetention.stop(); await telemetry.stop(); }, () => db.close()],
});
});
return { server, relay, events, callServer, subscriptions: subs, shutdown };
return { server, relay, events, callServer, subscriptions: subs, shutdown };
}
}
@@ -28,7 +28,7 @@ const TABLES = [
@@ -28,7 +28,7 @@ const TABLES = [
'ctx_users', 'ctx_streams', 'ctx_managed_streams', 'ctx_channels', 'ctx_sync', 'events_outbox', 'service_outbox',
'ctx_users', 'ctx_streams', 'ctx_managed_streams', 'ctx_channels', 'ctx_sync', 'events_outbox', 'service_outbox',
'chat_ingress_applied', 'audio_requests', 'import_hold', 'import_runs', 'chat_meta', 'deploy_releases',
'chat_ingress_applied', 'audio_requests', 'import_hold', 'import_runs', 'chat_meta', 'deploy_releases',
'chat_event_inbox', 'calls', 'rooms', 'room_members', 'room_messages', 'room_attachments', 'token_revocations', 'chat_control_log',
'chat_event_inbox', 'calls', 'rooms', 'room_members', 'room_messages', 'room_attachments', 'token_revocations', 'chat_control_log',
- 'network_blocks', 'account_data_events', 'ticket_conversations', 'ticket_messages',
+ 'network_blocks', 'account_data_events', 'ticket_conversations', 'ticket_messages', 'telemetry_samples',
];
];
const INDEXES = [
const INDEXES = [
'idx_chat_ts_deleted', 'idx_chat_page_live', 'idx_chat_channel_user_ts', 'idx_chat_stream_ts', 'idx_emotes_channel_code', 'idx_chat_tl_dedup',
'idx_chat_ts_deleted', 'idx_chat_page_live', 'idx_chat_channel_user_ts', 'idx_chat_stream_ts', 'idx_emotes_channel_code', 'idx_chat_tl_dedup',
new file mode 100644
new file mode 100644
@@ -0,0 +1,50 @@
@@ -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 'chat' with a PostgreSQL sink, starts the 14-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 } = require('./helpers');
+(async () => {
+ const read = (p) => fs.readFileSync(path.join(__dirname, '..', p), 'utf8');
+ assert.match(read('server/index.js'), /telemetry\.init\(\{ service: 'chat', 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: 'chat', 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 fetch(t.base + '/')).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, 'chat');
+ assert.strictEqual(hit[0].extra.count, 5);
+
+ samples.length = 0;
+ await await fetch(t.base + '/metrics');
+ await await fetch(t.base + '/shared/navbar.js');
+ await telemetry.flush();
+ assert.deepStrictEqual(requests(), [], '/metrics and /shared make no sample');
+
+ const db = require('../server/db/database').getDb();
+ 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), 'chat', '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);
+ process.stdout.write('telemetry ok\n');
+ } finally {
+ await telemetry.stop();
+ if (t && t.close) await t.close();
+ }
+})().then(() => process.exit(0), (err) => { process.stderr.write(`${(err && err.stack) || err}\n`); process.exit(1); }); // the helper keeps sockets open