openvibers/openvibe.trade Commits

Trade: telemetry flushes each minute and keeps 14 days (#65)

OP OpenVibers committed 29 days ago · c7aba33243 · 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 0225769..9ffc290 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -125,8 +125,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 every 15 s and on stop, and keeps the table to 30 days.
--    ctx.telemetry = createHttpTelemetry({ service: 'trade', sink: opts.telemetrySink || createPgTelemetrySink({ db: store.db, log }), skipPrefixes: ['/shared'] });
++    // none. server/index.js flushes it every minute and on stop, and keeps the table to 14 days.
++    ctx.telemetry = createHttpTelemetry({ service: 'trade', sink: opts.telemetrySink || createPgTelemetrySink({ db: store.db, log }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
      app.use(ctx.telemetry.middleware());
      app.locals.ctx = ctx;
      // Per-actor limits (http/actor-limits.js) on /api/v1 and the forms, counted once each router resolved
ddiff --git a/server/index.js b/server/index.js
iindex 60d6293..50f50f0 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -23,11 +23,11 @@ const server = app.listen(config.port, config.host, () => {
  server.keepAliveTimeout = 65_000;
  ctx.outbox.start();
  ctx.worker.start();
--// telemetry_samples rows older than 30 days go: a minute after start, then daily (openvibe-sdk/telemetry).
--ctx.retention = startTelemetryRetention({ db: ctx.store.db, log: console });
++// telemetry_samples rows older than 14 days go: a minute after start, then daily (openvibe-sdk/telemetry).
++ctx.retention = startTelemetryRetention({ db: ctx.store.db, 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();
  // The two account subscriptions at OpenVibe.Events (ADR-033), created when missing; off without EVENTS_URL,
  // TRADE_EVENTS_WEBHOOK_SECRET or the client secret. The sources.* one stays scripts/subscribe.js's.
ddiff --git a/test/telemetry.test.js b/test/telemetry.test.js
iindex 0a5d7c4..08cea00 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 'trade' 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.store.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), 'trade', '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(); }