openvibers/openvibe.media Commits
Truthful GET /api/ready and loopback-only GET /metrics (openvibe-shared v1.3.0)
Changes
@@@ -200,6 +200,14 @@ directories are shared across apps) and responses carry a `note` saying so.and header `X-OVMedia-Signature: sha256=<hex hmac-sha256 of the raw body withthe app's webhook_secret>`. 3 attempts with backoff, 10 s timeout.++## Health, readiness and metrics++++| path | behavior |++|---|---|++| `GET /healthz` | liveness: the process answers (unchanged; it checks nothing else) |++| `GET /api/ready` | readiness from real checks, each with `status`, `required`, `latency_ms`, `checked_at`. **Required** (503 when one fails): `db` (a query against the SQLite database), `storage_vods`, `storage_clips`, `storage_pastes`, `storage_thumbnails`, `storage_files`, `storage_objects` (a probe file is written and removed). **Optional** (listed in `degraded`, still 200): `network_jwks` (user JWTs and service tokens; app keys work without it), `remote_b2` / `remote_r2` (HeadBucket, at most once a minute; present only when the provider is configured), `events_outbox` (backlog over 1000 events). Also `recordings_in_progress`. |++| `GET /metrics` | Prometheus text (openvibe-shared/metrics) for **direct loopback callers only**; 404 through nginx. HTTP golden signals by route template, process metrics, `release_info`, plus `media_recordings_in_progress`, `media_object_locations{provider,state}`, `media_objects{lifecycle_status}`, `media_events_outbox{status}` (only while the outbox runs). |++## Public serving (no auth unless the item is private)| path | behavior |@@@ -17,6 +17,9 @@ server {proxy_read_timeout 3600s;proxy_send_timeout 3600s;++ # Prometheus metrics are for loopback scrapers only (the app 404s proxied callers too).++ location = /metrics { return 404; }++location / {proxy_pass http://127.0.0.1:4100;proxy_http_version 1.1;@@@ -18,7 +18,7 @@"multer": "^1.4.5-lts.1","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.8.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.0",++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0","sharp": "^0.33.4"}},@@@ -1944,9 +1944,8 @@}},"node_modules/openvibe-shared": {-- "version": "1.2.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.2.0",-- "integrity": "sha512-XYDP1I0f44T+ImWqcErW3P1bBVzsPZ8bgjYBk9QVJP6zDiK5/GYj5ynQhh5GhB9FhxvK92rFu7Lx1ZohZDNSkA==",++ "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"@@@ -20,7 +20,7 @@"dotenv": "^16.4.5","express": "^4.19.2","multer": "^1.4.5-lts.1",-- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.2.0",++ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0","sharp": "^0.33.4"}}@@@ -60,6 +60,9 @@ function startJwksRefresh() {if (_jwksTimer.unref) _jwksTimer.unref();}++/** Whether the Network public key has loaded (user JWTs and service tokens verify only once it has). */++function jwksLoaded() { return !!_networkPublicKeyPem; }++function stopJwksRefresh() {if (_jwksTimer) { clearInterval(_jwksTimer); _jwksTimer = null; }}@@@ -327,5 +330,6 @@ module.exports = {seedApps,startJwksRefresh,stopJwksRefresh,++ jwksLoaded,fetchNetworkPublicKey,};@@@ -27,6 +27,10 @@ auth.startJwksRefresh();const app = express();app.set('trust proxy', true);++// Metrics first, so every route below is measured (GET /metrics: loopback callers only).++const release = require('openvibe-shared/release').createRelease({ service: 'media', root: require('path').join(__dirname, '..') });++const observability = require('./observability');++const instrumented = observability.instrument(app, { release: release.release });// Object API v2 content uploads read the raw request body, so they sit ahead of the body parsers.app.put('/api/v2/:app/objects/:id/content', ...require('./objects/routes').contentHandlers);app.use(express.json({ limit: '2mb' }));@@@ -83,7 +87,6 @@ app.options(['/api/v1/:app/*', '/api/v2/:app/*'], (req, res) => {// ── Routes ───────────────────────────────────────────────────// What this server runs (ADR-016); the shared navbar's release-watch polls it.--const release = require('openvibe-shared/release').createRelease({ service: 'media', root: require('path').join(__dirname, '..') });app.get('/release.json', release.handler);app.get('/healthz', (req, res) => {@@@ -96,6 +99,17 @@ app.get('/healthz', (req, res) => {});});++// Readiness: required = database + writable storage; optional (degraded) = Network key, remote tiers, events outbox.++{++ const vodStorageForReady = require('./vod/vod-storage');++ observability.mountReady(app, instrumented.registry, {++ release: release.release, db, config, auth,++ recorder: require('./vod/recorder'),++ events: require('./events'),++ remote: { configured: (n) => vodStorageForReady.providerConfigured(n), probe: (n) => vodStorageForReady.probeProvider(n) },++ });++}++// Per-app media stats (app-key only) — hero/dashboard counters in the owning app.app.get('/api/v1/:app/stats', auth.tenantAuth(), (req, res) => {try {nnew file mode 100644@@@ -0,0 +1,109 @@++'use strict';++/**++ * Metrics and readiness for OpenVibe.Media (roadmap Track O, openvibe-shared/metrics + /ready).++ *++ * GET /metrics Prometheus text; direct loopback callers only (404 through nginx)++ * GET /api/ready named checks; 503 only when a required one fails, optional failures → degraded++ *++ * Required (Media cannot serve without them): the SQLite database answers a real query, and every++ * local storage directory accepts a write. Optional (one capability degrades, the rest keeps++ * working): the Network public key (user-JWT and service-token routes; app-key tenants are++ * unaffected), each CONFIGURED remote tier (B2 cold storage, R2 cache; a HeadBucket at most once a++ * minute), and the Events outbox (webhooks still deliver when it backs up).++ *++ * `instrument()` must run before any route is mounted; `mountReady()` any time after.++ */++const fs = require('fs');++const path = require('path');++const crypto = require('crypto');++const metrics = require('openvibe-shared/metrics');++const { createReadiness } = require('openvibe-shared/ready');++++const OUTBOX_BACKLOG_LIMIT = 1000;++++function instrument(app, { release }) {++ return metrics.instrument(app, {++ service: 'media',++ release,++ // Static brand files: one label for the directory, never the file name.++ normalize: (req) => (!req.route && req.path.startsWith('/assets/') ? '/assets/*' : null),++ });++}++++/** Write and remove a probe file: the directory exists and accepts writes. */++function writable(dir) {++ const probe = path.join(dir, `.ready-${process.pid}-${crypto.randomBytes(4).toString('hex')}`);++ fs.writeFileSync(probe, 'ok');++ fs.unlinkSync(probe);++ return true;++}++++function storageDirs(config) {++ return {++ vods: config.vod.path,++ clips: config.vod.clipsPath,++ pastes: config.pastes.path,++ thumbnails: config.thumbnails.path,++ files: config.files.path,++ objects: config.objects.path,++ };++}++++function domainMetrics(registry, { db, recorder, events }) {++ registry.gauge({ name: 'media_recordings_in_progress', help: 'Recordings ffmpeg is writing right now', collect: () => recorder.activeCount() });++ registry.gauge({++ name: 'media_object_locations', help: 'Stored copies of media objects by provider and verification state', labelNames: ['provider', 'state'],++ collect: () => db.all('SELECT provider, state, COUNT(*) AS n FROM media_locations GROUP BY provider, state').map((r) => ({ labels: { provider: r.provider, state: r.state }, value: r.n })),++ });++ registry.gauge({++ name: 'media_objects', help: 'Media objects by lifecycle status', labelNames: ['lifecycle_status'],++ collect: () => db.all('SELECT lifecycle_status, COUNT(*) AS n FROM media_objects GROUP BY lifecycle_status').map((r) => ({ labels: { lifecycle_status: r.lifecycle_status }, value: r.n })),++ });++ registry.gauge({++ name: 'media_events_outbox', help: 'Events outbox rows waiting to reach OpenVibe.Events, and rows Events rejected', labelNames: ['status'],++ collect: () => {++ const s = events.status();++ if (!s.enabled) return null; // no outbox running: no series, never a made-up zero++ return [{ labels: { status: 'pending' }, value: s.pending }, { labels: { status: 'rejected' }, value: s.rejected }];++ },++ });++}++++function createMediaReadiness({ release, db, config, auth, recorder, events, remote }) {++ const checks = [++ { name: 'db', required: true, check: () => { const r = db.get('SELECT COUNT(*) AS n FROM apps'); return { ok: Number.isInteger(r.n), detail: { apps: r.n } }; } },++ ];++ for (const [name, dir] of Object.entries(storageDirs(config))) {++ checks.push({ name: `storage_${name}`, required: true, check: () => writable(dir) });++ }++ checks.push({ name: 'network_jwks', required: false, description: 'user sign-in and service tokens (app keys work without it)', check: () => auth.jwksLoaded() || 'Network public key not loaded yet' });++ for (const name of ['b2', 'r2']) {++ if (!remote.configured(name)) continue; // not configured: no check, rather than a pretend pass++ checks.push({ name: `remote_${name}`, required: false, cacheMs: 60 * 1000, timeoutMs: 5000, check: () => remote.probe(name) });++ }++ checks.push({++ name: 'events_outbox', required: false, description: 'durable events to OpenVibe.Events (webhooks are separate)',++ check: () => {++ const s = events.status();++ if (!s.enabled) return { ok: true, detail: { enabled: false } };++ if (s.pending > OUTBOX_BACKLOG_LIMIT) return { ok: false, error: `${s.pending} events waiting (limit ${OUTBOX_BACKLOG_LIMIT})`, detail: { pending: s.pending } };++ return { ok: true, detail: { pending: s.pending, rejected: s.rejected } };++ },++ });++ return createReadiness({++ service: 'media',++ release,++ checks,++ details: () => ({ recordings_in_progress: recorder.activeCount() }),++ });++}++++/** Everything after instrument(): domain gauges and GET /api/ready. */++function mountReady(app, registry, deps) {++ domainMetrics(registry, deps);++ const readiness = createMediaReadiness(deps);++ app.get('/api/ready', readiness.handler);++ return readiness;++}++++module.exports = { instrument, mountReady, createMediaReadiness, domainMetrics, storageDirs };@@@ -1170,6 +1170,20 @@ async function bucketStatus() {return out;}++/** Live HeadBucket probe of one configured provider (readiness); throws when unreachable. */++async function probeProvider(name) {++ if (!providerConfigured(name)) throw new Error(`${name} is not configured`);++ loadSdk();++ try {++ await clientFor(name).send(new S3.HeadBucketCommand({ Bucket: PROVIDER_ENV[name].bucket }));++ providerHealthy[name] = true;++ return true;++ } catch (err) {++ providerHealthy[name] = false;++ throw new Error(`${name} HeadBucket failed: ${err.name || 'error'}`);++ }++}++function getStatus() {const settings = getSettings();const localDisk = diskUsage(config.vod.path);@@@ -1263,6 +1277,7 @@ module.exports = {stop,getStatus,bucketStatus,++ probeProvider,getBucketUsage,estimateCloudCosts,getSettings,nnew file mode 100644@@@ -0,0 +1,154 @@++'use strict';++// GET /metrics (loopback only, route templates, Media gauges) and GET /api/ready (required: db +++// writable storage; optional: Network key, configured remote tiers, events outbox), wired the way++// server/index.js wires them — and then the real server booted once to prove index.js does.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++const net = require('net');++const { spawn } = require('child_process');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-obs-'));++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++const ENV = {++ DB_PATH: path.join(tmp, 'media.db'), VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), PASTES_PATH: dir('pastes'),++ THUMBNAILS_PATH: dir('thumbnails'), FILES_PATH: dir('files'), OBJECTS_PATH: dir('objects'), ASSETS_PATH: dir('assets'),++ MEDIA_PUBLIC_URL: 'https://media.test', OV_NETWORK_URL: 'http://127.0.0.1:9', RELEASE_COMMIT: 'abcdef1234567',++};++for (const k of ['MEDIA_B2_ENDPOINT', 'MEDIA_B2_BUCKET', 'MEDIA_R2_ENDPOINT', 'MEDIA_R2_BUCKET', 'EVENTS_URL']) process.env[k] = ''; // never a real tier or Events (dotenv keeps set values)++Object.assign(process.env, ENV);++++const db = require('../server/db/database');++const config = require('../server/config');++const express = require('express');++const observability = require('../server/observability');++++function request(base, p, headers = {}) {++ return 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);++ });++}++const listen = (app) => new Promise((resolve) => { const s = app.listen(0, '127.0.0.1', () => resolve(s)); });++++(async () => {++ db.getDb();++ db.upsertApp({ app_id: 'live', api_key: 'live-key' });++++ // ── Wiring as in index.js, with injectable dependencies ──++ let jwks = true; let remoteUp = true; let remoteProbes = 0; let outbox = { enabled: true, pending: 2, rejected: 1 }; let recording = 1;++ const app = express();++ const inst = observability.instrument(app, { release: 'abcdef123456' });++ app.use('/o', require('express').Router().get('/:id', (req, res) => res.status(404).json({ error: 'Not found' })));++ observability.mountReady(app, inst.registry, {++ release: 'abcdef123456', db, config,++ auth: { jwksLoaded: () => jwks },++ recorder: { activeCount: () => recording },++ events: { status: () => outbox },++ remote: { configured: (n) => n === 'b2', probe: async () => { remoteProbes++; if (!remoteUp) throw new Error('b2 HeadBucket failed: NetworkingError'); return true; } },++ });++ const server = await listen(app);++ const base = `http://127.0.0.1:${server.address().port}`;++++ let r = await request(base, '/api/ready');++ assert.strictEqual(r.status, 200, r.body);++ let body = JSON.parse(r.body);++ assert.strictEqual(body.ready, true);++ assert.strictEqual(body.status, 'ready');++ assert.strictEqual(body.service, 'media');++ assert.strictEqual(body.recordings_in_progress, 1);++ const names = Object.keys(body.checks).sort();++ assert.deepStrictEqual(names, ['db', 'events_outbox', 'network_jwks', 'remote_b2', 'storage_clips', 'storage_files', 'storage_objects', 'storage_pastes', 'storage_thumbnails', 'storage_vods'], 'r2 is not configured, so it has no check');++ for (const [n, c] of Object.entries(body.checks)) {++ assert.strictEqual(c.status, 'ok', n);++ assert.strictEqual(typeof c.latency_ms, 'number');++ assert.ok(Date.parse(c.checked_at));++ assert.strictEqual(c.required, n === 'db' || n.startsWith('storage_'), `${n} required?`);++ }++ assert.deepStrictEqual(body.checks.db.detail, { apps: 1 });++ assert.deepStrictEqual(fs.readdirSync(config.vod.path), [], 'the write probe leaves nothing behind');++++ // Optional failures: degraded, still 200.++ jwks = false; remoteUp = false; outbox = { enabled: true, pending: 5000, rejected: 0 };++ r = await request(base, '/api/ready');++ body = JSON.parse(r.body);++ assert.strictEqual(r.status, 200);++ assert.strictEqual(body.status, 'degraded');++ assert.deepStrictEqual(body.degraded.sort(), ['events_outbox', 'network_jwks']);++ assert.strictEqual(body.checks.remote_b2.status, 'ok', 'the remote tier probe is cached for a minute');++ assert.strictEqual(remoteProbes, 1);++++ // A remote-tier failure on a fresh readiness (no cache): degraded, never "down".++ const fresh = observability.createMediaReadiness({ release: 'x', db, config, auth: { jwksLoaded: () => true }, recorder: { activeCount: () => 0 }, events: { status: () => ({ enabled: false }) }, remote: { configured: () => true, probe: async (n) => { throw new Error(`${n} HeadBucket failed: AccessDenied`); } } });++ body = await fresh.run();++ assert.strictEqual(body.ready, true);++ assert.deepStrictEqual(body.degraded, ['remote_b2', 'remote_r2']);++ assert.strictEqual(body.checks.remote_r2.error, 'r2 HeadBucket failed: AccessDenied');++ assert.deepStrictEqual(body.checks.events_outbox.detail, { enabled: false });++++ // A storage directory that cannot be written: not ready, 503, named.++ jwks = true; outbox = { enabled: false };++ fs.rmSync(config.vod.clipsPath, { recursive: true });++ fs.writeFileSync(config.vod.clipsPath, 'not a directory');++ r = await request(base, '/api/ready');++ body = JSON.parse(r.body);++ assert.strictEqual(r.status, 503);++ assert.strictEqual(body.ready, false);++ assert.deepStrictEqual(body.failed, ['storage_clips']);++ fs.rmSync(config.vod.clipsPath); fs.mkdirSync(config.vod.clipsPath);++++ // ── /metrics ──++ await request(base, '/o/med_01J9ZX4YQ0ABCDEFGHJKMNPQRS');++ await request(base, '/assets/icon-192.png');++ recording = 2;++ r = await request(base, '/metrics');++ assert.strictEqual(r.status, 200);++ assert.ok(r.body.includes('release_info{service="media",release="abcdef123456"} 1\n'));++ assert.ok(r.body.includes('http_requests_total{method="GET",route="/o/:id",status_class="4xx"} 1\n'), 'object ids never reach a label');++ assert.ok(r.body.includes('http_requests_total{method="GET",route="/assets/*",status_class="4xx"} 1\n'));++ assert.ok(r.body.includes('http_requests_total{method="GET",route="/api/ready",status_class="2xx"}'));++ assert.ok(!r.body.includes('med_01J9'));++ assert.ok(r.body.includes('media_recordings_in_progress 2\n'));++ assert.ok(r.body.includes('# TYPE media_object_locations gauge'));++ assert.ok(r.body.includes('# TYPE media_objects gauge'));++ assert.ok(!r.body.includes('media_events_outbox{'), 'no outbox running: no invented series');++ r = await request(base, '/metrics', { 'X-Forwarded-For': '203.0.113.7' });++ assert.strictEqual(r.status, 404, 'proxied callers never see metrics');++ server.close();++ inst.stop();++++ // ── The real index.js boots with both routes ──++ const port = await new Promise((resolve) => { const s = net.createServer().listen(0, '127.0.0.1', () => { const p = s.address().port; s.close(() => resolve(p)); }); });++ const bootDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-obs-boot-'));++ const env = { ...process.env, ...ENV, DB_PATH: path.join(bootDir, 'media.db'), PORT: String(port), HOST: '127.0.0.1', NODE_ENV: 'test' };++ for (const k of ['VOD_PATH', 'CLIPS_PATH', 'PASTES_PATH', 'THUMBNAILS_PATH', 'FILES_PATH', 'OBJECTS_PATH', 'ASSETS_PATH']) env[k] = path.join(bootDir, k.toLowerCase());++ const child = spawn(process.execPath, [path.join(__dirname, '..', 'server', 'index.js')], { env, stdio: ['ignore', 'pipe', 'pipe'] });++ let log = ''; child.stdout.on('data', (d) => { log += d; }); child.stderr.on('data', (d) => { log += d; });++ const b2 = `http://127.0.0.1:${port}`;++ let up = null;++ for (let i = 0; i < 100 && !up; i++) {++ await new Promise((res) => setTimeout(res, 100));++ try { up = await request(b2, '/api/ready'); } catch { /* not listening yet */ }++ }++ try {++ assert.ok(up, `server did not start:\n${log}`);++ body = JSON.parse(up.body);++ assert.strictEqual(up.status, 200, up.body);++ assert.strictEqual(body.release, 'abcdef123456');++ assert.ok(body.checks.db && body.checks.storage_objects);++ assert.ok(!body.checks.remote_b2, 'unconfigured tiers are not checked');++ const m = await request(b2, '/metrics');++ assert.strictEqual(m.status, 200);++ assert.ok(m.body.includes('http_requests_total{method="GET",route="/api/ready",status_class="2xx"} 1\n'), m.body.slice(0, 400));++ const h = await request(b2, '/healthz');++ assert.strictEqual(h.status, 200, '/healthz still answers');++ } finally {++ child.kill('SIGKILL');++ }++ console.log('observability: all checks passed');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -200,6 +200,14 @@ directories are shared across apps) and responses carry a `note` saying so.
@@ -200,6 +200,14 @@ directories are shared across apps) and responses carry a `note` saying so.
and header `X-OVMedia-Signature: sha256=<hex hmac-sha256 of the raw body with
and header `X-OVMedia-Signature: sha256=<hex hmac-sha256 of the raw body with
the app's webhook_secret>`. 3 attempts with backoff, 10 s timeout.
the app's webhook_secret>`. 3 attempts with backoff, 10 s timeout.
+## Health, readiness and metrics
+
+| path | behavior |
+|---|---|
+| `GET /healthz` | liveness: the process answers (unchanged; it checks nothing else) |
+| `GET /api/ready` | readiness from real checks, each with `status`, `required`, `latency_ms`, `checked_at`. **Required** (503 when one fails): `db` (a query against the SQLite database), `storage_vods`, `storage_clips`, `storage_pastes`, `storage_thumbnails`, `storage_files`, `storage_objects` (a probe file is written and removed). **Optional** (listed in `degraded`, still 200): `network_jwks` (user JWTs and service tokens; app keys work without it), `remote_b2` / `remote_r2` (HeadBucket, at most once a minute; present only when the provider is configured), `events_outbox` (backlog over 1000 events). Also `recordings_in_progress`. |
+| `GET /metrics` | Prometheus text (openvibe-shared/metrics) for **direct loopback callers only**; 404 through nginx. HTTP golden signals by route template, process metrics, `release_info`, plus `media_recordings_in_progress`, `media_object_locations{provider,state}`, `media_objects{lifecycle_status}`, `media_events_outbox{status}` (only while the outbox runs). |
+
## Public serving (no auth unless the item is private)
## Public serving (no auth unless the item is private)
| path | behavior |
| path | behavior |
@@ -17,6 +17,9 @@ server {
@@ -17,6 +17,9 @@ server {
proxy_read_timeout 3600s;
proxy_read_timeout 3600s;
proxy_send_timeout 3600s;
proxy_send_timeout 3600s;
+ # Prometheus metrics are for loopback scrapers only (the app 404s proxied callers too).
+ location = /metrics { return 404; }
+
location / {
location / {
proxy_pass http://127.0.0.1:4100;
proxy_pass http://127.0.0.1:4100;
proxy_http_version 1.1;
proxy_http_version 1.1;
@@ -18,7 +18,7 @@
@@ -18,7 +18,7 @@
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.8.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.8.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.2.2",
"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.0",
+ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0",
"sharp": "^0.33.4"
"sharp": "^0.33.4"
}
}
},
},
@@ -1944,9 +1944,8 @@
@@ -1944,9 +1944,8 @@
}
}
},
},
"node_modules/openvibe-shared": {
"node_modules/openvibe-shared": {
- "version": "1.2.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.2.0",
- "integrity": "sha512-XYDP1I0f44T+ImWqcErW3P1bBVzsPZ8bgjYBk9QVJP6zDiK5/GYj5ynQhh5GhB9FhxvK92rFu7Lx1ZohZDNSkA==",
+ "version": "1.3.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0",
"license": "MIT",
"license": "MIT",
"dependencies": {
"dependencies": {
"jsonwebtoken": "^9.0.0"
"jsonwebtoken": "^9.0.0"
@@ -20,7 +20,7 @@
@@ -20,7 +20,7 @@
"dotenv": "^16.4.5",
"dotenv": "^16.4.5",
"express": "^4.19.2",
"express": "^4.19.2",
"multer": "^1.4.5-lts.1",
"multer": "^1.4.5-lts.1",
- "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.2.0",
+ "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.3.0",
"sharp": "^0.33.4"
"sharp": "^0.33.4"
}
}
}
}
@@ -60,6 +60,9 @@ function startJwksRefresh() {
@@ -60,6 +60,9 @@ function startJwksRefresh() {
if (_jwksTimer.unref) _jwksTimer.unref();
if (_jwksTimer.unref) _jwksTimer.unref();
}
}
+/** Whether the Network public key has loaded (user JWTs and service tokens verify only once it has). */
+function jwksLoaded() { return !!_networkPublicKeyPem; }
+
function stopJwksRefresh() {
function stopJwksRefresh() {
if (_jwksTimer) { clearInterval(_jwksTimer); _jwksTimer = null; }
if (_jwksTimer) { clearInterval(_jwksTimer); _jwksTimer = null; }
}
}
@@ -327,5 +330,6 @@ module.exports = {
@@ -327,5 +330,6 @@ module.exports = {
seedApps,
seedApps,
startJwksRefresh,
startJwksRefresh,
stopJwksRefresh,
stopJwksRefresh,
+ jwksLoaded,
fetchNetworkPublicKey,
fetchNetworkPublicKey,
};
};
@@ -27,6 +27,10 @@ auth.startJwksRefresh();
@@ -27,6 +27,10 @@ auth.startJwksRefresh();
const app = express();
const app = express();
app.set('trust proxy', true);
app.set('trust proxy', true);
+// Metrics first, so every route below is measured (GET /metrics: loopback callers only).
+const release = require('openvibe-shared/release').createRelease({ service: 'media', root: require('path').join(__dirname, '..') });
+const observability = require('./observability');
+const instrumented = observability.instrument(app, { release: release.release });
// Object API v2 content uploads read the raw request body, so they sit ahead of the body parsers.
// Object API v2 content uploads read the raw request body, so they sit ahead of the body parsers.
app.put('/api/v2/:app/objects/:id/content', ...require('./objects/routes').contentHandlers);
app.put('/api/v2/:app/objects/:id/content', ...require('./objects/routes').contentHandlers);
app.use(express.json({ limit: '2mb' }));
app.use(express.json({ limit: '2mb' }));
@@ -83,7 +87,6 @@ app.options(['/api/v1/:app/*', '/api/v2/:app/*'], (req, res) => {
@@ -83,7 +87,6 @@ app.options(['/api/v1/:app/*', '/api/v2/:app/*'], (req, res) => {
// ── Routes ───────────────────────────────────────────────────
// ── Routes ───────────────────────────────────────────────────
// What this server runs (ADR-016); the shared navbar's release-watch polls it.
// What this server runs (ADR-016); the shared navbar's release-watch polls it.
-const release = require('openvibe-shared/release').createRelease({ service: 'media', root: require('path').join(__dirname, '..') });
app.get('/release.json', release.handler);
app.get('/release.json', release.handler);
app.get('/healthz', (req, res) => {
app.get('/healthz', (req, res) => {
@@ -96,6 +99,17 @@ app.get('/healthz', (req, res) => {
@@ -96,6 +99,17 @@ app.get('/healthz', (req, res) => {
});
});
});
});
+// Readiness: required = database + writable storage; optional (degraded) = Network key, remote tiers, events outbox.
+{
+ const vodStorageForReady = require('./vod/vod-storage');
+ observability.mountReady(app, instrumented.registry, {
+ release: release.release, db, config, auth,
+ recorder: require('./vod/recorder'),
+ events: require('./events'),
+ remote: { configured: (n) => vodStorageForReady.providerConfigured(n), probe: (n) => vodStorageForReady.probeProvider(n) },
+ });
+}
+
// Per-app media stats (app-key only) — hero/dashboard counters in the owning app.
// Per-app media stats (app-key only) — hero/dashboard counters in the owning app.
app.get('/api/v1/:app/stats', auth.tenantAuth(), (req, res) => {
app.get('/api/v1/:app/stats', auth.tenantAuth(), (req, res) => {
try {
try {
new file mode 100644
new file mode 100644
@@ -0,0 +1,109 @@
@@ -0,0 +1,109 @@
+'use strict';
+/**
+ * Metrics and readiness for OpenVibe.Media (roadmap Track O, openvibe-shared/metrics + /ready).
+ *
+ * GET /metrics Prometheus text; direct loopback callers only (404 through nginx)
+ * GET /api/ready named checks; 503 only when a required one fails, optional failures → degraded
+ *
+ * Required (Media cannot serve without them): the SQLite database answers a real query, and every
+ * local storage directory accepts a write. Optional (one capability degrades, the rest keeps
+ * working): the Network public key (user-JWT and service-token routes; app-key tenants are
+ * unaffected), each CONFIGURED remote tier (B2 cold storage, R2 cache; a HeadBucket at most once a
+ * minute), and the Events outbox (webhooks still deliver when it backs up).
+ *
+ * `instrument()` must run before any route is mounted; `mountReady()` any time after.
+ */
+const fs = require('fs');
+const path = require('path');
+const crypto = require('crypto');
+const metrics = require('openvibe-shared/metrics');
+const { createReadiness } = require('openvibe-shared/ready');
+
+const OUTBOX_BACKLOG_LIMIT = 1000;
+
+function instrument(app, { release }) {
+ return metrics.instrument(app, {
+ service: 'media',
+ release,
+ // Static brand files: one label for the directory, never the file name.
+ normalize: (req) => (!req.route && req.path.startsWith('/assets/') ? '/assets/*' : null),
+ });
+}
+
+/** Write and remove a probe file: the directory exists and accepts writes. */
+function writable(dir) {
+ const probe = path.join(dir, `.ready-${process.pid}-${crypto.randomBytes(4).toString('hex')}`);
+ fs.writeFileSync(probe, 'ok');
+ fs.unlinkSync(probe);
+ return true;
+}
+
+function storageDirs(config) {
+ return {
+ vods: config.vod.path,
+ clips: config.vod.clipsPath,
+ pastes: config.pastes.path,
+ thumbnails: config.thumbnails.path,
+ files: config.files.path,
+ objects: config.objects.path,
+ };
+}
+
+function domainMetrics(registry, { db, recorder, events }) {
+ registry.gauge({ name: 'media_recordings_in_progress', help: 'Recordings ffmpeg is writing right now', collect: () => recorder.activeCount() });
+ registry.gauge({
+ name: 'media_object_locations', help: 'Stored copies of media objects by provider and verification state', labelNames: ['provider', 'state'],
+ collect: () => db.all('SELECT provider, state, COUNT(*) AS n FROM media_locations GROUP BY provider, state').map((r) => ({ labels: { provider: r.provider, state: r.state }, value: r.n })),
+ });
+ registry.gauge({
+ name: 'media_objects', help: 'Media objects by lifecycle status', labelNames: ['lifecycle_status'],
+ collect: () => db.all('SELECT lifecycle_status, COUNT(*) AS n FROM media_objects GROUP BY lifecycle_status').map((r) => ({ labels: { lifecycle_status: r.lifecycle_status }, value: r.n })),
+ });
+ registry.gauge({
+ name: 'media_events_outbox', help: 'Events outbox rows waiting to reach OpenVibe.Events, and rows Events rejected', labelNames: ['status'],
+ collect: () => {
+ const s = events.status();
+ if (!s.enabled) return null; // no outbox running: no series, never a made-up zero
+ return [{ labels: { status: 'pending' }, value: s.pending }, { labels: { status: 'rejected' }, value: s.rejected }];
+ },
+ });
+}
+
+function createMediaReadiness({ release, db, config, auth, recorder, events, remote }) {
+ const checks = [
+ { name: 'db', required: true, check: () => { const r = db.get('SELECT COUNT(*) AS n FROM apps'); return { ok: Number.isInteger(r.n), detail: { apps: r.n } }; } },
+ ];
+ for (const [name, dir] of Object.entries(storageDirs(config))) {
+ checks.push({ name: `storage_${name}`, required: true, check: () => writable(dir) });
+ }
+ checks.push({ name: 'network_jwks', required: false, description: 'user sign-in and service tokens (app keys work without it)', check: () => auth.jwksLoaded() || 'Network public key not loaded yet' });
+ for (const name of ['b2', 'r2']) {
+ if (!remote.configured(name)) continue; // not configured: no check, rather than a pretend pass
+ checks.push({ name: `remote_${name}`, required: false, cacheMs: 60 * 1000, timeoutMs: 5000, check: () => remote.probe(name) });
+ }
+ checks.push({
+ name: 'events_outbox', required: false, description: 'durable events to OpenVibe.Events (webhooks are separate)',
+ check: () => {
+ const s = events.status();
+ if (!s.enabled) return { ok: true, detail: { enabled: false } };
+ if (s.pending > OUTBOX_BACKLOG_LIMIT) return { ok: false, error: `${s.pending} events waiting (limit ${OUTBOX_BACKLOG_LIMIT})`, detail: { pending: s.pending } };
+ return { ok: true, detail: { pending: s.pending, rejected: s.rejected } };
+ },
+ });
+ return createReadiness({
+ service: 'media',
+ release,
+ checks,
+ details: () => ({ recordings_in_progress: recorder.activeCount() }),
+ });
+}
+
+/** Everything after instrument(): domain gauges and GET /api/ready. */
+function mountReady(app, registry, deps) {
+ domainMetrics(registry, deps);
+ const readiness = createMediaReadiness(deps);
+ app.get('/api/ready', readiness.handler);
+ return readiness;
+}
+
+module.exports = { instrument, mountReady, createMediaReadiness, domainMetrics, storageDirs };
@@ -1170,6 +1170,20 @@ async function bucketStatus() {
@@ -1170,6 +1170,20 @@ async function bucketStatus() {
return out;
return out;
}
}
+/** Live HeadBucket probe of one configured provider (readiness); throws when unreachable. */
+async function probeProvider(name) {
+ if (!providerConfigured(name)) throw new Error(`${name} is not configured`);
+ loadSdk();
+ try {
+ await clientFor(name).send(new S3.HeadBucketCommand({ Bucket: PROVIDER_ENV[name].bucket }));
+ providerHealthy[name] = true;
+ return true;
+ } catch (err) {
+ providerHealthy[name] = false;
+ throw new Error(`${name} HeadBucket failed: ${err.name || 'error'}`);
+ }
+}
+
function getStatus() {
function getStatus() {
const settings = getSettings();
const settings = getSettings();
const localDisk = diskUsage(config.vod.path);
const localDisk = diskUsage(config.vod.path);
@@ -1263,6 +1277,7 @@ module.exports = {
@@ -1263,6 +1277,7 @@ module.exports = {
stop,
stop,
getStatus,
getStatus,
bucketStatus,
bucketStatus,
+ probeProvider,
getBucketUsage,
getBucketUsage,
estimateCloudCosts,
estimateCloudCosts,
getSettings,
getSettings,
new file mode 100644
new file mode 100644
@@ -0,0 +1,154 @@
@@ -0,0 +1,154 @@
+'use strict';
+// GET /metrics (loopback only, route templates, Media gauges) and GET /api/ready (required: db +
+// writable storage; optional: Network key, configured remote tiers, events outbox), wired the way
+// server/index.js wires them — and then the real server booted once to prove index.js does.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+const net = require('net');
+const { spawn } = require('child_process');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-obs-'));
+const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+const ENV = {
+ DB_PATH: path.join(tmp, 'media.db'), VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), PASTES_PATH: dir('pastes'),
+ THUMBNAILS_PATH: dir('thumbnails'), FILES_PATH: dir('files'), OBJECTS_PATH: dir('objects'), ASSETS_PATH: dir('assets'),
+ MEDIA_PUBLIC_URL: 'https://media.test', OV_NETWORK_URL: 'http://127.0.0.1:9', RELEASE_COMMIT: 'abcdef1234567',
+};
+for (const k of ['MEDIA_B2_ENDPOINT', 'MEDIA_B2_BUCKET', 'MEDIA_R2_ENDPOINT', 'MEDIA_R2_BUCKET', 'EVENTS_URL']) process.env[k] = ''; // never a real tier or Events (dotenv keeps set values)
+Object.assign(process.env, ENV);
+
+const db = require('../server/db/database');
+const config = require('../server/config');
+const express = require('express');
+const observability = require('../server/observability');
+
+function request(base, p, headers = {}) {
+ return 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);
+ });
+}
+const listen = (app) => new Promise((resolve) => { const s = app.listen(0, '127.0.0.1', () => resolve(s)); });
+
+(async () => {
+ db.getDb();
+ db.upsertApp({ app_id: 'live', api_key: 'live-key' });
+
+ // ── Wiring as in index.js, with injectable dependencies ──
+ let jwks = true; let remoteUp = true; let remoteProbes = 0; let outbox = { enabled: true, pending: 2, rejected: 1 }; let recording = 1;
+ const app = express();
+ const inst = observability.instrument(app, { release: 'abcdef123456' });
+ app.use('/o', require('express').Router().get('/:id', (req, res) => res.status(404).json({ error: 'Not found' })));
+ observability.mountReady(app, inst.registry, {
+ release: 'abcdef123456', db, config,
+ auth: { jwksLoaded: () => jwks },
+ recorder: { activeCount: () => recording },
+ events: { status: () => outbox },
+ remote: { configured: (n) => n === 'b2', probe: async () => { remoteProbes++; if (!remoteUp) throw new Error('b2 HeadBucket failed: NetworkingError'); return true; } },
+ });
+ const server = await listen(app);
+ const base = `http://127.0.0.1:${server.address().port}`;
+
+ let r = await request(base, '/api/ready');
+ assert.strictEqual(r.status, 200, r.body);
+ let body = JSON.parse(r.body);
+ assert.strictEqual(body.ready, true);
+ assert.strictEqual(body.status, 'ready');
+ assert.strictEqual(body.service, 'media');
+ assert.strictEqual(body.recordings_in_progress, 1);
+ const names = Object.keys(body.checks).sort();
+ assert.deepStrictEqual(names, ['db', 'events_outbox', 'network_jwks', 'remote_b2', 'storage_clips', 'storage_files', 'storage_objects', 'storage_pastes', 'storage_thumbnails', 'storage_vods'], 'r2 is not configured, so it has no check');
+ for (const [n, c] of Object.entries(body.checks)) {
+ assert.strictEqual(c.status, 'ok', n);
+ assert.strictEqual(typeof c.latency_ms, 'number');
+ assert.ok(Date.parse(c.checked_at));
+ assert.strictEqual(c.required, n === 'db' || n.startsWith('storage_'), `${n} required?`);
+ }
+ assert.deepStrictEqual(body.checks.db.detail, { apps: 1 });
+ assert.deepStrictEqual(fs.readdirSync(config.vod.path), [], 'the write probe leaves nothing behind');
+
+ // Optional failures: degraded, still 200.
+ jwks = false; remoteUp = false; outbox = { enabled: true, pending: 5000, rejected: 0 };
+ r = await request(base, '/api/ready');
+ body = JSON.parse(r.body);
+ assert.strictEqual(r.status, 200);
+ assert.strictEqual(body.status, 'degraded');
+ assert.deepStrictEqual(body.degraded.sort(), ['events_outbox', 'network_jwks']);
+ assert.strictEqual(body.checks.remote_b2.status, 'ok', 'the remote tier probe is cached for a minute');
+ assert.strictEqual(remoteProbes, 1);
+
+ // A remote-tier failure on a fresh readiness (no cache): degraded, never "down".
+ const fresh = observability.createMediaReadiness({ release: 'x', db, config, auth: { jwksLoaded: () => true }, recorder: { activeCount: () => 0 }, events: { status: () => ({ enabled: false }) }, remote: { configured: () => true, probe: async (n) => { throw new Error(`${n} HeadBucket failed: AccessDenied`); } } });
+ body = await fresh.run();
+ assert.strictEqual(body.ready, true);
+ assert.deepStrictEqual(body.degraded, ['remote_b2', 'remote_r2']);
+ assert.strictEqual(body.checks.remote_r2.error, 'r2 HeadBucket failed: AccessDenied');
+ assert.deepStrictEqual(body.checks.events_outbox.detail, { enabled: false });
+
+ // A storage directory that cannot be written: not ready, 503, named.
+ jwks = true; outbox = { enabled: false };
+ fs.rmSync(config.vod.clipsPath, { recursive: true });
+ fs.writeFileSync(config.vod.clipsPath, 'not a directory');
+ r = await request(base, '/api/ready');
+ body = JSON.parse(r.body);
+ assert.strictEqual(r.status, 503);
+ assert.strictEqual(body.ready, false);
+ assert.deepStrictEqual(body.failed, ['storage_clips']);
+ fs.rmSync(config.vod.clipsPath); fs.mkdirSync(config.vod.clipsPath);
+
+ // ── /metrics ──
+ await request(base, '/o/med_01J9ZX4YQ0ABCDEFGHJKMNPQRS');
+ await request(base, '/assets/icon-192.png');
+ recording = 2;
+ r = await request(base, '/metrics');
+ assert.strictEqual(r.status, 200);
+ assert.ok(r.body.includes('release_info{service="media",release="abcdef123456"} 1\n'));
+ assert.ok(r.body.includes('http_requests_total{method="GET",route="/o/:id",status_class="4xx"} 1\n'), 'object ids never reach a label');
+ assert.ok(r.body.includes('http_requests_total{method="GET",route="/assets/*",status_class="4xx"} 1\n'));
+ assert.ok(r.body.includes('http_requests_total{method="GET",route="/api/ready",status_class="2xx"}'));
+ assert.ok(!r.body.includes('med_01J9'));
+ assert.ok(r.body.includes('media_recordings_in_progress 2\n'));
+ assert.ok(r.body.includes('# TYPE media_object_locations gauge'));
+ assert.ok(r.body.includes('# TYPE media_objects gauge'));
+ assert.ok(!r.body.includes('media_events_outbox{'), 'no outbox running: no invented series');
+ r = await request(base, '/metrics', { 'X-Forwarded-For': '203.0.113.7' });
+ assert.strictEqual(r.status, 404, 'proxied callers never see metrics');
+ server.close();
+ inst.stop();
+
+ // ── The real index.js boots with both routes ──
+ const port = await new Promise((resolve) => { const s = net.createServer().listen(0, '127.0.0.1', () => { const p = s.address().port; s.close(() => resolve(p)); }); });
+ const bootDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-obs-boot-'));
+ const env = { ...process.env, ...ENV, DB_PATH: path.join(bootDir, 'media.db'), PORT: String(port), HOST: '127.0.0.1', NODE_ENV: 'test' };
+ for (const k of ['VOD_PATH', 'CLIPS_PATH', 'PASTES_PATH', 'THUMBNAILS_PATH', 'FILES_PATH', 'OBJECTS_PATH', 'ASSETS_PATH']) env[k] = path.join(bootDir, k.toLowerCase());
+ const child = spawn(process.execPath, [path.join(__dirname, '..', 'server', 'index.js')], { env, stdio: ['ignore', 'pipe', 'pipe'] });
+ let log = ''; child.stdout.on('data', (d) => { log += d; }); child.stderr.on('data', (d) => { log += d; });
+ const b2 = `http://127.0.0.1:${port}`;
+ let up = null;
+ for (let i = 0; i < 100 && !up; i++) {
+ await new Promise((res) => setTimeout(res, 100));
+ try { up = await request(b2, '/api/ready'); } catch { /* not listening yet */ }
+ }
+ try {
+ assert.ok(up, `server did not start:\n${log}`);
+ body = JSON.parse(up.body);
+ assert.strictEqual(up.status, 200, up.body);
+ assert.strictEqual(body.release, 'abcdef123456');
+ assert.ok(body.checks.db && body.checks.storage_objects);
+ assert.ok(!body.checks.remote_b2, 'unconfigured tiers are not checked');
+ const m = await request(b2, '/metrics');
+ assert.strictEqual(m.status, 200);
+ assert.ok(m.body.includes('http_requests_total{method="GET",route="/api/ready",status_class="2xx"} 1\n'), m.body.slice(0, 400));
+ const h = await request(b2, '/healthz');
+ assert.strictEqual(h.status, 200, '/healthz still answers');
+ } finally {
+ child.kill('SIGKILL');
+ }
+ console.log('observability: all checks passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });