openvibers/openvibe.billing Commits
Billing: request telemetry in its own table, 30-day retention (#63)
Changes
nnew file mode 100644@@@ -0,0 +1,28 @@++-- phase: expand++-- Request telemetry (openvibe-sdk/telemetry telemetrySchema()): one row per aggregated sample, pruned after 30 days (server/index.js).++CREATE TABLE IF NOT EXISTS telemetry_samples (++ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,++ at timestamptz NOT NULL,++ service text NOT NULL,++ project text,++ subject text,++ resource text,++ provider text,++ node text,++ cell text,++ region text,++ operation text NOT NULL,++ latency_ms double precision,++ queue_delay_ms double precision,++ ttfb_ms double precision,++ throughput_per_second double precision,++ bytes bigint,++ status text,++ cache_status text,++ cost_estimate double precision,++ route_epoch bigint,++ trace_id text,++ extra jsonb NOT NULL DEFAULT '{}'::jsonb++);++CREATE INDEX IF NOT EXISTS telemetry_samples_at_idx ON telemetry_samples (at);++CREATE INDEX IF NOT EXISTS telemetry_samples_service_idx ON telemetry_samples (service, operation, at);@@@ -66,6 +66,7 @@ async function createApp(opts = {}) {const metrics = createMetrics({ db, config, now: ctx.now, outbox });const release = createRelease({ service: 'billing', root: path.join(__dirname, '..') });instrument(app, { service: 'billing', release: release.release, registry: metrics.registry });++ app.use(require('openvibe-sdk/service').telemetryMiddleware); // request telemetry: a no-op until server/index.js calls telemetry.init()// GET /release.json (ADR-016, D43) and POST /release-metrics: what this deployment is, for tabs and deploy checks.release.mount(app, { registry: metrics.registry });app.use(http.middleware());@@@ -19,7 +19,8 @@ const subscriptions = require('./ops/subscriptions');const rating = require('./ops/rating');const { isFrozen } = require('./ops/common');const { runScheduled } = require('./reconcile');--const { gracefulStop } = require('openvibe-sdk/service');++const { gracefulStop, telemetry } = require('openvibe-sdk/service');++const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');(async () => {const config = loadConfig();@@@ -61,6 +62,10 @@ const { gracefulStop } = require('openvibe-sdk/service');console.log(`[Billing] money authority: ${config.authority}${config.authority === 'live' ? ' (shadow: EXTERNAL tips are recorded, not announced)' : ' (EXTERNAL tips are announced as billing.receipt.external)'}`);});server.keepAliveTimeout = 65_000;++ // Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status++ // class each 15 s, stored in this service's telemetry_samples table; rows older than 30 days go daily.++ telemetry.init({ service: 'billing', sink: createPgTelemetrySink({ db: ctx.db, log: console }), skipPrefixes: ['/shared'] });++ const telemetryRetention = startTelemetryRetention({ db: ctx.db, log: console });// SIGTERM/SIGINT (openvibe-sdk/service): the kit installs the handlers, stops the job timers, the outbox// relay and the Network key refresh, drains the HTTP server (requests in flight get Connection: close),@@@ -78,6 +83,7 @@ const { gracefulStop } = require('openvibe-sdk/service');],close: [() => app.locals.valkey && app.locals.valkey.close().catch(() => {}),++ async () => { if (telemetryRetention) await telemetryRetention.stop(); await telemetry.stop(); },() => ctx.db.close(),],});@@@ -45,7 +45,7 @@ async function waitFor(fn, ms = 2000) {console.log('service-kit');await check('server/index.js imports gracefulStop from openvibe-sdk/service and owns no signal handler', () => {-- assert.match(source, /const \{ gracefulStop \} = require\('openvibe-sdk\/service'\)/);++ assert.match(source, /const \{ gracefulStop(, [A-Za-z]+)* \} = require\('openvibe-sdk\/service'\)/);assert.doesNotMatch(source, /process\.on\('SIGTERM'/, 'server/index.js still installs its own SIGTERM handler');assert.doesNotMatch(source, /process\.on\('SIGINT'/, 'server/index.js still installs its own SIGINT handler');assert.doesNotMatch(source, /process\.exit\(0\), 5000\)/, 'the old 5 s exit-0 timer is gone');nnew file mode 100644@@@ -0,0 +1,50 @@++'use strict';++/**++ * Request telemetry (openvibe-sdk/service, plan T1 step 10): app.js mounts the process-wide telemetryMiddleware (a no-op++ * until init), server/index.js calls telemetry.init() for service 'billing' with a PostgreSQL sink, starts the 30-day++ * retention and flushes on stop. Five hits on one route flush as ONE aggregated http.request sample; /metrics and++ * /shared make none; the retention deletes only rows older than 30 days.++ */++const assert = require('assert');++const fs = require('fs');++const path = require('path');++const { telemetry } = require('openvibe-sdk/service');++const { pruneTelemetry } = require('openvibe-sdk/telemetry');++const { boot } = require('./helpers/app');++(async () => {++ const read = (p) => fs.readFileSync(path.join(__dirname, '..', p), 'utf8');++ assert.match(read('server/index.js'), /telemetry\.init\(\{ service: 'billing', sink: createPgTelemetrySink\(/, 'index.js starts the collector');++ assert.match(read('server/index.js'), /startTelemetryRetention\(/, 'index.js starts the retention');++ assert.match(read('server/index.js'), /[Rr]etention\.stop\(\); await telemetry\.stop\(\);/, 'the stop flushes');++++ const t = await boot();++ const samples = [];++ telemetry.init({ service: 'billing', sink: async (batch) => { samples.push(...batch); }, intervalMs: 3600e3 });++ const requests = () => samples.filter((s) => s.operation === 'http.request');++ try {++ for (let i = 0; i < 5; i++) assert.strictEqual((await t.call('GET', '/policy', { token: null })).status, 200);++ await telemetry.flush();++ const hit = requests().filter((s) => s.resource === '/policy' && s.status === '2xx');++ assert.strictEqual(hit.length, 1, JSON.stringify(requests()));++ assert.strictEqual(hit[0].service, 'billing');++ assert.strictEqual(hit[0].extra.count, 5);++++ samples.length = 0;++ await await t.call('GET', '/metrics', { token: null });++ await await t.call('GET', '/shared/navbar.js', { token: null });++ await telemetry.flush();++ assert.deepStrictEqual(requests(), [], '/metrics and /shared make no sample');++++ const db = t.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), 'billing', 'http.request')", [days]);++ await insert(45);++ await insert(1);++ assert.strictEqual(await pruneTelemetry({ db }), 1);++ assert.strictEqual((await db.query('SELECT count(*)::int AS n FROM telemetry_samples')).rows[0].n, 1);++ console.log('telemetry ok');++ } finally {++ await telemetry.stop();++ await t.close();++ }++})().catch((err) => { console.error(err); process.exit(1); });
new file mode 100644
new file mode 100644
@@ -0,0 +1,28 @@
@@ -0,0 +1,28 @@
+-- phase: expand
+-- Request telemetry (openvibe-sdk/telemetry telemetrySchema()): one row per aggregated sample, pruned after 30 days (server/index.js).
+CREATE TABLE IF NOT EXISTS telemetry_samples (
+ id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+ at timestamptz NOT NULL,
+ service text NOT NULL,
+ project text,
+ subject text,
+ resource text,
+ provider text,
+ node text,
+ cell text,
+ region text,
+ operation text NOT NULL,
+ latency_ms double precision,
+ queue_delay_ms double precision,
+ ttfb_ms double precision,
+ throughput_per_second double precision,
+ bytes bigint,
+ status text,
+ cache_status text,
+ cost_estimate double precision,
+ route_epoch bigint,
+ trace_id text,
+ extra jsonb NOT NULL DEFAULT '{}'::jsonb
+);
+CREATE INDEX IF NOT EXISTS telemetry_samples_at_idx ON telemetry_samples (at);
+CREATE INDEX IF NOT EXISTS telemetry_samples_service_idx ON telemetry_samples (service, operation, at);
@@ -66,6 +66,7 @@ async function createApp(opts = {}) {
@@ -66,6 +66,7 @@ async function createApp(opts = {}) {
const metrics = createMetrics({ db, config, now: ctx.now, outbox });
const metrics = createMetrics({ db, config, now: ctx.now, outbox });
const release = createRelease({ service: 'billing', root: path.join(__dirname, '..') });
const release = createRelease({ service: 'billing', root: path.join(__dirname, '..') });
instrument(app, { service: 'billing', release: release.release, registry: metrics.registry });
instrument(app, { service: 'billing', release: release.release, registry: metrics.registry });
+ app.use(require('openvibe-sdk/service').telemetryMiddleware); // request telemetry: a no-op until server/index.js calls telemetry.init()
// GET /release.json (ADR-016, D43) and POST /release-metrics: what this deployment is, for tabs and deploy checks.
// GET /release.json (ADR-016, D43) and POST /release-metrics: what this deployment is, for tabs and deploy checks.
release.mount(app, { registry: metrics.registry });
release.mount(app, { registry: metrics.registry });
app.use(http.middleware());
app.use(http.middleware());
@@ -19,7 +19,8 @@ const subscriptions = require('./ops/subscriptions');
@@ -19,7 +19,8 @@ const subscriptions = require('./ops/subscriptions');
const rating = require('./ops/rating');
const rating = require('./ops/rating');
const { isFrozen } = require('./ops/common');
const { isFrozen } = require('./ops/common');
const { runScheduled } = require('./reconcile');
const { runScheduled } = require('./reconcile');
-const { gracefulStop } = require('openvibe-sdk/service');
+const { gracefulStop, telemetry } = require('openvibe-sdk/service');
+const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');
(async () => {
(async () => {
const config = loadConfig();
const config = loadConfig();
@@ -61,6 +62,10 @@ const { gracefulStop } = require('openvibe-sdk/service');
@@ -61,6 +62,10 @@ const { gracefulStop } = require('openvibe-sdk/service');
console.log(`[Billing] money authority: ${config.authority}${config.authority === 'live' ? ' (shadow: EXTERNAL tips are recorded, not announced)' : ' (EXTERNAL tips are announced as billing.receipt.external)'}`);
console.log(`[Billing] money authority: ${config.authority}${config.authority === 'live' ? ' (shadow: EXTERNAL tips are recorded, not announced)' : ' (EXTERNAL tips are announced as billing.receipt.external)'}`);
});
});
server.keepAliveTimeout = 65_000;
server.keepAliveTimeout = 65_000;
+ // Request telemetry (openvibe-sdk/service, plan T1): one aggregated sample per route template, method and status
+ // class each 15 s, stored in this service's telemetry_samples table; rows older than 30 days go daily.
+ telemetry.init({ service: 'billing', sink: createPgTelemetrySink({ db: ctx.db, log: console }), skipPrefixes: ['/shared'] });
+ const telemetryRetention = startTelemetryRetention({ db: ctx.db, log: console });
// SIGTERM/SIGINT (openvibe-sdk/service): the kit installs the handlers, stops the job timers, the outbox
// SIGTERM/SIGINT (openvibe-sdk/service): the kit installs the handlers, stops the job timers, the outbox
// relay and the Network key refresh, drains the HTTP server (requests in flight get Connection: close),
// relay and the Network key refresh, drains the HTTP server (requests in flight get Connection: close),
@@ -78,6 +83,7 @@ const { gracefulStop } = require('openvibe-sdk/service');
@@ -78,6 +83,7 @@ const { gracefulStop } = require('openvibe-sdk/service');
],
],
close: [
close: [
() => app.locals.valkey && app.locals.valkey.close().catch(() => {}),
() => app.locals.valkey && app.locals.valkey.close().catch(() => {}),
+ async () => { if (telemetryRetention) await telemetryRetention.stop(); await telemetry.stop(); },
() => ctx.db.close(),
() => ctx.db.close(),
],
],
});
});
@@ -45,7 +45,7 @@ async function waitFor(fn, ms = 2000) {
@@ -45,7 +45,7 @@ async function waitFor(fn, ms = 2000) {
console.log('service-kit');
console.log('service-kit');
await check('server/index.js imports gracefulStop from openvibe-sdk/service and owns no signal handler', () => {
await check('server/index.js imports gracefulStop from openvibe-sdk/service and owns no signal handler', () => {
- assert.match(source, /const \{ gracefulStop \} = require\('openvibe-sdk\/service'\)/);
+ assert.match(source, /const \{ gracefulStop(, [A-Za-z]+)* \} = require\('openvibe-sdk\/service'\)/);
assert.doesNotMatch(source, /process\.on\('SIGTERM'/, 'server/index.js still installs its own SIGTERM handler');
assert.doesNotMatch(source, /process\.on\('SIGTERM'/, 'server/index.js still installs its own SIGTERM handler');
assert.doesNotMatch(source, /process\.on\('SIGINT'/, 'server/index.js still installs its own SIGINT handler');
assert.doesNotMatch(source, /process\.on\('SIGINT'/, 'server/index.js still installs its own SIGINT handler');
assert.doesNotMatch(source, /process\.exit\(0\), 5000\)/, 'the old 5 s exit-0 timer is gone');
assert.doesNotMatch(source, /process\.exit\(0\), 5000\)/, 'the old 5 s exit-0 timer is gone');
new file mode 100644
new file mode 100644
@@ -0,0 +1,50 @@
@@ -0,0 +1,50 @@
+'use strict';
+/**
+ * Request telemetry (openvibe-sdk/service, plan T1 step 10): app.js mounts the process-wide telemetryMiddleware (a no-op
+ * until init), server/index.js calls telemetry.init() for service 'billing' with a PostgreSQL sink, starts the 30-day
+ * retention and flushes on stop. Five hits on one route flush as ONE aggregated http.request sample; /metrics and
+ * /shared make none; the retention deletes only rows older than 30 days.
+ */
+const assert = require('assert');
+const fs = require('fs');
+const path = require('path');
+const { telemetry } = require('openvibe-sdk/service');
+const { pruneTelemetry } = require('openvibe-sdk/telemetry');
+const { boot } = require('./helpers/app');
+(async () => {
+ const read = (p) => fs.readFileSync(path.join(__dirname, '..', p), 'utf8');
+ assert.match(read('server/index.js'), /telemetry\.init\(\{ service: 'billing', sink: createPgTelemetrySink\(/, 'index.js starts the collector');
+ assert.match(read('server/index.js'), /startTelemetryRetention\(/, 'index.js starts the retention');
+ assert.match(read('server/index.js'), /[Rr]etention\.stop\(\); await telemetry\.stop\(\);/, 'the stop flushes');
+
+ const t = await boot();
+ const samples = [];
+ telemetry.init({ service: 'billing', sink: async (batch) => { samples.push(...batch); }, intervalMs: 3600e3 });
+ const requests = () => samples.filter((s) => s.operation === 'http.request');
+ try {
+ for (let i = 0; i < 5; i++) assert.strictEqual((await t.call('GET', '/policy', { token: null })).status, 200);
+ await telemetry.flush();
+ const hit = requests().filter((s) => s.resource === '/policy' && s.status === '2xx');
+ assert.strictEqual(hit.length, 1, JSON.stringify(requests()));
+ assert.strictEqual(hit[0].service, 'billing');
+ assert.strictEqual(hit[0].extra.count, 5);
+
+ samples.length = 0;
+ await await t.call('GET', '/metrics', { token: null });
+ await await t.call('GET', '/shared/navbar.js', { token: null });
+ await telemetry.flush();
+ assert.deepStrictEqual(requests(), [], '/metrics and /shared make no sample');
+
+ const db = t.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), 'billing', 'http.request')", [days]);
+ await insert(45);
+ await insert(1);
+ assert.strictEqual(await pruneTelemetry({ db }), 1);
+ assert.strictEqual((await db.query('SELECT count(*)::int AS n FROM telemetry_samples')).rows[0].n, 1);
+ console.log('telemetry ok');
+ } finally {
+ await telemetry.stop();
+ await t.close();
+ }
+})().catch((err) => { console.error(err); process.exit(1); });