openvibers/openvibe.wiki Commits
Wiki: telemetry flushes each minute and keeps 14 days (#62)
Changes
@@@ -83,15 +83,15 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch: null);// 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: 'wiki', 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: 'wiki', sink: telemetrySink || createPgTelemetrySink({ db, log }), skipPrefixes: ['/shared'], intervalMs: 60_000 });let retention = null;const app = createApp({ config, telemetry, svc, viewers, platform, keys, db, valkey, log, rateLimits, fetchImpl, limitsNow, indexnow, accountData, accountSend });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 });let scheduling = false;timers.push(setInterval(async () => {if (scheduling) return;@@@ -3,7 +3,7 @@* Request telemetry (openvibe-sdk/service, plan T1 step 10): start() builds one collector for service 'wiki' 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), 'wiki', '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 {
@@ -83,15 +83,15 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
@@ -83,15 +83,15 @@ async function start({ config, db: givenDb = null, now = () => Date.now(), fetch
: null);
: null);
// Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status
// 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
// 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: 'wiki', 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: 'wiki', sink: telemetrySink || createPgTelemetrySink({ db, log }), skipPrefixes: ['/shared'], intervalMs: 60_000 });
let retention = null;
let retention = null;
const app = createApp({ config, telemetry, svc, viewers, platform, keys, db, valkey, log, rateLimits, fetchImpl, limitsNow, indexnow, accountData, accountSend });
const app = createApp({ config, telemetry, svc, viewers, platform, keys, db, valkey, log, rateLimits, fetchImpl, limitsNow, indexnow, accountData, accountSend });
const timers = [];
const timers = [];
if (workers) {
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 });
let scheduling = false;
let scheduling = false;
timers.push(setInterval(async () => {
timers.push(setInterval(async () => {
if (scheduling) return;
if (scheduling) return;
@@ -3,7 +3,7 @@
@@ -3,7 +3,7 @@
* Request telemetry (openvibe-sdk/service, plan T1 step 10): start() builds one collector for service 'wiki' and
* Request telemetry (openvibe-sdk/service, plan T1 step 10): start() builds one collector for service 'wiki' and
* createApp mounts it; a page hit five times flushes as ONE aggregated http.request sample (route template, 2xx, count
* 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
* 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 assert = require('assert');
const fs = require('fs');
const fs = require('fs');
@@ -14,7 +14,7 @@ const H = require('./helpers');
@@ -14,7 +14,7 @@ const H = require('./helpers');
(async () => {
(async () => {
const index = fs.readFileSync(path.join(__dirname, '..', 'server', 'index.js'), 'utf8');
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, /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');
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 = [];
const samples = [];
@@ -36,9 +36,9 @@ const H = require('./helpers');
@@ -36,9 +36,9 @@ const H = require('./helpers');
await h.db.query('DELETE FROM telemetry_samples');
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), 'wiki', 'http.request')", [days]);
const insert = (days) => h.db.query("INSERT INTO telemetry_samples (at, service, operation) VALUES (now() - make_interval(days => $1), 'wiki', 'http.request')", [days]);
- await insert(45);
+ await insert(20);
await insert(1);
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);
assert.strictEqual((await h.db.query('SELECT count(*)::int AS n FROM telemetry_samples')).rows[0].n, 1);
console.log('telemetry ok');
console.log('telemetry ok');
} finally {
} finally {