openvibers/openvibe.inventory Commits

Inventory: write request telemetry every 15 s, not only at stop (#50)

OP OpenVibers committed 29 days ago · 4ecdb37571 · 2 files, +13 −1

The request telemetry added in the previous PR reached `telemetry_samples` only when the process stopped. In production the table stayed empty between deploys (Food: three requests at the origin, no rows after two flush intervals).

openvibe-sdk 0.46's `createHttpTelemetry` keeps each interval's aggregates in memory and moves them to the sink only in its own `flush()` or `stop()`. The SDK's inner timer flushes only the already-emitted buffer, and only `telemetry.init()` ticks the collector.

- `server/index.js` now calls `ctx.telemetry.flush()` every 15 s from `start()` (unref'd), so production writes each interval's samples. Tests aren't affected: they flush by hand.
- On stop, the tick is cleared before the retention stops and the last flush runs.
- `test/telemetry.test.js` checks that `index.js` does both.

Changes

ddiff --git a/server/index.js b/server/index.js
iindex cf49dc7..676183d 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -17,7 +17,7 @@ const { startSubscriptions } = require('openvibe-sdk/account-data');
  function createLifecycle({ server, ctx, exit, signals, timers = [], extra = [] }) {
      return gracefulStop({
          name: 'OpenVibe.Inventory', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,
--        close: [() => { for (const t of timers) clearInterval(t); }, ...extra, () => ctx.searchIndex.stop(), () => ctx.outbox.stop(), () => ctx.keys.client.stop(), async () => { if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); }, () => ctx.s.close()],
++        close: [() => { for (const t of timers) clearInterval(t); }, ...extra, () => ctx.searchIndex.stop(), () => ctx.outbox.stop(), () => ctx.keys.client.stop(), async () => { clearInterval(ctx.telemetryTick); if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); }, () => ctx.s.close()],
      });
  }
  
@@@ -31,6 +31,10 @@ async function start() {
      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 });
++    // 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);
++    ctx.telemetryTick.unref();
      ctx.keys.client.start();
      ctx.outbox.start();
      ctx.searchIndex.start();
ddiff --git a/test/telemetry.test.js b/test/telemetry.test.js
iindex 3c5b711..fbecb2d 100644
---- a/test/telemetry.test.js
++++ b/test/telemetry.test.js
@@@ -7,8 +7,16 @@
  const assert = require('assert');
  const { pruneTelemetry } = require('openvibe-sdk/telemetry');
  const { boot, check, done } = require('./helpers/boot');
++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', () => {
++        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\(\);/);
++    });
++
      const samples = [];
      const t = await boot({ telemetrySink: async (batch) => { samples.push(...batch); } });
      const requests = () => samples.filter((s) => s.operation === 'http.request');