openvibers/openvibe.food Commits
Food: write request telemetry every 15 s, not only at stop (#45)
Changes
@@@ -18,7 +18,7 @@ function createLifecycle({ server, ctx, exit, signals, timers, extra = [] }) {const clearTimers = timers || [ctx.geoPruneTimer].filter(Boolean);return gracefulStop({name: 'OpenVibe.Food', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,-- close: [() => { for (const t of clearTimers) clearInterval(t); }, () => ctx.keys.client.stop(), async () => { if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); }, () => ctx.s.close(), ...extra],++ close: [() => { for (const t of clearTimers) clearInterval(t); }, () => 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(), ...extra],});}@@@ -33,6 +33,10 @@ async function start() {ctx.keys.client.start();// 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();// Subscribe to the two ADR-033 topics at OpenVibe.Events (idempotent; off without FOOD_EVENTS_URL and// FOOD_EVENTS_SECRET). The consumer itself is mounted in server/app.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');
@@ -18,7 +18,7 @@ function createLifecycle({ server, ctx, exit, signals, timers, extra = [] }) {
@@ -18,7 +18,7 @@ function createLifecycle({ server, ctx, exit, signals, timers, extra = [] }) {
const clearTimers = timers || [ctx.geoPruneTimer].filter(Boolean);
const clearTimers = timers || [ctx.geoPruneTimer].filter(Boolean);
return gracefulStop({
return gracefulStop({
name: 'OpenVibe.Food', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,
name: 'OpenVibe.Food', server, deadlineExitCode: 0, exit, signals, deadlineMs: 10_000,
- close: [() => { for (const t of clearTimers) clearInterval(t); }, () => ctx.keys.client.stop(), async () => { if (ctx.retention) await ctx.retention.stop(); if (ctx.telemetry) await ctx.telemetry.stop(); }, () => ctx.s.close(), ...extra],
+ close: [() => { for (const t of clearTimers) clearInterval(t); }, () => 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(), ...extra],
});
});
}
}
@@ -33,6 +33,10 @@ async function start() {
@@ -33,6 +33,10 @@ async function start() {
ctx.keys.client.start();
ctx.keys.client.start();
// telemetry_samples rows older than 30 days go: a minute after start, then daily (openvibe-sdk/telemetry).
// 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 });
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();
// Subscribe to the two ADR-033 topics at OpenVibe.Events (idempotent; off without FOOD_EVENTS_URL and
// Subscribe to the two ADR-033 topics at OpenVibe.Events (idempotent; off without FOOD_EVENTS_URL and
// FOOD_EVENTS_SECRET). The consumer itself is mounted in server/app.js.
// FOOD_EVENTS_SECRET). The consumer itself is mounted in server/app.js.
@@ -7,8 +7,16 @@
@@ -7,8 +7,16 @@
const assert = require('assert');
const assert = require('assert');
const { pruneTelemetry } = require('openvibe-sdk/telemetry');
const { pruneTelemetry } = require('openvibe-sdk/telemetry');
const { boot, check, done } = require('./helpers/boot');
const { boot, check, done } = require('./helpers/boot');
+const fs = require('fs');
+const path = require('path');
(async () => {
(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 samples = [];
const t = await boot({ telemetrySink: async (batch) => { samples.push(...batch); } });
const t = await boot({ telemetrySink: async (batch) => { samples.push(...batch); } });
const requests = () => samples.filter((s) => s.operation === 'http.request');
const requests = () => samples.filter((s) => s.operation === 'http.request');