openvibers/openvibe.tools Commits

Tools: request telemetry for the gateway and every satellite (#69)

OP OpenVibers committed 29 days ago · 40f485a108 · 11 files, +154 −8

Plan T1 step 10 (universal telemetry) for Tools' eight processes, using openvibe-sdk 0.46's process-wide collector.

- **Migration** `0006_telemetry.sql` (expand): the `telemetry_samples` table from `telemetrySchema()`, in the one `tools` database.
- **`apps/_shared/observe.js`**: a new `telemetry` option, openvibe-sdk/service's `telemetry`. When given, its middleware is mounted right after the metrics. It does nothing until `init()`.
- **Every app's `server/index.js`** (gateway and the seven satellites):
  - passes `telemetry` to `observe()`;
  - calls `telemetry.init()` with its existing service id (`tools`, `tools-img`, …) and a PostgreSQL sink. Each minute that writes one aggregated `http.request` sample per route template, method and status class, plus the HTTP gauges.
  - flushes before the database closes.
  - The requires sit directly above the `observe()` call, which runs before most of each file's other requires.
- **Retention**: only the gateway runs `startTelemetryRetention` (14 days), because all eight apps write one table.
- **Volume**: about 15 MB per process with indexes.
- **`apps/_shared/test/telemetry.test.js`**:
  - checks each app's wiring in the source;
  - checks that five hits through `observe()` flush as one sample (count 5) and that `/metrics` makes none;
  - checks that the migration's table takes the retention.
- The observe test, which boots every real satellite and the gateway, passes with the new startup.

Changes

ddiff --git a/apps/_shared/observe.js b/apps/_shared/observe.js
iindex 4f7a793..adff2ce 100644
---- a/apps/_shared/observe.js
++++ b/apps/_shared/observe.js
@@@ -32,6 +32,8 @@ const path = require('path');
* @param {Function} [o.skip] requests never recorded (long-lived streams)
* @param {Function} [o.details] extra readiness fields
* @param {boolean} [o.mountReady] false: the caller mounts readiness.handler where it belongs
++ * @param {object} [o.telemetry] openvibe-sdk/service's `telemetry` (request telemetry, plan T1): its middleware is
++ * mounted here, a no-op until the app's index.js calls telemetry.init()
*/
function observe(o) {
const app = o.app;
@@@ -48,6 +50,7 @@ function observe(o) {
// Job progress streams (SSE) last as long as the job; their duration is not a latency.
skip: (req) => /^\/api\/v1\/jobs\/[^/]+\/events$/.test(req.path) || (o.skip ? o.skip(req) : false),
});
++ if (o.telemetry) app.use(o.telemetry.middleware);
const appName = o.service.replace(/^tools-?/, '') || 'gateway';
if (o.jobs) {
inst.registry.gauge({
ddiff --git a/apps/_shared/test/telemetry.test.js b/apps/_shared/test/telemetry.test.js
nnew file mode 100644
iindex 0000000..3a1cbd6
---- /dev/null
++++ b/apps/_shared/test/telemetry.test.js
@@@ -0,0 +1,66 @@
++'use strict';
++// Request telemetry for Tools (plan T1 step 10): observe() mounts openvibe-sdk/service's process-wide telemetry middleware
++// when the app passes it (a no-op until init); every app passes it and calls telemetry.init() with its own service id
++// into the tools database; the gateway alone prunes the shared table to 14 days. Five hits on one route flush as ONE
++// aggregated http.request sample, /metrics makes none, and the migration's table takes the retention.
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const { dep } = require('./deps');
++const { observe } = require('../observe');
++const { testDb, closeAllTestDbs } = require('./testdb');
++
++const express = dep('express');
++const metrics = dep('openvibe-shared/metrics');
++const ready = dep('openvibe-shared/ready');
++const { telemetry } = dep('openvibe-sdk/service');
++const { pruneTelemetry } = dep('openvibe-sdk/telemetry');
++const SERVICES = {'audio': 'tools-audio', 'docs': 'tools-docs', 'food': 'tools-food', 'gateway': 'tools', 'img': 'tools-img', 'maps': 'tools-maps', 'text': 'tools-text', 'yt': 'tools-yt'};
++
++(async () => {
++ const apps = path.resolve(__dirname, '..', '..');
++ for (const [app, service] of Object.entries(SERVICES)) {
++ const src = fs.readFileSync(path.join(apps, app, 'server', 'index.js'), 'utf8');
++ assert.match(src, /observe\(\{ telemetry,/, `${app} passes telemetry to observe()`);
++ assert.ok(src.includes(`telemetry.init({ service: '${service}', sink: createPgTelemetrySink({ db: toolsDb.db`), `${app} starts its collector`);
++ assert.ok(/telemetry\.stop\(\)/.test(src), `${app} flushes on stop`);
++ assert.strictEqual(/startTelemetryRetention\(/.test(src), app === 'gateway', `${app}: only the gateway prunes`);
++ assert.ok(src.indexOf("require('openvibe-sdk/service')") < src.indexOf('observe({ telemetry,'), `${app} requires telemetry before observe() uses it`);
++ }
++
++ const app = express();
++ observe({ app, metrics, ready, service: 'tools-test', release: 'test', telemetry });
++ app.get('/hello', (req, res) => res.send('hi'));
++ const server = await new Promise((r) => { const s = app.listen(0, '127.0.0.1', () => r(s)); });
++ const base = `http://127.0.0.1:${server.address().port}`;
++ const samples = [];
++ telemetry.init({ service: 'tools-test', sink: async (batch) => { samples.push(...batch); }, intervalMs: 3600e3 });
++ const requests = () => samples.filter((s) => s.operation === 'http.request');
++ const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'tools-telemetry-'));
++ try {
++ for (let i = 0; i < 5; i++) assert.strictEqual((await fetch(`${base}/hello`)).status, 200);
++ await telemetry.flush();
++ const hit = requests().filter((s) => s.resource === '/hello' && s.status === '2xx');
++ assert.strictEqual(hit.length, 1, JSON.stringify(requests()));
++ assert.strictEqual(hit[0].service, 'tools-test');
++ assert.strictEqual(hit[0].extra.count, 5);
++ samples.length = 0;
++ await fetch(`${base}/metrics`);
++ await telemetry.flush();
++ assert.deepStrictEqual(requests(), [], '/metrics makes no sample');
++
++ const db = await testDb(dir);
++ const insert = (days) => db.query("INSERT INTO telemetry_samples (at, service, operation) VALUES (now() - make_interval(days => $1), 'tools', '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 new Promise((r) => server.close(r));
++ await closeAllTestDbs();
++ fs.rmSync(dir, { recursive: true, force: true });
++ }
++})().catch((err) => { console.error(err); process.exit(1); });
ddiff --git a/apps/audio/server/index.js b/apps/audio/server/index.js
iindex eefffe3..3f72b56 100644
---- a/apps/audio/server/index.js
++++ b/apps/audio/server/index.js
@@@ -64,7 +64,9 @@ const { observe, checks: ready, which } = require('../../_shared/observe');
// Serve only once the schema is in: the first PGlite migration must not run inside a request (it
// would block the event loop and time out a satellite's readiness check).
app.use((req, res, next) => toolsDb.ready.then(() => next(), next));
--const obs = observe({
++const { telemetry } = require('openvibe-sdk/service'); // request telemetry (apps/_shared/observe.js, plan T1)
++const { createPgTelemetrySink } = require('openvibe-sdk/telemetry');
++const obs = observe({ telemetry,
app, metrics: require('openvibe-shared/metrics'), ready: require('openvibe-shared/ready'),
service: 'tools-audio', release: release.release,
checks: [
@@@ -328,6 +330,9 @@ const server = app.listen(config.port, config.host, () => {
// end); the server stops taking connections and lets requests in flight finish (4 s at most); then
// recent-tool writes are flushed (1 s at most), the analytics, the jobs store and the guard close,
// and the process exits 0, within the manifest's 5 s.
++// Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status class
++// each minute, into the tools database's telemetry_samples table (the gateway prunes it).
++telemetry.init({ service: 'tools-audio', sink: createPgTelemetrySink({ db: toolsDb.db, log: console }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
require('../../_shared/graceful').gracefulStop({
name: 'Audio.OpenVibe', server,
stop: [
@@@ -339,6 +344,7 @@ require('../../_shared/graceful').gracefulStop({
() => analytics.destroy(),
() => jobs.close(), // running jobs stay 'running' in the tools database; the next boot re-queues them
() => guard.close(),
++ () => telemetry.stop(),
() => toolsDb.db.close(),
() => toolsValkey && toolsValkey.close(),
],
ddiff --git a/apps/docs/server/index.js b/apps/docs/server/index.js
iindex bd6e03f..8c801b2 100644
---- a/apps/docs/server/index.js
++++ b/apps/docs/server/index.js
@@@ -81,7 +81,9 @@ const { observe, checks: ready } = require('../../_shared/observe');
// Serve only once the schema is in: the first PGlite migration must not run inside a request (it
// would block the event loop and time out a satellite's readiness check).
app.use((req, res, next) => toolsDb.ready.then(() => next(), next));
--const obs = observe({
++const { telemetry } = require('openvibe-sdk/service'); // request telemetry (apps/_shared/observe.js, plan T1)
++const { createPgTelemetrySink } = require('openvibe-sdk/telemetry');
++const obs = observe({ telemetry,
app, metrics: require('openvibe-shared/metrics'), ready: require('openvibe-shared/ready'),
service: 'tools-docs', release: release.release,
checks: [
@@@ -328,6 +330,9 @@ const server = app.listen(config.port, config.host, () => {
// end); the server stops taking connections and lets requests in flight finish (4 s at most); then
// recent-tool writes are flushed (1 s at most), the analytics, the jobs store, the worker pool and the
// guard close, and the process exits 0, within the manifest's 5 s.
++// Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status class
++// each minute, into the tools database's telemetry_samples table (the gateway prunes it).
++telemetry.init({ service: 'tools-docs', sink: createPgTelemetrySink({ db: toolsDb.db, log: console }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
require('../../_shared/graceful').gracefulStop({
name: 'Docs.OpenVibe', server,
stop: [
@@@ -340,6 +345,7 @@ require('../../_shared/graceful').gracefulStop({
() => jobs.close(), // running jobs stay 'running' in the tools database; the next boot re-queues them
() => pool.close(),
() => guard.close(),
++ () => telemetry.stop(),
() => toolsDb.db.close(),
() => toolsValkey && toolsValkey.close(),
],
ddiff --git a/apps/food/server/index.js b/apps/food/server/index.js
iindex 2d8a42b..cd372e2 100644
---- a/apps/food/server/index.js
++++ b/apps/food/server/index.js
@@@ -46,7 +46,9 @@ const { observe, checks: ready } = require('../../_shared/observe');
// Serve only once the schema is in: the first PGlite migration must not run inside a request (it
// would block the event loop and time out a satellite's readiness check).
app.use((req, res, next) => toolsDb.ready.then(() => next(), next));
--const obs = observe({
++const { telemetry } = require('openvibe-sdk/service'); // request telemetry (apps/_shared/observe.js, plan T1)
++const { createPgTelemetrySink } = require('openvibe-sdk/telemetry');
++const obs = observe({ telemetry,
app, metrics: require('openvibe-shared/metrics'), ready: require('openvibe-shared/ready'),
service: 'tools-food', release: release.release,
checks: [
@@@ -171,11 +173,15 @@ const server = app.listen(PORT, '127.0.0.1', () => {
// ── Graceful stop (roadmap WS-P lifecycle; apps/_shared/graceful.js) ──
// SIGTERM: the server stops taking connections and lets requests in flight finish (4 s at most);
// then the analytics and the guard close, and the process exits 0, within the manifest's 5 s.
++// Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status class
++// each minute, into the tools database's telemetry_samples table (the gateway prunes it).
++telemetry.init({ service: 'tools-food', sink: createPgTelemetrySink({ db: toolsDb.db, log: console }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
require('../../_shared/graceful').gracefulStop({
name: 'Food.OpenVibe', server,
close: [
() => analytics.destroy(),
() => guard.close(),
++ () => telemetry.stop(),
() => toolsDb.db.close(),
() => toolsValkey && toolsValkey.close(),
],
ddiff --git a/apps/gateway/server/index.js b/apps/gateway/server/index.js
iindex a466ada..d31363e 100644
---- a/apps/gateway/server/index.js
++++ b/apps/gateway/server/index.js
@@@ -72,7 +72,9 @@ const guard = createGuard({
// Serve only once the schema is in: the first PGlite migration must not run inside a request (it
// would block the event loop and time out a satellite's readiness check).
app.use((req, res, next) => toolsDb.ready.then(() => next(), next));
--const obs = observe({
++const { telemetry } = require('openvibe-sdk/service'); // request telemetry (apps/_shared/observe.js, plan T1)
++const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');
++const obs = observe({ telemetry,
app, metrics: require('openvibe-shared/metrics'), ready: require('openvibe-shared/ready'),
service: 'tools', release: release.release, mountReady: false,
normalize: (req) => {
@@@ -520,6 +522,10 @@ const server = app.listen(config.port, config.host, () => {
// taking connections and lets requests in flight finish (4 s at most; event streams are closed and
// reconnect); then recent-tool writes are flushed (1 s at most), the engine pool, the guard and the
// cutoff store close, and the process exits 0 — within the manifest's 5 s.
++// Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status class
++// each minute, into the tools database's telemetry_samples table; the gateway keeps it to 14 days for every app.
++telemetry.init({ service: 'tools', sink: createPgTelemetrySink({ db: toolsDb.db, log: console }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
++const telemetryRetention = startTelemetryRetention({ db: toolsDb.db, log: console, olderThanDays: 14 });
require('../../_shared/graceful').gracefulStop({
name: 'Tools gateway', server,
stop: [
@@@ -532,6 +538,7 @@ require('../../_shared/graceful').gracefulStop({
() => require('../../_shared/usage').stopRecorder(800),
() => runApi.pool.close(),
() => guard.close(),
++ async () => { await telemetryRetention.stop(); await telemetry.stop(); },
() => toolsDb.db.close(),
() => toolsValkey && toolsValkey.close(),
() => require('./revocation-events').close(),
ddiff --git a/apps/img/server/index.js b/apps/img/server/index.js
iindex c018c10..91b98cf 100644
---- a/apps/img/server/index.js
++++ b/apps/img/server/index.js
@@@ -70,7 +70,9 @@ const { observe, checks: ready } = require('../../_shared/observe');
// Serve only once the schema is in: the first PGlite migration must not run inside a request (it
// would block the event loop and time out a satellite's readiness check).
app.use((req, res, next) => toolsDb.ready.then(() => next(), next));
--const obs = observe({
++const { telemetry } = require('openvibe-sdk/service'); // request telemetry (apps/_shared/observe.js, plan T1)
++const { createPgTelemetrySink } = require('openvibe-sdk/telemetry');
++const obs = observe({ telemetry,
app, metrics: require('openvibe-shared/metrics'), ready: require('openvibe-shared/ready'),
service: 'tools-img', release: release.release,
checks: [
@@@ -293,6 +295,9 @@ const server = app.listen(config.port, config.host, () => {
// end); the server stops taking connections and lets requests in flight finish (4 s at most); then
// recent-tool writes are flushed (1 s at most), the analytics, the jobs store, the worker pool and the
// guard close, and the process exits 0, within the manifest's 5 s.
++// Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status class
++// each minute, into the tools database's telemetry_samples table (the gateway prunes it).
++telemetry.init({ service: 'tools-img', sink: createPgTelemetrySink({ db: toolsDb.db, log: console }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
require('../../_shared/graceful').gracefulStop({
name: 'Img.OpenVibe', server,
stop: [
@@@ -305,6 +310,7 @@ require('../../_shared/graceful').gracefulStop({
() => jobs.close(), // running jobs stay 'running' in the tools database; the next boot re-queues them
() => pool.close(),
() => guard.close(),
++ () => telemetry.stop(),
() => toolsDb.db.close(),
() => toolsValkey && toolsValkey.close(),
],
ddiff --git a/apps/maps/server/index.js b/apps/maps/server/index.js
iindex e59c849..af41841 100644
---- a/apps/maps/server/index.js
++++ b/apps/maps/server/index.js
@@@ -50,7 +50,9 @@ const { observe, checks: ready } = require('../../_shared/observe');
// Serve only once the schema is in: the first PGlite migration must not run inside a request (it
// would block the event loop and time out a satellite's readiness check).
app.use((req, res, next) => toolsDb.ready.then(() => next(), next));
--const obs = observe({
++const { telemetry } = require('openvibe-sdk/service'); // request telemetry (apps/_shared/observe.js, plan T1)
++const { createPgTelemetrySink } = require('openvibe-sdk/telemetry');
++const obs = observe({ telemetry,
app, metrics: require('openvibe-shared/metrics'), ready: require('openvibe-shared/ready'),
service: 'tools-maps', release: release.release,
checks: [
@@@ -500,12 +502,16 @@ const server = app.listen(config.port, config.host, () => {
// SIGTERM: the server stops taking connections and lets requests in flight finish (4 s at most;
// event streams are closed and reconnect); then recent-tool writes are flushed (1 s at most), the
// analytics and the guard close, and the process exits 0, within the manifest's 5 s.
++// Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status class
++// each minute, into the tools database's telemetry_samples table (the gateway prunes it).
++telemetry.init({ service: 'tools-maps', sink: createPgTelemetrySink({ db: toolsDb.db, log: console }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
require('../../_shared/graceful').gracefulStop({
name: 'Maps.OpenVibe', server,
close: [
() => require('../../_shared/usage').stopRecorder(800),
() => analytics.destroy(),
() => guard.close(),
++ () => telemetry.stop(),
() => toolsDb.db.close(),
() => toolsValkey && toolsValkey.close(),
],
ddiff --git a/apps/text/server/index.js b/apps/text/server/index.js
iindex 5dc849c..27f2a87 100644
---- a/apps/text/server/index.js
++++ b/apps/text/server/index.js
@@@ -41,7 +41,9 @@ const { observe, checks: ready } = require('../../_shared/observe');
// Serve only once the schema is in: the first PGlite migration must not run inside a request (it
// would block the event loop and time out a satellite's readiness check).
app.use((req, res, next) => toolsDb.ready.then(() => next(), next));
--const obs = observe({
++const { telemetry } = require('openvibe-sdk/service'); // request telemetry (apps/_shared/observe.js, plan T1)
++const { createPgTelemetrySink } = require('openvibe-sdk/telemetry');
++const obs = observe({ telemetry,
app, metrics: require('openvibe-shared/metrics'), ready: require('openvibe-shared/ready'),
service: 'tools-text', release: release.release,
checks: [
@@@ -182,12 +184,16 @@ const server = app.listen(config.port, config.host, () => {
// SIGTERM: the server stops taking connections and lets requests in flight finish (4 s at most;
// event streams are closed and reconnect); then recent-tool writes are flushed (1 s at most), the
// analytics and the guard close, and the process exits 0, within the manifest's 5 s.
++// Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status class
++// each minute, into the tools database's telemetry_samples table (the gateway prunes it).
++telemetry.init({ service: 'tools-text', sink: createPgTelemetrySink({ db: toolsDb.db, log: console }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
require('../../_shared/graceful').gracefulStop({
name: 'Text.OpenVibe', server,
close: [
() => require('../../_shared/usage').stopRecorder(800),
() => analytics.destroy(),
() => guard.close(),
++ () => telemetry.stop(),
() => toolsDb.db.close(),
() => toolsValkey && toolsValkey.close(),
],
ddiff --git a/apps/yt/server/index.js b/apps/yt/server/index.js
iindex bf95d51..887fc01 100644
---- a/apps/yt/server/index.js
++++ b/apps/yt/server/index.js
@@@ -46,7 +46,9 @@ const { observe, checks: ready, which } = require('../../_shared/observe');
// Serve only once the schema is in: the first PGlite migration must not run inside a request (it
// would block the event loop and time out a satellite's readiness check).
app.use((req, res, next) => toolsDb.ready.then(() => next(), next));
--const obs = observe({
++const { telemetry } = require('openvibe-sdk/service'); // request telemetry (apps/_shared/observe.js, plan T1)
++const { createPgTelemetrySink } = require('openvibe-sdk/telemetry');
++const obs = observe({ telemetry,
app, metrics: require('openvibe-shared/metrics'), ready: require('openvibe-shared/ready'),
service: 'tools-yt', release: release.release,
checks: [
@@@ -331,6 +333,9 @@ const server = app.listen(config.port, config.host, () => {
// in flight finish (4 s at most; progress streams are closed and reconnect); then recent-tool writes
// are flushed (1 s at most), the analytics and the guard close, and the process exits 0, within the
// manifest's 5 s.
++// Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status class
++// each minute, into the tools database's telemetry_samples table (the gateway prunes it).
++telemetry.init({ service: 'tools-yt', sink: createPgTelemetrySink({ db: toolsDb.db, log: console }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
require('../../_shared/graceful').gracefulStop({
name: 'YT.OpenVibe', server,
stop: [
@@@ -340,6 +345,7 @@ require('../../_shared/graceful').gracefulStop({
() => require('../../_shared/usage').stopRecorder(800),
() => analytics.destroy(),
() => guard.close(),
++ () => telemetry.stop(),
() => toolsDb.db.close(),
() => toolsValkey && toolsValkey.close(),
],
ddiff --git a/migrations/0006_telemetry.sql b/migrations/0006_telemetry.sql
nnew file mode 100644
iindex 0000000..14260fa
---- /dev/null
++++ b/migrations/0006_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);