openvibers/openvibe.live Commits

Observability (Track O): loopback-only /metrics with golden signals by route template and live gauges (streams, sockets per server, outbox); truthful /api/ready (db required; SFU, Media, Network optional → degraded); openvibe-shared v1.3.0

OP OpenVibers committed 23 Sept 2026 · 6090d00b7d · 5 files, +314 −34


Changes

ddiff --git a/package-lock.json b/package-lock.json
iindex 7a0b1665..0c980ff1 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -27,7 +27,7 @@
"onnxruntime-node": "^1.20.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.13.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.2.2",
-- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.2.1",
++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0",
"sanitize-html": "^2.17.7",
"sdp-transform": "^3.0.0",
"sharp": "^0.33.0",
@@@ -2839,9 +2839,8 @@
}
},
"node_modules/openvibe-shared": {
-- "version": "1.2.1",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.2.1",
-- "integrity": "sha512-t8EHgju5/H2VqkM89LzdCKxBogJ8xilxmvZVJWQIYaGPKe7vMe4yrITlJi/Sy98phsffeFRogpaukDuO6QlDUw==",
++ "version": "1.3.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0",
"license": "MIT",
"dependencies": {
"jsonwebtoken": "^9.0.0"
ddiff --git a/package.json b/package.json
iindex 48e0cb18..d2c3c218 100644
---- a/package.json
++++ b/package.json
@@@ -45,7 +45,7 @@
"onnxruntime-node": "^1.20.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.13.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.2.2",
-- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.2.1",
++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0",
"sanitize-html": "^2.17.7",
"sdp-transform": "^3.0.0",
"sharp": "^0.33.0",
ddiff --git a/server/index.js b/server/index.js
iindex 7a4c3598..4bba5042 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -131,6 +131,13 @@ const whipHandler = require('./streaming/whip-handler');
// ── Express App ──────────────────────────────────────────────
const app = express();
const server = http.createServer(app);
++
++// What this server runs (ADR-016): GET /release.json below, and release_info in /metrics.
++const release = require('openvibe-shared/release').createRelease({ service: 'live', root: path.join(__dirname, '..') });
++// Metrics first, so every request is counted (by route template, never by raw URL). GET /metrics
++// answers direct loopback callers only; through nginx it is a 404 (server/web/observability.js).
++const observability = require('./web/observability');
++const { registry: metricsRegistry } = observability.mountMetrics(app, { release });
const vibeCodingPublishServer = new VibeCodingPublishServer(chatServer, db);
function normalizeOrigin(origin) {
@@@ -723,36 +730,27 @@ app.get('/api/health', (req, res) => {
* genuine gate.
*/
let _bootComplete = false;
--// What this server runs (ADR-016). The shared navbar loads /shared/release-watch.js, which polls
--// this and prompts open tabs after a deploy; it never reloads a tab that is watching, broadcasting
--// or typing.
--{
-- const release = require('openvibe-shared/release').createRelease({ service: 'live', root: path.join(__dirname, '..') });
-- app.get('/release.json', release.handler);
--}
++// The shared navbar loads /shared/release-watch.js, which polls this and prompts open tabs after a
++// deploy; it never reloads a tab that is watching, broadcasting or typing.
++app.get('/release.json', release.handler);
++
++// Required: boot finished and the database answers. Optional (a failure is "degraded", still 200):
++// the WebRTC SFU, OpenVibe.Media and the Network signing key — Live serves channels, chat and RTMP
++// without them, so they must neither pass silently nor take the whole service out of rotation.
++const readiness = observability.createLiveReadiness({
++ release,
++ bootComplete: () => _bootComplete,
++ dbQuery: () => db.get('SELECT 1 AS ok'),
++ sfuReady: () => webrtcSFU.ready === true && !!webrtcSFU.worker,
++ mediaUrl: mediaClient.MEDIA_URL,
++ networkKey: () => require('./auth/auth').getNetworkPublicKey(),
++});
++app.get('/api/ready', readiness.handler);
--app.get('/api/ready', (req, res) => {
-- res.set('Cache-Control', 'no-store');
-- const checks = { boot: _bootComplete, db: false };
-- try {
-- // Cheapest possible proof that the connection is open and the schema exists.
-- db.get('SELECT 1 AS ok');
-- checks.db = true;
-- } catch (err) {
-- checks.dbError = String(err && err.message || err).slice(0, 200);
-- }
-- const ready = checks.boot && checks.db;
-- res.status(ready ? 200 : 503).json({
-- ready,
-- checks,
-- uptime: Math.round(process.uptime()),
-- // Advisory only — a deploy gate should not wait on optional subsystems, which can be
-- // legitimately unavailable (no mediasoup worker on a box that only relays RTMP, say).
-- optional: {
-- sfu: (() => { try { return !!webrtcSFU.isReady?.(); } catch { return null; } })(),
-- media: (() => { try { return require('./media-client').lastOk === true; } catch { return null; } })(),
-- },
-- });
++observability.registerDomainGauges(metricsRegistry, {
++ liveStreams: () => db.getLiveStreams().length,
++ wsServers: { chat: chatServer, broadcast: broadcastServer, control: controlServer, call: callServer },
++ outboxStatus: () => require('./events/stream-events').status(),
});
// ── Updates / Changelog ──────────────────────────────────────
ddiff --git a/server/web/observability.js b/server/web/observability.js
nnew file mode 100644
iindex 00000000..f1170c4a
---- /dev/null
++++ b/server/web/observability.js
@@@ -0,0 +1,108 @@
++'use strict';
++/**
++ * Metrics and readiness for Live (roadmap Track O, §15.19), on openvibe-shared/metrics and
++ * openvibe-shared/ready.
++ *
++ * GET /metrics Prometheus text, direct loopback callers only (404 through nginx or any proxy):
++ * http_requests_total{method,route,status_class}, http_request_duration_seconds,
++ * http_requests_in_flight, process/event-loop metrics, release_info{service,release},
++ * live_streams_live, live_ws_connections{server}, live_events_outbox{status}
++ * GET /api/ready 200 while Live can serve (boot finished, the database answers); 503 otherwise.
++ * Media, the WebRTC SFU and the Network signing key are optional: when one is
++ * missing the answer stays 200 with status "degraded" and names it, because Live
++ * still serves channels, chat and RTMP without them.
++ *
++ * Everything the checks and gauges read is injected, so the test drives this module without booting
++ * the whole server.
++ */
++const metrics = require('openvibe-shared/metrics');
++const { createReadiness } = require('openvibe-shared/ready');
++
++// Long-lived responses whose "duration" is a viewing session, not a request.
++const STREAMING_PATHS = [/^\/api\/live-events$/, /^\/api\/streams\/rtmp-proxy\//];
++// Static mounts: served by express.static (no route), labelled by mount instead of 'unmatched'.
++const STATIC_MOUNT = /^\/(js|css|fragments|shared|assets|data|media|docs-assets|images|img|fonts|sounds)\//;
++
++function normalize(req) {
++ if (req.route) return null; // the matched route's template (openvibe-shared/metrics routeLabel)
++ const m = STATIC_MOUNT.exec(String(req.originalUrl || req.url || '').split('?')[0]);
++ return m ? `/${m[1]}/*` : null;
++}
++
++/**
++ * Mount the HTTP metrics middleware and GET /metrics. Call before any other middleware.
++ * @returns {{ registry }}
++ */
++function mountMetrics(app, { release }) {
++ const m = metrics.instrument(app, {
++ service: 'live',
++ release: release && release.release,
++ normalize,
++ skip: (req) => STREAMING_PATHS.some((re) => re.test(req.path)),
++ });
++ return m;
++}
++
++/**
++ * Domain gauges, read at scrape time. A source that throws is left out of that scrape.
++ * liveStreams() -> number of live streams (the rows /api/streams lists)
++ * wsServers -> { chat, broadcast, control, call } objects with a `wss` (ws.Server) once initialised
++ * outboxStatus() -> server/events/stream-events.js status()
++ */
++function registerDomainGauges(registry, { liveStreams, wsServers = {}, outboxStatus }) {
++ registry.gauge({ name: 'live_streams_live', help: 'Streams currently marked live (the list /api/streams serves)', collect: () => liveStreams() });
++ registry.gauge({
++ name: 'live_ws_connections', help: 'Open WebSocket connections per WebSocket server', labelNames: ['server'],
++ collect: () => Object.entries(wsServers)
++ .filter(([, s]) => s && s.wss && s.wss.clients) // a server that never initialised reports nothing
++ .map(([server, s]) => ({ labels: { server }, value: s.wss.clients.size })),
++ });
++ registry.gauge({
++ name: 'live_events_outbox', help: 'Stream lifecycle events in the transactional outbox, by status (absent while the outbox is disabled)', labelNames: ['status'],
++ collect: () => {
++ const s = outboxStatus();
++ if (!s || !s.enabled) return null;
++ return [{ labels: { status: 'pending' }, value: Number(s.pending) || 0 }, { labels: { status: 'rejected' }, value: Number(s.rejected) || 0 }];
++ },
++ });
++}
++
++/**
++ * The /api/ready handler.
++ * bootComplete() -> boolean required
++ * dbQuery() -> row required (a real query; throws when the database cannot answer)
++ * sfuReady() -> boolean optional (WebRTC broadcasting/viewing through mediasoup)
++ * mediaUrl optional (OpenVibe.Media: VODs, clips, thumbnails), checked by
++ * GET <mediaUrl>/healthz, cached 30 s
++ * networkKey() -> PEM or null optional (sign-in: RS256 key from OpenVibe.Network)
++ */
++function createLiveReadiness({ release, bootComplete, dbQuery, sfuReady, mediaUrl, networkKey, fetchImpl = globalThis.fetch }) {
++ return createReadiness({
++ service: 'live',
++ release: release && release.release,
++ checks: [
++ { name: 'boot', required: true, description: 'boot finished', check: () => (bootComplete() ? true : 'boot not finished') },
++ { name: 'db', required: true, description: 'SQLite answers a query', check: () => { const row = dbQuery(); return row ? true : 'no row'; } },
++ { name: 'sfu', required: false, description: 'WebRTC SFU (mediasoup worker)', check: () => (sfuReady() ? true : 'mediasoup worker not running') },
++ {
++ name: 'media', required: false, cacheMs: 30000, timeoutMs: 2000, description: 'OpenVibe.Media (VODs, clips, thumbnails)',
++ check: async () => {
++ const res = await fetchImpl(`${mediaUrl}/healthz`, { signal: AbortSignal.timeout(2000) });
++ return res.ok ? true : `Media answered ${res.status}`;
++ },
++ },
++ { name: 'network_key', required: false, description: 'OpenVibe.Network RS256 key (sign-in)', check: () => (/BEGIN [A-Z ]*PUBLIC KEY/.test(String(networkKey() || '')) ? true : 'Network public key not loaded: sign-in cannot be verified') },
++ ],
++ // Fields the previous /api/ready served, so nothing that read them breaks.
++ details: (body) => ({
++ uptime: Math.round(process.uptime()),
++ optional: {
++ sfu: body.checks.sfu.status === 'ok',
++ media: body.checks.media.status === 'ok',
++ network_key: body.checks.network_key.status === 'ok',
++ },
++ }),
++ });
++}
++
++module.exports = { mountMetrics, registerDomainGauges, createLiveReadiness, normalize };
ddiff --git a/test/observability.test.js b/test/observability.test.js
nnew file mode 100644
iindex 00000000..a052903f
---- /dev/null
++++ b/test/observability.test.js
@@@ -0,0 +1,175 @@
++'use strict';
++/**
++ * Metrics and readiness (server/web/observability.js, roadmap Track O):
++ * - /metrics answers direct loopback callers only; through a proxy (X-Forwarded-For) it is a 404
++ * - HTTP metrics are labelled by route template, never by raw URL, id or query
++ * - release_info and the domain gauges (live streams, WebSocket connections, outbox)
++ * - /api/ready: boot + db required (503 when either fails); SFU, Media and the Network key optional
++ * (200 "degraded", naming what is missing)
++ * A temp SQLite database stands in for Live's; no Media, Network or mediasoup is needed.
++ *
++ * Run: node test/observability.test.js
++ */
++const assert = require('assert');
++const fs = require('fs');
++const http = require('http');
++const os = require('os');
++const path = require('path');
++const express = require('express');
++const Database = require('better-sqlite3');
++const observability = require('../server/web/observability');
++
++const tmpDb = path.join(os.tmpdir(), `openvibe-observability-test-${process.pid}-${Date.now()}.db`);
++
++(async () => {
++ const sqlite = new Database(tmpDb);
++ sqlite.exec('CREATE TABLE streams (id INTEGER PRIMARY KEY, is_live INTEGER)');
++ sqlite.prepare('INSERT INTO streams (is_live) VALUES (1), (1), (0)').run();
++ let dbBroken = false;
++ const dbQuery = () => { if (dbBroken) throw new Error('SQLITE_IOERR: disk I/O error'); return sqlite.prepare('SELECT 1 AS ok').get(); };
++
++ // Fake Media: /healthz answers 200 or 502.
++ let mediaUp = true; let mediaHits = 0;
++ const media = http.createServer((req, res) => { mediaHits++; res.statusCode = req.url === '/healthz' && mediaUp ? 200 : 502; res.end('{}'); });
++ await new Promise((r) => media.listen(0, '127.0.0.1', r));
++ const mediaUrl = `http://127.0.0.1:${media.address().port}`;
++
++ const release = { release: 'abc123def456' };
++ let booted = false; let sfu = true; let key = '-----BEGIN PUBLIC KEY-----\nMII…\n-----END PUBLIC KEY-----';
++ const fakeWss = (n) => ({ wss: { clients: { size: n } } });
++ let outbox = { enabled: true, pending: 4, rejected: 1 };
++
++ const app = express();
++ const { registry } = observability.mountMetrics(app, { release });
++ const readiness = observability.createLiveReadiness({
++ release, bootComplete: () => booted, dbQuery, sfuReady: () => sfu, mediaUrl, networkKey: () => key,
++ });
++ app.get('/api/ready', readiness.handler);
++ observability.registerDomainGauges(registry, {
++ liveStreams: () => sqlite.prepare('SELECT COUNT(*) AS n FROM streams WHERE is_live = 1').get().n,
++ wsServers: { chat: fakeWss(12), broadcast: fakeWss(2), control: fakeWss(0), call: { wss: null } },
++ outboxStatus: () => outbox,
++ });
++ const vods = express.Router();
++ vods.get('/:id', (req, res) => res.json({ id: req.params.id }));
++ app.use('/api/vods', vods);
++ app.get('/api/live-events', (_req, res) => { res.setHeader('Content-Type', 'text/event-stream'); res.end(); });
++ app.use('/js', express.static(path.join(__dirname, '..', 'public', 'js')));
++ app.get('*', (_req, res) => res.send('spa'));
++
++ const server = await new Promise((r) => { const s = app.listen(0, '127.0.0.1', () => r(s)); });
++ const base = `http://127.0.0.1:${server.address().port}`;
++ const get = (p, headers = {}) => new Promise((resolve, reject) => {
++ http.get(base + p, { headers }, (res) => { let b = ''; res.on('data', (d) => { b += d; }); res.on('end', () => resolve({ status: res.statusCode, body: b })); }).on('error', reject);
++ });
++
++ // ── readiness ──
++ let r = await get('/api/ready');
++ assert.strictEqual(r.status, 503, 'not ready before boot completes');
++ let body = JSON.parse(r.body);
++ assert.strictEqual(body.ready, false);
++ assert.deepStrictEqual(body.failed, ['boot']);
++
++ booted = true;
++ r = await get('/api/ready');
++ body = JSON.parse(r.body);
++ assert.strictEqual(r.status, 200);
++ assert.strictEqual(body.status, 'ready');
++ assert.strictEqual(body.service, 'live');
++ assert.strictEqual(body.release, 'abc123def456');
++ for (const name of ['boot', 'db', 'sfu', 'media', 'network_key']) {
++ const c = body.checks[name];
++ assert.strictEqual(c.status, 'ok', name);
++ assert.strictEqual(typeof c.latency_ms, 'number');
++ assert.ok(Date.parse(c.checked_at), `${name} checked_at`);
++ }
++ assert.strictEqual(body.checks.db.required, true);
++ assert.strictEqual(body.checks.media.required, false);
++ assert.deepStrictEqual(body.optional, { sfu: true, media: true, network_key: true }, 'the previous optional booleans are still served');
++ assert.strictEqual(typeof body.uptime, 'number');
++
++ // Optional dependencies down: still 200, but degraded and named.
++ sfu = false; key = null;
++ r = await get('/api/ready');
++ body = JSON.parse(r.body);
++ assert.strictEqual(r.status, 200);
++ assert.strictEqual(body.status, 'degraded');
++ assert.deepStrictEqual(body.degraded.sort(), ['network_key', 'sfu']);
++ assert.match(body.checks.sfu.error, /mediasoup/);
++ assert.strictEqual(body.optional.sfu, false);
++
++ // Media is cached for 30 s: its checked_at is when it really ran, and a down Media is not re-asked.
++ const hitsBefore = mediaHits;
++ mediaUp = false;
++ r = await get('/api/ready');
++ assert.strictEqual(mediaHits, hitsBefore, 'Media check is cached');
++ assert.strictEqual(JSON.parse(r.body).checks.media.status, 'ok', 'cached result, with its own checked_at');
++
++ // The database failing is not ready (503), whatever else is up.
++ sfu = true; key = '-----BEGIN PUBLIC KEY-----x';
++ dbBroken = true;
++ r = await get('/api/ready');
++ body = JSON.parse(r.body);
++ assert.strictEqual(r.status, 503);
++ assert.deepStrictEqual(body.failed, ['db']);
++ assert.match(body.checks.db.error, /SQLITE_IOERR/);
++ dbBroken = false;
++
++ // A fresh readiness with Media down reports it degraded.
++ const r2 = observability.createLiveReadiness({ release, bootComplete: () => true, dbQuery, sfuReady: () => true, mediaUrl, networkKey: () => key });
++ const b2 = await r2.run();
++ assert.strictEqual(b2.ready, true);
++ assert.deepStrictEqual(b2.degraded, ['media']);
++ assert.strictEqual(b2.checks.media.error, 'Media answered 502');
++ const r3 = observability.createLiveReadiness({ release, bootComplete: () => true, dbQuery, sfuReady: () => true, mediaUrl: 'http://127.0.0.1:1', networkKey: () => key });
++ const b3 = await r3.run();
++ assert.deepStrictEqual(b3.degraded, ['media'], 'Media unreachable is degraded, not down');
++
++ // ── metrics ──
++ await get('/api/vods/12345?token=secret');
++ await get('/api/vods/67890');
++ await get('/some/channel-name-xyz');
++ await get('/api/live-events');
++ await get('/js/app.js?v=deadbeef');
++ const m = await get('/metrics');
++ assert.strictEqual(m.status, 200);
++ const text = m.body;
++ assert.ok(text.includes('http_requests_total{method="GET",route="/api/vods/:id",status_class="2xx"} 2\n'), text);
++ assert.ok(text.includes('route="*"'), 'the SPA fallback is its template');
++ assert.ok(text.includes('route="/js/*"'), 'static mounts are labelled by mount');
++ const labels = text.split('\n').filter((l) => l.startsWith('http_')).map((l) => (l.match(/\{[^}]*\}/) || [''])[0]).join('\n');
++ assert.ok(!/12345|67890|secret|channel-name|app\.js|deadbeef/.test(labels), 'no ids, queries or raw paths in labels');
++ assert.ok(!labels.includes('/api/live-events'), 'the SSE stream is not a request');
++ assert.ok(text.includes('release_info{service="live",release="abc123def456"} 1\n'));
++ assert.ok(text.includes('live_streams_live 2\n'));
++ assert.ok(text.includes('live_ws_connections{server="chat"} 12\n'));
++ assert.ok(text.includes('live_ws_connections{server="broadcast"} 2\n'));
++ assert.ok(text.includes('live_ws_connections{server="control"} 0\n'));
++ assert.ok(!text.includes('live_ws_connections{server="call"}'), 'an uninitialised server reports nothing');
++ assert.ok(text.includes('live_events_outbox{status="pending"} 4\n'));
++ assert.ok(text.includes('live_events_outbox{status="rejected"} 1\n'));
++ assert.ok(text.includes('\nprocess_resident_memory_bytes '));
++ assert.ok(text.includes('\nnodejs_eventloop_lag_seconds{stat="p99"} '));
++
++ outbox = { enabled: false };
++ assert.ok(!(await get('/metrics')).body.includes('live_events_outbox{'), 'a disabled outbox reports nothing, not zero');
++
++ // Through nginx (X-Forwarded-For on a loopback connection) /metrics does not exist.
++ assert.strictEqual((await get('/metrics', { 'X-Forwarded-For': '203.0.113.7' })).status, 404);
++ assert.strictEqual((await get('/metrics', { 'X-Real-IP': '203.0.113.7' })).status, 404);
++
++ // server/index.js mounts metrics before anything else and serves /api/ready from this module.
++ const src = fs.readFileSync(path.join(__dirname, '..', 'server', 'index.js'), 'utf8');
++ const mountAt = src.indexOf('observability.mountMetrics(app');
++ assert.ok(mountAt > 0 && mountAt < src.indexOf('app.use(helmet('), 'metrics middleware is mounted before the rest of the middleware');
++ assert.ok(/app\.get\('\/api\/ready', readiness\.handler\)/.test(src));
++
++ server.close(); media.close(); sqlite.close();
++ for (const f of [tmpDb, `${tmpDb}-wal`, `${tmpDb}-shm`]) { try { fs.unlinkSync(f); } catch { /* */ } }
++ console.log('observability: all checks passed');
++ process.exit(0);
++})().catch((err) => {
++ console.error(err);
++ for (const f of [tmpDb, `${tmpDb}-wal`, `${tmpDb}-shm`]) { try { fs.unlinkSync(f); } catch { /* */ } }
++ process.exit(1);
++});