openvibers/openvibe.zone Commits

Zone: telemetry flushes each minute and keeps 14 days (#17)

OP OpenVibers committed 29 days ago · 0b40ba0c08 · 3 files, +11 −11

The request telemetry added earlier today flushed every 15 s and kept 30 days. The idle gauges alone write three rows per flush. Measured in production, that is about 650 rows an hour at about 270 B each including the indexes. At 30 days that's about 470k rows and 125 MB per service, roughly 4.4 GB across the estate on a disk with 16 GB free.

This changes it to one flush a minute and 14 days of retention: about 55k rows and 15 MB per service.

- The collector's interval (the tick, or `intervalMs` for `createHttpTelemetry` and `telemetry.init`) is 60 s.
- `startTelemetryRetention` passes `olderThanDays: 14`.
- The comments and `test/telemetry.test.js` follow: the retention check inserts rows 20 days and 1 day old and expects one deleted with `olderThanDays: 14`.

Applied migrations are unchanged (their comment still says 30 days; a migration that has run is never edited).

Changes

ddiff --git a/server/app.js b/server/app.js
iindex 2e86199..9c378e7 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -68,8 +68,8 @@ async function createApp(opts = {}) {
      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: 'zone', sink: opts.telemetrySink || createPgTelemetrySink({ db: s.db, log }), skipPrefixes: ['/shared'] });
++    // none. server/index.js flushes it on stop and keeps the table to 14 days.
++    ctx.telemetry = createHttpTelemetry({ service: 'zone', sink: opts.telemetrySink || createPgTelemetrySink({ db: s.db, log }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
      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);
ddiff --git a/server/index.js b/server/index.js
iindex 563b614..17794bd 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -29,11 +29,11 @@ async function start() {
          console.log(`[OpenVibe.Zone] ${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 });
++    // telemetry_samples rows older than 14 days go: a minute after start, then daily (openvibe-sdk/telemetry).
++    ctx.retention = startTelemetryRetention({ db: ctx.s.db, log: ctx.log || console, olderThanDays: 14 });
      // The request collector aggregates in memory and has no timer of its own (openvibe-sdk 0.46 createHttpTelemetry):
--    // this tick writes each 15-second interval's samples; the stop clears it and flushes once more.
--    ctx.telemetryTick = setInterval(() => { ctx.telemetry.flush().catch(() => {}); }, 15_000);
++    // this tick writes each minute's samples; the stop clears it and flushes once more.
++    ctx.telemetryTick = setInterval(() => { ctx.telemetry.flush().catch(() => {}); }, 60_000);
      ctx.telemetryTick.unref();
      ctx.keys.client.start();
      // The zone job (zone/zones.js tick): pushes what Media has not acknowledged and finishes deletions, every 15 s.
ddiff --git a/test/telemetry.test.js b/test/telemetry.test.js
iindex 0e23a71..c04fea8 100644
---- a/test/telemetry.test.js
++++ b/test/telemetry.test.js
@@@ -2,7 +2,7 @@
  /**
   * Request telemetry (openvibe-sdk/service, plan T1 step 10): app.js builds one collector for service 'zone' 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.
++ * /metrics and /shared files make none, and the 14-day retention deletes only older rows.
   */
  const assert = require('assert');
  const { pruneTelemetry } = require('openvibe-sdk/telemetry');
@@@ -11,7 +11,7 @@ const fs = require('fs');
  const path = require('path');
  
  (async () => {
--    await check('index.js flushes the collector every 15 s and clears that tick before the last flush on stop', () => {
++    await check('index.js flushes the collector every minute and clears that tick before the last flush on stop', () => {
          const index = fs.readFileSync(path.join(__dirname, '..', 'server', 'index.js'), 'utf8');
          assert.match(index, /ctx\.telemetryTick = setInterval\(\(\) => \{ ctx\.telemetry\.flush\(\)/);
          assert.match(index, /clearInterval\(ctx\.telemetryTick\); if \(ctx\.retention\) await ctx\.retention\.stop\(\); if \(ctx\.telemetry\) await ctx\.telemetry\.stop\(\);/);
@@@ -38,13 +38,13 @@ const path = require('path');
              assert.deepStrictEqual(requests(), []);
          });
  
--        await check('the table exists and the retention deletes rows older than 30 days, keeping newer ones', async () => {
++        await check('the table exists and the retention deletes rows older than 14 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), 'zone', 'http.request')", [days]);
--            await insert(45);
++            await insert(20);
              await insert(1);
--            assert.strictEqual(await pruneTelemetry({ db }), 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);
          });
      } finally { await t.close(); }