openvibers/openvibe.network Commits

Emit Network's telemetry through the openvibe-sdk/service kit

OP OpenVibers committed 14 Oct 2026 · 8ec6309845 · 8 files, +153 −415

Asked: T1 step 5 — OpenVibe.Network: adopt openvibe-sdk/service's telemetry kit and delete Network's hand-rolled copy (the last one in the estate; SDK docs/service.md:164 says Network deletes server/telemetry.js and the telemetry block of server/observability.js). Spec: dossier /home/workstation/openvibe/…
Harness job network-t1-step-5-openvibe-network-002601 (plan T1).

Changes

ddiff --git a/README.md b/README.md
iindex 0515fa1d..ba7cacf5 100644
---- a/README.md
++++ b/README.md
@@@ -530,8 +530,8 @@ Google-style account management supporting up to 5 accounts:
`platform.telemetry-sample@1` per key per flush (count, sum, max and p95 in `extra`, mean in
`latency_ms`), plus the HTTP autoscaling gauges (active requests, p95, event-loop lag) once per flush;
health, readiness, metrics, the OpenVibe Frame and static/shared assets are skipped. Buffered by
-- `openvibe-sdk/telemetry` and written into the analytics store (`server/telemetry.js`,
-- `server/observability.js`); `gracefulStop` flushes it once at shutdown.
++ `openvibe-sdk/service`'s telemetry kit and written into the analytics store as `telemetry.*` events
++ (`server/telemetry-sink.js`); `gracefulStop` flushes it once at shutdown.
- `GET /release.json` is Network's release manifest (ADR-016, `registry.release-manifest@1`, from
`openvibe-shared/release`'s `release.mount`): the deployed commit, the library versions and, since
openvibe-contracts 0.32.0, the 1.1.0 fields. No components are declared, so every release still
ddiff --git a/STATUS.json b/STATUS.json
iindex 3e3bc937..b3fb15e3 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -6,7 +6,7 @@
"runtime": "node 22, express, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9",
"port": 4000,
"domain": "openvibe.network (with auth., api., admin., my. and themes.openvibe.network in its manifest)",
-- "contracts": "openvibe-contracts v0.152.0",
++ "contracts": "openvibe-contracts v0.153.0",
"sdk": "openvibe-sdk v0.44.0",
"shared": "openvibe-shared v4.1.3",
"capabilities": [
ddiff --git a/server/index.js b/server/index.js
iindex 134be007..31f01165 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -339,12 +339,20 @@ config._registry = resolvedRegistry;
const analytics = networkAnalytics.openAnalytics(db);
app.locals.analytics = analytics;
app.use(analytics.middleware());
--// Universal telemetry (plan T1, server/telemetry.js): one platform.telemetry-sample@1 per request plus
--// the HTTP autoscaling signals, into the same analytics store. Wired before any route runs; flushed once
--// by gracefulStop's stop array below.
--const telemetry = require('./telemetry');
--telemetry.init({ analytics });
--app.use(observability.telemetryMiddleware);
++// Universal telemetry (plan T1, openvibe-sdk/service): requests aggregated into
++// platform.telemetry-sample@1 plus the HTTP autoscaling signals, written into the same analytics store by
++// server/telemetry-sink.js. Wired before any route runs; flushed once by gracefulStop's stop array below.
++const svc = require('openvibe-sdk/service');
++const { analyticsSink } = require('./telemetry-sink');
++svc.telemetry.init({
++ service: 'network',
++ sink: analyticsSink(analytics),
++ skipPrefixes: ['/shared', '/api/chrome'],
++ // svc.telemetryMiddleware (sdk 0.44.0) reads only `skipped`, not skipPrefixes: pass the same rule explicitly.
++ skipped: (req) => svc.telemetrySkipped(req, { prefixes: ['/shared', '/api/chrome'] }),
++ routeLabel: require('openvibe-shared/metrics').routeLabel, // the route template, as /metrics labels it
++});
++app.use(svc.telemetryMiddleware);
// ── Extract bearer token (available as req.token for optional-auth proxies) ─
app.use((req, _res, next) => {
@@@ -989,7 +997,7 @@ const server = app.listen(config.port, config.host, () => {
// (unsent rows stay in the outbox), analytics flush and close, the database and the Valkey connection
// close, and the process exits 0, within 10 s (well inside the unit's stop timeout). The email queue
// resumes on the next start.
--const { gracefulStop, within } = require('openvibe-sdk/service');
++const { gracefulStop, within } = svc;
gracefulStop({
name: 'Network', server, drainMs: 8000, deadlineMs: 10000, deadlineExitCode: 1,
stop: [
@@@ -1001,7 +1009,7 @@ gracefulStop({
() => require('./identity/profile-events').stop(),
() => require('./identity/grants-admin').stop(),
() => require('./developer/confirmations').stop(),
-- () => telemetry.stop(),
++ () => svc.telemetry.stop(),
],
close: [
() => within(1500, require('./developer/event-relay').stopRelay(db)),
ddiff --git a/server/observability.js b/server/observability.js
iindex 9a5355b1..858813b7 100644
---- a/server/observability.js
++++ b/server/observability.js
@@@ -9,11 +9,9 @@
* and service-token (principal) failures at Network's capability-guarded internal routes.
* Labels are fixed vocabularies; a client id, subject or token never becomes a label.
*/
--const path = require('path');
const jwt = require('jsonwebtoken');
const metrics = require('openvibe-shared/metrics');
const { createReadiness, skip } = require('openvibe-shared/ready');
--const telemetry = require('./telemetry');
const registry = metrics.createRegistry();
@@@ -52,79 +50,6 @@ function principalDenied({ req, code }) {
principalFailures.inc({ code: c });
}
--// ── Universal telemetry (plan T1, §15.19): aggregated platform.telemetry-sample@1 + autoscaling ──
--// A request is not a row: the middleware hands each finished request to server/telemetry.js, which
--// aggregates per route|method|status_class and emits one http.request sample per key per flush (plus the
--// HTTP autoscaling gauges once per flush). A sample carries the route template, method and status class —
--// never a raw URL, client id or token. Health, readiness, metrics, chrome and static/shared assets carry
--// no product signal and are skipped before anything is counted.
--
--// The event-loop monitor: created and enabled by startEventLoopMonitor (called from telemetry.init), and
--// disabled by stopEventLoopMonitor (telemetry.stop). Nothing runs at module load.
--let loopMonitor = null;
--function startEventLoopMonitor() {
-- if (loopMonitor) return;
-- try {
-- loopMonitor = require('perf_hooks').monitorEventLoopDelay({ resolution: 20 });
-- loopMonitor.enable();
-- } catch { loopMonitor = null; }
--}
--function stopEventLoopMonitor() {
-- if (!loopMonitor) return;
-- try { loopMonitor.disable(); } catch { /* already gone */ }
-- loopMonitor = null;
--}
--/** Whether the monitor is running (tests assert requiring the module starts nothing). */
--function eventLoopMonitorEnabled() { return loopMonitor != null; }
--
--/** The event-loop lag since the last flush, ms (mean includes the sampling interval: subtracted); null
-- * before the monitor has a sample. The monitor resets per read, so each gauge covers one flush interval. */
--function eventLoopLagMs() {
-- if (!loopMonitor || !(loopMonitor.count > 0 || loopMonitor.max > 0)) return null;
-- const lag = Math.max(0, loopMonitor.mean / 1e6 - 20);
-- loopMonitor.reset();
-- return lag;
--}
--
--// server/telemetry.js owns the aggregation; it pulls the lag and calls these lifecycle hooks.
--telemetry.registerSignals({ start: startEventLoopMonitor, stop: stopEventLoopMonitor, lag: eventLoopLagMs });
--
--const SKIP_EXACT = new Set(['/api/health', '/ready', '/api/ready', '/metrics']);
--const SKIP_PREFIXES = ['/shared', '/api/chrome'];
--// Anything express.static serves: public/ files and /data/avatars images. JSON is deliberately not here
--// (routes such as /release.json and /contracts/*.json are product/observability API, not static assets).
--const STATIC_EXTENSIONS = new Set([
-- '.js', '.mjs', '.cjs', '.css', '.map', '.html', '.htm', '.txt', '.xml', '.webmanifest', '.wasm',
-- '.png', '.jpg', '.jpeg', '.gif', '.svg', '.webp', '.avif', '.ico', '.bmp',
-- '.woff', '.woff2', '.ttf', '.otf', '.eot', '.mp3', '.mp4', '.webm', '.ogg', '.pdf',
--]);
--function telemetrySkipped(req) {
-- const p = req.path || '';
-- if (SKIP_EXACT.has(p)) return true;
-- if (SKIP_PREFIXES.some((prefix) => p === prefix || p.startsWith(`${prefix}/`))) return true;
-- return STATIC_EXTENSIONS.has(path.extname(p).toLowerCase());
--}
--
--/** Express middleware: hand the finished request to the aggregator; the in-flight peak is sampled at start. */
--function telemetryMiddleware(req, res, next) {
-- if (telemetrySkipped(req)) return next();
-- telemetry.requestStarted();
-- const t0 = process.hrtime.bigint();
-- let done = false;
-- const finish = () => {
-- if (done) return;
-- done = true;
-- try {
-- const latencyMs = Number(process.hrtime.bigint() - t0) / 1e6;
-- const httpStatus = res.writableFinished || res.finished ? (res.statusCode || 0) : 0;
-- telemetry.requestFinished({ route: metrics.routeLabel(req), method: req.method, httpStatus, latencyMs });
-- } catch { /* telemetry never breaks a request */ }
-- };
-- res.once('finish', finish);
-- res.once('close', finish);
-- next();
--}
--
/**
* Readiness with Network's real dependencies:
* db (required) a query against the identity database
@@@ -173,4 +98,4 @@ function createNetworkReadiness({ db, getKeys, release, ecosystem = null, discor
return createReadiness({ service: 'network', release, checks });
}
--module.exports = { registry, tokenEndpointMetrics, principalDenied, telemetryMiddleware, telemetrySkipped, createNetworkReadiness, grantLabel, eventLoopMonitorEnabled };
++module.exports = { registry, tokenEndpointMetrics, principalDenied, createNetworkReadiness, grantLabel };
ddiff --git a/server/telemetry-sink.js b/server/telemetry-sink.js
nnew file mode 100644
iindex 00000000..4d29200a
---- /dev/null
++++ b/server/telemetry-sink.js
@@@ -0,0 +1,39 @@
++'use strict';
++/**
++ * Network's telemetry sink (openvibe-sdk/service telemetry, plan T1 §15.19). The kit aggregates requests
++ * and emits platform.telemetry-sample@1; this is the one Network-specific piece: each flushed batch goes
++ * into Network's ADR-021 analytics tracker as custom `telemetry.<operation>` events (the dashboard reads
++ * them), then the tracker is flushed so one batch lands together. Wired in server/index.js.
++ */
++
++// analytics/event.v1 event_type: [A-Za-z][A-Za-z0-9_.:-]{0,63}; an operation is sanitised to fit.
++const EVENT_RE = /^[A-Za-z][A-Za-z0-9_.:-]{0,63}$/;
++function eventName(operation) {
++ const name = `telemetry.${String(operation == null ? 'operation' : operation)}`.replace(/[^A-Za-z0-9_.:-]/g, '_').slice(0, 64);
++ return EVENT_RE.test(name) ? name : 'telemetry.operation';
++}
++
++/**
++ * The sink: write each sample in the batch into Network's analytics tracker (its `trackEvent`, the
++ * non-HTTP form; ADR-021 reduces the personal fields), then await its flush so the batch lands together.
++ * Only the fields analytics/event.v1 can hold are passed; the numbers live in `extra.http_status`.
++ */
++function analyticsSink(analytics) {
++ if (!analytics || typeof analytics.trackEvent !== 'function') {
++ throw new TypeError('analyticsSink: Network\'s analytics tracker (trackEvent) is required');
++ }
++ return async function sink(samples) {
++ for (const s of samples) {
++ const extra = s.extra || {};
++ analytics.trackEvent(eventName(s.operation), {
++ path: typeof s.resource === 'string' ? s.resource : (typeof extra.route === 'string' ? extra.route : undefined),
++ method: typeof extra.method === 'string' ? extra.method : undefined,
++ status_code: Number.isInteger(extra.http_status) ? extra.http_status : undefined,
++ response_time_ms: Number.isFinite(s.latency_ms) ? s.latency_ms : undefined,
++ });
++ }
++ if (typeof analytics.flush === 'function') await analytics.flush();
++ };
++}
++
++module.exports = { eventName, analyticsSink };
ddiff --git a/server/telemetry.js b/server/telemetry.js
ddeleted file mode 100644
iindex 89852542..00000000
---- a/server/telemetry.js
++++ /dev/null
@@@ -1,226 +0,0 @@
--'use strict';
--/**
-- * Network's platform.telemetry-sample@1 collector (openvibe-sdk/telemetry; plan T1 universal Fabric,
-- * §15.19). One sample schema every service emits; autoscaling and routing consume it.
-- *
-- * createNetworkTelemetry({ analytics }) service 'network', with a sink that writes each flushed batch
-- * into Network's ADR-021 analytics tracker as custom (non-HTTP)
-- * events and flushes it, so one batch lands together.
-- * init({ analytics }) the process-wide collector, wired once at boot (server/index.js).
-- * record / gauge / count / flush / stop the collector's API; before init they are no-ops.
-- *
-- * Volume: one analytics row per request does not scale (at 50 req/s that is ~200 rows/s). HTTP requests
-- * are therefore aggregated per `route|method|status_class` per flush interval into count, sum, max and
-- * p95, and ONE `http.request` sample per key is emitted per flush — not one per request. The HTTP
-- * autoscaling gauges (active requests, p95, event-loop lag) are emitted once per flush from a timer
-- * started in init(), never per request, so the SDK buffer (maxBuffered 1000) cannot overflow.
-- *
-- * Field mapping (platform.telemetry-sample@1; `latency_ms` is the schema's latency field, there is no
-- * generic gauge field): an aggregated request carries its mean in `latency_ms` and count/sum/max/p95 in
-- * `extra`; the p95 gauge carries its value in `latency_ms` (it is a latency); the active-requests and
-- * event-loop-lag gauges carry their value in `extra` — never in `latency_ms`, which the analytics sink
-- * maps to `response_time_ms` and would make indistinguishable from a latency.
-- *
-- * server/observability.js feeds the middleware's per-request observations and registers the event-loop
-- * monitor (started by init, stopped by stop) so nothing starts at module load; gracefulStop's stop array
-- * flushes it once at shutdown.
-- */
--const { createTelemetry, telemetrySample, validateTelemetrySample } = require('openvibe-sdk/telemetry');
--
--const SERVICE = 'network';
--const DEFAULT_INTERVAL_MS = 15000;
--
--// analytics/event.v1 event_type: [A-Za-z][A-Za-z0-9_.:-]{0,63}; an operation is sanitised to fit.
--const EVENT_RE = /^[A-Za-z][A-Za-z0-9_.:-]{0,63}$/;
--function eventName(operation) {
-- const name = `telemetry.${String(operation == null ? 'operation' : operation)}`.replace(/[^A-Za-z0-9_.:-]/g, '_').slice(0, 64);
-- return EVENT_RE.test(name) ? name : 'telemetry.operation';
--}
--
--/**
-- * The sink: write each sample in the batch into Network's analytics tracker (its `trackEvent`, the
-- * non-HTTP form; ADR-021 reduces the personal fields), then await its flush so the batch lands together.
-- * Only the fields analytics/event.v1 can hold are passed; the numbers live in `extra.http_status`.
-- */
--function analyticsSink(analytics) {
-- if (!analytics || typeof analytics.trackEvent !== 'function') {
-- throw new TypeError('analyticsSink: Network\'s analytics tracker (trackEvent) is required');
-- }
-- return async function sink(samples) {
-- for (const s of samples) {
-- const extra = s.extra || {};
-- analytics.trackEvent(eventName(s.operation), {
-- path: typeof s.resource === 'string' ? s.resource : (typeof extra.route === 'string' ? extra.route : undefined),
-- method: typeof extra.method === 'string' ? extra.method : undefined,
-- status_code: Number.isInteger(extra.http_status) ? extra.http_status : undefined,
-- response_time_ms: Number.isFinite(s.latency_ms) ? s.latency_ms : undefined,
-- });
-- }
-- if (typeof analytics.flush === 'function') await analytics.flush();
-- };
--}
--
--// ── HTTP aggregation (volume) ─────────────────────────────────────
--const HTTP_METHODS = new Set(['GET', 'HEAD', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS']);
--const statusClass = (code) => (code ? `${Math.floor(code / 100)}xx` : 'aborted');
--const round = (n, digits = 3) => { const f = 10 ** digits; return Math.round(n * f) / f; };
--
--/** p95 (nearest-rank) of an array of ms latencies; null when empty. Computed at most once per flush. */
--function percentile95(values) {
-- if (!values.length) return null;
-- const sorted = [...values].sort((a, b) => a - b);
-- return sorted[Math.min(sorted.length - 1, Math.floor(sorted.length * 0.95))];
--}
--
--// The event-loop monitor lives in observability.js and is injected here at its require: requiring either
--// module starts nothing. telemetry.init() calls start(), telemetry.stop() calls stop(), and the gauges
--// read lag(). Unregistered (unit tests, a bare collector) they are safe no-ops.
--let signals = { start() {}, stop() {}, lag() { return null; } };
--function registerSignals(injected) {
-- signals = { ...signals, ...(injected || {}) };
--}
--
--/**
-- * Build one collector. `sink` overrides the analytics sink (tests inject one); everything else is the SDK.
-- * HTTP observations are aggregated in memory and drained on flush(); the flush timer itself starts in
-- * init(), so a bare collector only flushes when asked.
-- */
--function createNetworkTelemetry({ analytics = null, sink = null, instance = null, intervalMs = DEFAULT_INTERVAL_MS, now, log, maxBuffered } = {}) {
-- const write = typeof sink === 'function' ? sink : analyticsSink(analytics);
-- const sdk = createTelemetry({ service: SERVICE, instance, sink: write, intervalMs, now, log, maxBuffered });
--
-- let routes = new Map(); // key -> { route, method, status, count, sumMs, maxMs, latencies, codes }
-- let allLatencies = []; // the flush interval's latencies, for the single p95 gauge
-- let active = 0; // requests in flight right now
-- let activeMax = 0; // the interval's peak, sampled at request start (item 3)
-- let stopped = false;
--
-- /** A request began: track the in-flight peak here, not after the decrement. */
-- function requestStarted() {
-- active += 1;
-- if (active > activeMax) activeMax = active;
-- }
--
-- function observeRequest(info = {}) {
-- const r = String(info.route == null ? 'unmatched' : info.route).slice(0, 200);
-- const m = HTTP_METHODS.has(info.method) ? info.method : 'OTHER';
-- const s = statusClass(info.httpStatus);
-- const latency = Number.isFinite(info.latencyMs) && info.latencyMs >= 0 ? info.latencyMs : 0;
-- const key = `${r}|${m}|${s}`;
-- let bucket = routes.get(key);
-- if (!bucket) {
-- bucket = { route: r, method: m, status: s, count: 0, sumMs: 0, maxMs: 0, latencies: [], codes: new Set() };
-- routes.set(key, bucket);
-- }
-- bucket.count += 1;
-- bucket.sumMs += latency;
-- if (latency > bucket.maxMs) bucket.maxMs = latency;
-- bucket.latencies.push(latency);
-- if (Number.isInteger(info.httpStatus) && info.httpStatus > 0) bucket.codes.add(info.httpStatus);
-- allLatencies.push(latency);
-- }
--
-- /** A request ended: free the in-flight slot, then fold it into its aggregation key. */
-- function requestFinished(info) {
-- if (active > 0) active -= 1;
-- observeRequest(info);
-- }
--
-- /** Push a gauge whose value(s) live in `extra`; the schema has no generic gauge field and latency_ms
-- * is the sink's response_time_ms, so extra is the documented home (never latency_ms). */
-- function emitExtraGauge(name, extra) {
-- const sample = sdk.record(name, 0, { extra });
-- delete sample.latency_ms;
-- return sample;
-- }
--
-- /** Drain this interval: one http.request sample per route|method|status_class, then the three gauges. */
-- function emitSamples() {
-- const buckets = [...routes.values()];
-- const latencies = allLatencies;
-- routes = new Map();
-- allLatencies = [];
-- for (const b of buckets) {
-- const mean = b.count ? b.sumMs / b.count : 0;
-- const p95 = percentile95(b.latencies);
-- const extra = {
-- method: b.method,
-- status_class: b.status,
-- count: b.count,
-- sum_ms: round(b.sumMs),
-- max_ms: round(b.maxMs),
-- };
-- if (p95 != null) extra.p95_ms = round(p95);
-- if (b.codes.size === 1) extra.http_status = b.codes.values().next().value;
-- sdk.record('http.request', round(mean), { status: b.status, resource: b.route, extra });
-- }
-- // The gauges, once per flush (never per request).
-- emitExtraGauge('http.active_requests', { last: active, max: activeMax });
-- activeMax = active; // the next interval's peak starts from what is still in flight
-- const p95 = percentile95(latencies);
-- if (p95 != null) sdk.record('http.latency_p95_ms', round(p95)); // latency_ms: the schema's latency field
-- const lag = signals.lag();
-- if (lag != null) emitExtraGauge('http.eventloop_lag_ms', { value: round(lag) });
-- }
--
-- /** Emit the interval's samples, then hand the SDK everything buffered (a flush is also the tick). */
-- function flush() {
-- if (stopped) return Promise.resolve();
-- emitSamples();
-- return sdk.flush();
-- }
--
-- /** Emit once, then let the SDK clear its timer and flush (idempotent). */
-- async function stop() {
-- if (stopped) return;
-- stopped = true;
-- emitSamples();
-- await sdk.stop();
-- }
--
-- return { record: sdk.record, gauge: sdk.gauge, count: sdk.count, flush, stop, requestStarted, requestFinished, observeRequest };
--}
--
--// The process-wide collector. It exists only once the analytics tracker does (server/index.js), so an
--// early request is recorded by neither; the HTTP middleware is mounted before any route runs.
--let current = null;
--let timer = null;
--function init(opts = {}) {
-- current = createNetworkTelemetry(opts);
-- signals.start(); // the event-loop monitor (observability.js): created and enabled here, not at load
-- if (timer) clearInterval(timer);
-- const interval = Number.isFinite(opts.intervalMs) ? opts.intervalMs : DEFAULT_INTERVAL_MS;
-- timer = setInterval(() => { current.flush(); }, interval);
-- if (timer.unref) timer.unref();
-- return current;
--}
--
--function record(...args) { return current ? current.record(...args) : undefined; }
--function gauge(...args) { return current ? current.gauge(...args) : undefined; }
--function count(...args) { return current ? current.count(...args) : undefined; }
--function requestStarted() { return current ? current.requestStarted() : undefined; }
--function requestFinished(info) { return current ? current.requestFinished(info) : undefined; }
--function flush() { return current ? current.flush() : Promise.resolve(); }
--async function stop() {
-- if (timer) { clearInterval(timer); timer = null; }
-- if (current) await current.stop();
-- signals.stop(); // disable the event-loop monitor that init enabled
--}
--
--module.exports = {
-- SERVICE,
-- eventName,
-- analyticsSink,
-- createNetworkTelemetry,
-- telemetrySample,
-- validateTelemetrySample,
-- registerSignals,
-- init,
-- record,
-- gauge,
-- count,
-- requestStarted,
-- requestFinished,
-- flush,
-- stop,
--};
ddiff --git a/test/service-kit.test.js b/test/service-kit.test.js
iindex 4e2bf9c5..b1a47093 100644
---- a/test/service-kit.test.js
++++ b/test/service-kit.test.js
@@@ -14,7 +14,7 @@ const root = path.join(__dirname, '..');
const index = fs.readFileSync(path.join(root, 'server', 'index.js'), 'utf8');
// Static: the wiring is the SDK's, with today's Network values.
--assert.ok(/const \{ gracefulStop, within \} = require\('openvibe-sdk\/service'\)/.test(index), "server/index.js imports gracefulStop and within from openvibe-sdk/service");
++assert.ok(/const svc = require\('openvibe-sdk\/service'\)/.test(index) && /const \{ gracefulStop, within \} = svc;/.test(index), "server/index.js imports gracefulStop and within from openvibe-sdk/service");
assert.ok(!/require\(['"]\.\/graceful['"]\)/.test(index), 'server/index.js no longer requires ./graceful');
assert.ok(!fs.existsSync(path.join(root, 'server', 'graceful' + '.js')), 'the local shutdown copy is deleted');
assert.ok(/name: 'Network', server, drainMs: 8000, deadlineMs: 10000, deadlineExitCode: 1/.test(index), 'drainMs 8000, deadlineMs 10000, deadlineExitCode 1');
@@@ -22,7 +22,7 @@ assert.ok(/name: 'Network', server, drainMs: 8000, deadlineMs: 10000, deadlineEx
const stopAt = index.indexOf('stop: [');
const closeAt = index.indexOf('close: [');
assert.ok(stopAt > 0 && closeAt > stopAt, 'the stop list precedes the close list');
--for (const marker of ['ecosystem.stop()', 'deployDrift.stop()', 'libraryTags.stop()', 'frameService.stop()']) {
++for (const marker of ['ecosystem.stop()', 'deployDrift.stop()', 'libraryTags.stop()', 'frameService.stop()', 'svc.telemetry.stop()']) {
assert.ok(index.indexOf(marker) > stopAt && index.indexOf(marker) < closeAt, `stop step ${marker} is in the stop list`);
}
for (const marker of ["require('./developer/event-relay').stopRelay(db)", 'analytics.destroy()', 'db.close()', 'valkey && valkey.close()']) {
ddiff --git a/test/telemetry.test.js b/test/telemetry.test.js
iindex fb8b623b..5c897afa 100644
---- a/test/telemetry.test.js
++++ b/test/telemetry.test.js
@@@ -1,146 +1,138 @@
'use strict';
--// Universal telemetry (server/telemetry.js on openvibe-sdk/telemetry, plan T1): every emitted
--// platform.telemetry-sample@1 validates, HTTP requests are aggregated per route|method|status_class
--// (one sample for each such group at each flush, with count, sum, max and p95 in extra), skipped paths produce nothing, the three
--// autoscaling gauges are emitted once per flush, one flush posts one batch, stop() flushes once and is
--// idempotent, the analytics sink maps a batch to custom events, and nothing starts at module load.
--// No database: the sink is injected.
++// Universal telemetry (plan T1) on openvibe-sdk/service's telemetry kit: Network's hand-rolled copy is
++// gone (server/index.js wires the kit, server/observability.js exports none of the old telemetry names),
++// HTTP requests are aggregated per route|method|status_class (one http.request sample per group per flush),
++// stop() flushes once and is idempotent, Network's analytics sink (server/telemetry-sink.js) writes each
++// sample as a telemetry.<operation> event and flushes the tracker once per batch, and the middleware wired
++// with Network's options skips probes, chrome and static/shared assets. No database: the tracker is a fake.
// node --test test/telemetry.test.js
const assert = require('assert');
++const fs = require('fs');
++const path = require('path');
const http = require('http');
const express = require('express');
--const telemetry = require('../server/telemetry');
--const { createNetworkTelemetry, analyticsSink, validateTelemetrySample } = require('../server/telemetry');
++const svc = require('openvibe-sdk/service');
++const metrics = require('openvibe-shared/metrics');
++const { analyticsSink, eventName } = require('../server/telemetry-sink');
const observability = require('../server/observability');
const quiet = { log() {}, warn() {}, error() {} };
const AT = '2026-09-30T10:00:00.000Z';
const valid = (sample, label) => {
-- const r = validateTelemetrySample(sample);
++ const r = svc.validateTelemetrySample(sample);
assert.ok(r.ok, `${label}: ${JSON.stringify(r.errors)} — ${JSON.stringify(sample)}`);
};
++/** A stand-in for Network's ADR-021 tracker: records trackEvent calls and counts flushes. */
++function fakeAnalytics() {
++ const a = { events: [], flushes: 0 };
++ a.trackEvent = (name, data) => { a.events.push({ name, data }); };
++ a.flush = () => { a.flushes += 1; return Promise.resolve(); };
++ return a;
++}
(async () => {
-- // ── Item 2: requiring the modules starts nothing (no monitor, no timer) ──
-- assert.strictEqual(observability.eventLoopMonitorEnabled(), false, 'requiring observability starts no event-loop monitor');
--
-- // ── Every emitted sample is a well-formed platform.telemetry-sample@1 ──
-- const batches = [];
-- const t = createNetworkTelemetry({ sink: async (samples) => { batches.push(samples); }, intervalMs: 3600e3, now: () => Date.parse(AT), log: quiet });
-- const request = t.record('http.request', 12.5, { status: '2xx', resource: '/api/v1/x', extra: { method: 'GET', http_status: 200 } });
-- const gauge = t.gauge('demo.queue_depth', 3);
-- const count = t.count('cache.miss');
-- for (const [sample, label] of [[request, 'record'], [gauge, 'gauge'], [count, 'count']]) {
-- valid(sample, label);
-- assert.strictEqual(sample.service, 'network', `${label} carries the service`);
-- assert.strictEqual(sample.at, AT, `${label} carries at`);
++ // ── Static: the kit replaces Network's copy ──
++ const root = path.join(__dirname, '..');
++ const index = fs.readFileSync(path.join(root, 'server/index.js'), 'utf8');
++ assert.ok(index.includes("require('openvibe-sdk/service')"), 'server/index.js requires the kit');
++ assert.ok(index.includes('svc.telemetry.init('), 'server/index.js initialises the kit\'s collector');
++ assert.ok(index.includes('app.use(svc.telemetryMiddleware)'), 'server/index.js mounts the kit\'s middleware');
++ assert.ok(index.includes('svc.telemetry.stop()'), 'gracefulStop stops the kit\'s collector');
++ assert.ok(index.includes("svc.telemetrySkipped(req, { prefixes: ['/shared', '/api/chrome'] })"), 'the middleware skips /shared and /api/chrome');
++ assert.ok(!/require\(['"]\.\/telemetry['"]\)/.test(index), 'server/index.js no longer requires ./telemetry');
++ assert.ok(!fs.existsSync(path.join(root, 'server/telemetry.js')), 'server/telemetry.js is deleted');
++ for (const name of ['telemetryMiddleware', 'telemetrySkipped', 'eventLoopMonitorEnabled']) {
++ assert.ok(!(name in observability), `observability.js no longer exports ${name}`);
}
++ assert.strictEqual(svc.eventLoopMonitorEnabled(), false, 'requiring the modules starts no event-loop monitor');
-- // ── A flush posts exactly one well-formed batch, drains the interval's gauges, stop() is idempotent ──
-- await t.flush();
-- assert.strictEqual(batches.length, 1, 'one flush = one sink call');
-- assert.deepStrictEqual(batches[0].slice(0, 3), [request, gauge, count], 'the buffered samples, in order');
-- const active0 = batches[0].find((s) => s.operation === 'http.active_requests');
-- assert.ok(active0, 'the active-requests gauge is emitted once per flush');
-- assert.deepStrictEqual(active0.extra, { last: 0, max: 0 }, 'the gauge value lives in extra, not latency_ms');
-- assert.ok(!('latency_ms' in active0), 'a gauge is never in the latency field');
-- for (const s of batches[0]) valid(s, 'flushed batch');
-- await t.flush();
-- assert.strictEqual(batches.length, 2, 'a second flush emits the gauges again (once per flush)');
-- assert.strictEqual(batches[1].length, 1);
-- assert.strictEqual(batches[1][0].operation, 'http.active_requests');
-- t.record('http.request', 7, { status: '5xx' });
-- await t.stop();
-- assert.strictEqual(batches.length, 3, 'stop() flushes the remaining buffer once');
-- assert.strictEqual(batches[2].length, 2);
-- await t.stop();
-- await t.flush();
-- assert.strictEqual(batches.length, 3, 'a second stop() (or a later flush) posts nothing');
++ // ── 5 requests to one route = ONE http.request sample; the sink writes telemetry.<op> events ──
++ const analytics = fakeAnalytics();
++ const samples = [];
++ const sink = analyticsSink(analytics);
++ const c = svc.createHttpTelemetry({
++ service: 'network', intervalMs: 3600e3, now: () => Date.parse(AT), log: quiet,
++ sink: async (batch) => { samples.push(...batch); await sink(batch); },
++ });
++ for (let i = 0; i < 5; i++) c.requestStarted();
++ for (let i = 0; i < 5; i++) c.requestFinished({ route: '/api/v1/things/:id', method: 'GET', httpStatus: 200, latencyMs: 10 + i });
++ await c.flush();
++ assert.strictEqual(analytics.flushes, 1, 'one flush = one tracker flush');
++ const requests = samples.filter((s) => s.operation === 'http.request');
++ assert.strictEqual(requests.length, 1, `five requests to one route = one sample: ${JSON.stringify(samples)}`);
++ assert.strictEqual(requests[0].service, 'network');
++ assert.strictEqual(requests[0].resource, '/api/v1/things/:id', 'the route template, never the raw id');
++ assert.strictEqual(requests[0].latency_ms, 12, 'the mean latency is the sample\'s latency_ms');
++ assert.deepStrictEqual(requests[0].extra, { method: 'GET', status_class: '2xx', count: 5, sum_ms: 60, max_ms: 14, p95_ms: 14, http_status: 200 });
++ assert.strictEqual(samples.filter((s) => s.operation === 'http.active_requests').length, 1, 'the active gauge once per flush');
++ for (const s of samples) valid(s, 'aggregated batch');
++ const reqEvents = analytics.events.filter((e) => e.name === 'telemetry.http.request');
++ assert.strictEqual(reqEvents.length, 1);
++ assert.deepStrictEqual(reqEvents[0].data, { path: '/api/v1/things/:id', method: 'GET', status_code: 200, response_time_ms: 12 },
++ 'the sink maps a sample to path/method/status_code/response_time_ms');
++ assert.strictEqual(analytics.events.length, samples.length, 'one event per sample');
++ assert.ok(analytics.events.every((e) => e.name.startsWith('telemetry.')), 'every event is telemetry.<operation>');
-- // ── Item 1: N requests to one route aggregate into ONE http.request sample with count N ──
-- const aggBatches = [];
-- const a = createNetworkTelemetry({ sink: async (s) => { aggBatches.push(s); }, intervalMs: 3600e3, now: () => Date.parse(AT), log: quiet });
-- for (let i = 0; i < 5; i++) a.requestStarted(); // all in flight at once: the peak is 5
-- for (let i = 0; i < 5; i++) a.requestFinished({ route: '/api/v1/things/:id', method: 'GET', httpStatus: 200, latencyMs: 10 + i });
-- await a.flush();
-- const httpSamples = aggBatches[0].filter((s) => s.operation === 'http.request');
-- assert.strictEqual(httpSamples.length, 1, `five requests to one route = one sample: ${JSON.stringify(aggBatches[0])}`);
-- const agg = httpSamples[0];
-- assert.strictEqual(agg.resource, '/api/v1/things/:id', 'the route template, never the raw id');
-- assert.strictEqual(agg.status, '2xx');
-- assert.strictEqual(agg.latency_ms, 12, 'the mean latency is the sample’s latency_ms');
-- assert.deepStrictEqual(agg.extra, { method: 'GET', status_class: '2xx', count: 5, sum_ms: 60, max_ms: 14, p95_ms: 14, http_status: 200 },
-- 'count, sum, max and p95 live in extra');
-- for (const s of aggBatches[0]) valid(s, 'aggregated batch');
-- const activeAgg = aggBatches[0].filter((s) => s.operation === 'http.active_requests');
-- assert.strictEqual(activeAgg.length, 1, 'the active gauge is emitted once per flush');
-- assert.deepStrictEqual(activeAgg[0].extra, { last: 0, max: 5 }, 'item 3: the peak is sampled at request start');
-- assert.strictEqual(aggBatches[0].filter((s) => s.operation === 'http.latency_p95_ms').length, 1, 'p95 is emitted once per flush');
-- assert.strictEqual(aggBatches[0].find((s) => s.operation === 'http.latency_p95_ms').latency_ms, 14);
-- await a.flush();
-- assert.strictEqual(aggBatches[1].filter((s) => s.operation === 'http.request').length, 0, 'an idle interval emits no request sample');
-- assert.strictEqual(aggBatches[1].filter((s) => s.operation === 'http.active_requests').length, 1);
-- assert.strictEqual(aggBatches[1].filter((s) => s.operation === 'http.latency_p95_ms').length, 0, 'no requests: no p95');
-- await a.stop();
++ // ── stop() flushes once; a second stop() (or a later flush) changes nothing ──
++ await c.stop();
++ assert.strictEqual(analytics.flushes, 2, 'stop() flushes once');
++ const after = analytics.events.length;
++ await c.stop();
++ await c.flush();
++ assert.strictEqual(analytics.flushes, 2, 'a second stop() is a no-op');
++ assert.strictEqual(analytics.events.length, after);
-- // ── The analytics sink writes the batch as custom events and flushes once ──
-- const events = [];
-- const analytics = { trackEvent: (name, data) => events.push({ name, data }), flush() { analytics.flushes = (analytics.flushes || 0) + 1; return Promise.resolve(); } };
-- await analyticsSink(analytics)([
++ // ── The sink on its own: the event mapping Network's ADR-021 dashboard reads ──
++ const direct = fakeAnalytics();
++ await analyticsSink(direct)([
{ service: 'network', operation: 'http.request', at: AT, latency_ms: 12.5, status: '2xx', resource: '/api/v1/x', extra: { method: 'GET', http_status: 200 } },
{ service: 'network', operation: 'a b/c:d', at: AT, latency_ms: 1 },
]);
-- assert.strictEqual(analytics.flushes, 1, 'the sink flushes the tracker once per batch');
-- assert.strictEqual(events.length, 2);
-- assert.match(events[0].name, /^telemetry\.http\.request$/);
-- assert.deepStrictEqual(events[0].data, { path: '/api/v1/x', method: 'GET', status_code: 200, response_time_ms: 12.5 });
-- assert.strictEqual(events[1].name, 'telemetry.a_b_c:d', 'spaces and slashes are sanitised; : is valid in analytics/event.v1');
++ assert.strictEqual(direct.flushes, 1, 'the sink flushes the tracker once per batch');
++ assert.deepStrictEqual(direct.events[0], { name: 'telemetry.http.request', data: { path: '/api/v1/x', method: 'GET', status_code: 200, response_time_ms: 12.5 } });
++ assert.strictEqual(direct.events[1].name, 'telemetry.a_b_c:d', 'spaces and slashes are sanitised; : is valid in analytics/event.v1');
++ assert.strictEqual(eventName(null), 'telemetry.operation');
++ assert.throws(() => analyticsSink({}), TypeError, 'the sink needs a tracker');
-- // ── Items 1 & 2: the middleware aggregates; skipped paths are never counted; init starts the monitor ──
++ // ── The singleton with Network's options: the middleware aggregates and skips probes/chrome/assets ──
const raw = [];
-- telemetry.init({ sink: async (samples) => { for (const s of samples) raw.push(s); }, intervalMs: 3600e3, log: quiet });
-- assert.strictEqual(observability.eventLoopMonitorEnabled(), true, 'init() creates and enables the event-loop monitor');
-- await new Promise((r) => setTimeout(r, 60)); // let the monitor take a sample before the flush
++ svc.telemetry.init({
++ service: 'network', intervalMs: 3600e3, log: quiet,
++ sink: async (batch) => { raw.push(...batch); },
++ skipPrefixes: ['/shared', '/api/chrome'],
++ skipped: (req) => svc.telemetrySkipped(req, { prefixes: ['/shared', '/api/chrome'] }), // as server/index.js
++ routeLabel: metrics.routeLabel,
++ });
++ assert.strictEqual(svc.eventLoopMonitorEnabled(), true, 'init() enables the event-loop monitor');
const app = express();
-- app.use(observability.telemetryMiddleware);
-- app.get('/api/v1/things/:id', (req, res) => res.json({ ok: true }));
-- app.get('/api/health', (_req, res) => res.json({ ok: true }));
-- app.get('/metrics', (_req, res) => res.type('text/plain').send('ok'));
-- app.get('/api/ready', (_req, res) => res.json({ ok: true }));
-- app.get('/api/chrome', (_req, res) => res.json({ ok: true }));
++ app.use(svc.telemetryMiddleware);
++ app.get('/api/v1/things/:id', (_req, res) => res.json({ ok: true }));
++ for (const p of ['/api/health', '/metrics', '/api/ready', '/api/chrome', '/release.json']) app.get(p, (_req, res) => res.json({ ok: true }));
app.get('/shared/navbar.js', (_req, res) => res.type('application/javascript').send('ok'));
app.get('/logo.png', (_req, res) => res.type('image/png').send('ok'));
-- app.get('/data/avatars/a.png', (_req, res) => res.type('image/png').send('ok'));
-- app.get('/release.json', (_req, res) => res.json({ ok: true }));
const server = http.createServer(app);
await new Promise((r) => server.listen(0, '127.0.0.1', r));
const base = `http://127.0.0.1:${server.address().port}`;
await fetch(`${base}/api/v1/things/42`);
await fetch(`${base}/api/v1/things/43`);
-- for (const p of ['/api/health', '/metrics', '/api/ready', '/api/chrome', '/shared/navbar.js', '/logo.png', '/data/avatars/a.png']) await fetch(base + p);
++ for (const p of ['/api/health', '/metrics', '/api/ready', '/api/chrome', '/shared/navbar.js', '/logo.png']) await fetch(base + p);
await fetch(`${base}/release.json`); // a JSON route is product API, not a static asset: still counted
-- await telemetry.flush();
++ await svc.telemetry.flush();
server.close();
const sampled = raw.filter((s) => s.operation === 'http.request');
const things = sampled.filter((s) => s.resource === '/api/v1/things/:id');
assert.strictEqual(things.length, 1, `two requests to one route = one sample: ${JSON.stringify(sampled)}`);
assert.strictEqual(things[0].extra.count, 2);
-- assert.strictEqual(things[0].status, '2xx');
const resources = new Set(sampled.map((s) => s.resource));
-- for (const skipped of ['/api/health', '/metrics', '/api/ready', '/api/chrome', '/shared/*', '/logo.png', '/data/avatars/a.png']) {
++ for (const skipped of ['/api/health', '/metrics', '/api/ready', '/api/chrome', '/shared/navbar.js', '/logo.png']) {
assert.ok(!resources.has(skipped), `skipped path ${skipped} produced no sample: ${JSON.stringify([...resources])}`);
}
assert.ok(resources.has('/release.json'), '/release.json is a route, not a static asset');
-- assert.strictEqual(raw.filter((s) => s.operation === 'http.active_requests').length, 1, 'gauges once per flush');
-- assert.deepStrictEqual(raw.find((s) => s.operation === 'http.active_requests').extra, { last: 0, max: 1 }, 'the two requests are sequential, so the peak is one in flight');
-- const lag = raw.filter((s) => s.operation === 'http.eventloop_lag_ms');
-- assert.strictEqual(lag.length, 1, 'the event-loop lag gauge is emitted once per flush');
-- assert.ok(Number.isFinite(lag[0].extra.value), 'lag lives in extra');
-- assert.ok(!('latency_ms' in lag[0]), 'lag is never in the latency field');
for (const s of raw) valid(s, 'middleware batch');
-- await telemetry.stop();
-- assert.strictEqual(observability.eventLoopMonitorEnabled(), false, 'stop() disables the event-loop monitor');
++ await svc.telemetry.stop();
++ assert.strictEqual(svc.eventLoopMonitorEnabled(), false, 'stop() disables the event-loop monitor');
++ await svc.telemetry.stop();
console.log('telemetry: all checks passed');
})().catch((err) => { console.error(err); process.exit(1); });