openvibers/openvibe.reviews Commits

Reviews: telemetry flushes each minute and keeps 14 days (#62)

OP OpenVibers committed 29 days ago · 61d2bc841a · 2 files, +8 −8

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/index.js b/server/index.js
iindex c53b5a8..f8f5c56 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -62,8 +62,8 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
      const viewers = createViewerResolver({ keys, config });
      // 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 (app.js mounts the middleware). With workers
--    // on, a 15 s tick writes each interval (the collector has no timer of its own) and the table keeps 30 days.
--    const telemetry = createHttpTelemetry({ service: 'reviews', sink: telemetrySink || createPgTelemetrySink({ db, log }), skipPrefixes: ['/shared'] });
++    // on, a one-minute tick writes each interval (the collector has no timer of its own) and the table keeps 14 days.
++    const telemetry = createHttpTelemetry({ service: 'reviews', sink: telemetrySink || createPgTelemetrySink({ db, log }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
      let retention = null;
      const app = createApp({ config, telemetry, svc, viewers, platform, sources, sync, keys, db, log, rateLimits, fetchImpl, limitsNow, indexnow });
  
@@@ -72,8 +72,8 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
  
      const timers = [];
      if (workers) {
--        timers.push(setInterval(() => { telemetry.flush().catch(() => {}); }, 15_000));
--        retention = startTelemetryRetention({ db, log });
++        timers.push(setInterval(() => { telemetry.flush().catch(() => {}); }, 60_000));
++        retention = startTelemetryRetention({ db, log, olderThanDays: 14 });
          if (config.sync.enabled && sources.enabled) {
              const pull = () => sync.pull().then((r) => { if (!r.ok) log.warn(`[Reviews] Sources pull: ${r.reason}`); }).catch((err) => log.warn(`[Reviews] Sources pull: ${err.message}`));
              timers.push(setTimeout(pull, 5000));
ddiff --git a/test/telemetry.test.js b/test/telemetry.test.js
iindex 39f39cc..b46ceab 100644
---- a/test/telemetry.test.js
++++ b/test/telemetry.test.js
@@@ -3,7 +3,7 @@
   * Request telemetry (openvibe-sdk/service, plan T1 step 10): start() builds one collector for service 'reviews' and
   * createApp mounts it; a page hit five times flushes as ONE aggregated http.request sample (route template, 2xx, count
   * 5); /metrics and /shared files make none; the lifecycle flushes before the database closes; with workers on, a 15 s
-- * tick and the 30-day retention run; the retention deletes only rows older than 30 days.
++ * tick and the 14-day retention run; the retention deletes only rows older than 14 days.
   */
  const assert = require('assert');
  const fs = require('fs');
@@@ -14,7 +14,7 @@ const H = require('./helpers');
  (async () => {
      const index = fs.readFileSync(path.join(__dirname, '..', 'server', 'index.js'), 'utf8');
      assert.match(index, /timers\.push\(setInterval\(\(\) => \{ telemetry\.flush\(\)/, 'the workers tick flushes the collector');
--    assert.match(index, /retention = startTelemetryRetention\(\{ db, log \}\)/, 'the workers start the retention');
++    assert.match(index, /retention = startTelemetryRetention\(\{ db, log, olderThanDays: 14 \}\)/, 'the workers start the retention');
      assert.match(index, /close: \[\n\s+async \(\) => \{ if \(retention\) await retention\.stop\(\); if \(telemetry\) await telemetry\.stop\(\); \},/, 'the flush runs before the database closes');
  
      const samples = [];
@@@ -36,9 +36,9 @@ const H = require('./helpers');
  
          await h.db.query('DELETE FROM telemetry_samples');
          const insert = (days) => h.db.query("INSERT INTO telemetry_samples (at, service, operation) VALUES (now() - make_interval(days => $1), 'reviews', 'http.request')", [days]);
--        await insert(45);
++        await insert(20);
          await insert(1);
--        assert.strictEqual(await pruneTelemetry({ db: h.db }), 1);
++        assert.strictEqual(await pruneTelemetry({ db: h.db, olderThanDays: 14 }), 1);
          assert.strictEqual((await h.db.query('SELECT count(*)::int AS n FROM telemetry_samples')).rows[0].n, 1);
          console.log('telemetry ok');
      } finally {