openvibers/openvibe.zone Commits
Zone: Media's per-zone rollups (media.usage.recorded) arrive signed and sum into the usage series and the console; contracts 0.175.0 (#2)
Changes
@@@ -30,6 +30,10 @@ OV_OAUTH_CLIENT_SECRET=# OpenVibe.Media: Zone provisions each zone there and asks it for usage, deletion progress and access keys (loopback,# with Zone's own client-credentials token: grants media.zone.provision and media.access_key.manage).MEDIA_INTERNAL_URL=http://127.0.0.1:4100++# OpenVibe.Events: Media's per-zone rollups (media.usage.recorded) are delivered to POST /internal/events, signed with++# ZONE_EVENTS_SECRET (32+ characters; comma-separated to rotate). The subscription is created at boot when both are set.++EVENTS_URL=++ZONE_EVENTS_SECRET=# The S3 endpoint the console and the docs name (s3.openvibe.media once its DNS record exists).ZONE_S3_ENDPOINT=https://s3.openvibe.media# The zones one project environment may hold besides its default.@@@ -19,6 +19,11 @@ bucket at `https://s3.openvibe.media`, path style.Zone never lists keys and never holds bytes.++**Usage.** Zone never counts. Media sends each zone's closed hours as `media.usage.recorded`: requests by capability,++read egress, and stored bytes as byte-hours. Zone keeps them (`zone_usage`, delivered signed to `POST /internal/events`,++loopback only). `GET /zones/:zone_id/usage` sums the current UTC month beside Media's live snapshot, and the console++shows each zone's requests and egress so far this month.++## What it does- **Default zones.** Every project environment has a `default` zone. Its namespace is the environment root, so it holds the environment's existing objects. Zone creates it the first time the environment is used, and it cannot be deleted on its own.nnew file mode 100644@@@ -0,0 +1,15 @@++-- phase: expand++-- A zone's metered usage (ADR-031 section 8, server/zone/usage.js): the hourly rollups Media sends as++-- media.usage.recorded, one row per zone, hour and series. Zone never counts; it keeps what Media reports and sums it.++CREATE TABLE zone_usage (++ zone_id text COLLATE "C" NOT NULL,++ window_start bigint NOT NULL, -- ms since the epoch, on the hour (UTC)++ capability text COLLATE "C" NOT NULL, -- media.object.upload | read | list | delete++ unit text COLLATE "C" NOT NULL, -- byte_hours | requests | egress_bytes++ quantity bigint NOT NULL,++ errors bigint NOT NULL DEFAULT 0,++ revision integer NOT NULL,++ event_id text COLLATE "C" NOT NULL,++ recorded_at bigint NOT NULL,++ PRIMARY KEY (zone_id, window_start, capability, unit)++);@@@ -16,7 +16,7 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.173.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.175.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.4","pg": "^8.23.0"@@@ -872,9 +872,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.173.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.173.0",-- "integrity": "sha512-6zHbU9zghIYDd3PPGtgMTsIgxrWwsoBHO/rvuLW1rghr6kRK6VwCmSMZRcjQbcPODPH1XrUAFuLg/T1K4nOCMQ==",++ "version": "0.175.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.175.0",++ "integrity": "sha512-AJNkb/g7mMQDi1oGPiKpVVEmk+oGvzJYLsgxCzd7bPkxN7aos3y/VKh7+tAlfnAOKTajpOTAz5cvLXcGJQHiiQ==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -23,7 +23,7 @@"express-rate-limit": "^7.5.0","helmet": "^8.1.0","iovalkey": "^0.4.0",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.173.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.175.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.4","pg": "^8.23.0"@@@ -26,6 +26,8 @@ const { createKeyStore } = require('./auth/keys');const { createSso } = require('./auth/sso');const { createPrincipal } = require('./http/principal');const { createApi } = require('./http/api');++const { createUsage } = require('./zone/usage');++const { eventsIntake } = require('./http/events');const { createPageRoutes } = require('./http/pages');const { createServiceReadiness } = require('./observability');const { createCallerLimits } = require('./http/caller-limits');@@@ -51,8 +53,9 @@ async function createApp(opts = {}) {const sso = createSso({ config, keys, fetchImpl, now: s.now, log });const principal = createPrincipal({ config, keys });const clients = opts.clients || createClients({ config, fetchImpl });-- const zones = createZones({ s, clients, config, log });-- const ctx = { config, s, keys, sso, principal, log, clients, zones };++ const usage = createUsage({ s, log });++ const zones = createZones({ s, clients, config, log, metering: usage });++ const ctx = { config, s, keys, sso, principal, log, clients, zones, usage };const app = express();app.disable('x-powered-by');@@@ -91,6 +94,10 @@ async function createApp(opts = {}) {}));app.use(cookieParser());++ // ── OpenVibe.Events → Zone (loopback only: nginx answers 404 for /internal/) ──++ // media.usage.recorded: Media's hourly rollups of each zone, kept for the usage view (server/zone/usage.js).++ app.post('/internal/events', eventsIntake({ secrets: config.events.secrets, usage, log }));++// ── Machine endpoints ───────────────────────────────────app.get('/api/health', (_req, res) => res.json({ status: 'ok', service: 'openvibe-zone', version: VERSION }));release.mount(app, { registry: metrics.registry });@@@ -60,6 +60,13 @@ function load(env = process.env) {internalUrl: trim(env.MEDIA_INTERNAL_URL || 'http://127.0.0.1:4100'),s3Endpoint: trim(env.ZONE_S3_ENDPOINT || 'https://s3.openvibe.media'),},++ // OpenVibe.Events: ZONE_EVENTS_SECRET signs the deliveries to POST /internal/events (Media's per-zone rollups,++ // media.usage.recorded; comma-separated for rotation, 32+ characters each). Unset, the route answers 503 and no++ // subscription is created at boot.++ events: {++ url: trim(env.EVENTS_URL || ''),++ secrets: String(env.ZONE_EVENTS_SECRET || '').split(',').map((x) => x.trim()).filter(Boolean),++ },// The zones one project environment may hold besides its default.maxZonesPerEnv: Math.max(1, int(env.ZONE_MAX_PER_ENV, 20)),};nnew file mode 100644@@@ -0,0 +1,43 @@++'use strict';++/**++ * POST /internal/events: OpenVibe.Events delivers Zone's subscriptions here (media.usage.recorded). No body parser runs++ * before it: the v2 signature covers the raw body. Loopback only: a request that came through a proxy is refused.++ * 2xx once applied (or ignored with a reason); anything else makes Events retry.++ */++const { parseDelivery } = require('openvibe-sdk/events');++++const LIMIT = 256 * 1024;++++function eventsIntake({ secrets = [], usage, log = console, now = () => Date.now() }) {++ const keys = (secrets || []).filter((x) => typeof x === 'string' && x.length >= 32);++ const reply = (res, status, body) => {++ res.status(status).set('Cache-Control', 'no-store').type(status < 300 ? 'application/json' : 'application/problem+json').send(JSON.stringify(body));++ };++ const problem = (res, status, code, detail) => reply(res, status, { type: 'about:blank', title: code, status, code, detail });++ const readRaw = (req) => new Promise((resolve, reject) => {++ const chunks = [];++ let size = 0;++ req.on('data', (c) => { size += c.length; if (size > LIMIT) { reject(Object.assign(new Error('too large'), { status: 413 })); req.destroy(); } else chunks.push(c); });++ req.on('end', () => resolve(Buffer.concat(chunks)));++ req.on('error', reject);++ });++ return async function handle(req, res) {++ if (req.headers['x-forwarded-for'] || req.headers['x-real-ip'] || req.headers['cf-connecting-ip']) return problem(res, 403, 'zone.internal_only', 'internal route');++ if (!keys.length) return problem(res, 503, 'zone.events_disabled', 'no events secret is configured');++ let raw;++ try { raw = await readRaw(req); } catch (err) { return problem(res, err.status || 400, 'bad_request', 'unreadable body'); }++ let delivery = null;++ for (const k of keys) { delivery = parseDelivery(raw, req.headers, k, { requireV2: true, now: now() }); if (delivery) break; }++ if (!delivery || !delivery.event || typeof delivery.event.event_type !== 'string') return problem(res, 401, 'zone.bad_signature', 'X-OpenVibe-Signature-V2 does not verify or is outside the replay window');++ const ev = delivery.event;++ try {++ const outcome = ev.event_type === usage.TOPIC ? await usage.apply(ev) : 'ignored:type';++ return reply(res, 200, { event_id: ev.event_id, outcome });++ } catch (err) {++ log.warn(`[Events] ${ev.event_type} ${ev.event_id} failed: ${(err && err.message) || err}`);++ return problem(res, 503, 'zone.retry', 'not applied; Events will retry');++ }++ };++}++++module.exports = { eventsIntake };@@@ -48,7 +48,7 @@ const plural = (n, one, many) => `${Number(n).toLocaleString('en-US')} ${n === 1const day = (iso) => (iso ? String(iso).slice(0, 10) : '—');function createPageRoutes(ctx) {-- const { config, clients, zones } = ctx;++ const { config, clients, zones, usage: metering } = ctx;const r = asyncRouter();const form = express.urlencoded({ extended: false, limit: '8kb' });const PUBLIC_CACHE = cache.htmlHeaders({ maxAge: 300 });@@@ -144,10 +144,21 @@ ${projects.length ? html`<ul class="zn-cards">${projects.map((p) => html`<li clafor (const env of envs) await zones.ensureDefault(project, env);const list = (await zones.list(project, {})).zones;const usage = new Map();++ const metered = new Map();++ const monthStart = (() => { const d = new Date(); return Date.UTC(d.getUTCFullYear(), d.getUTCMonth(), 1); })();await Promise.all(list.filter((z) => z.state !== 'deleted').map(async (z) => {const rep = await clients.report(z.id).catch(() => null);if (rep) usage.set(z.id, rep);++ if (metering) metered.set(z.id, (await metering.series(z.id, { from: monthStart }).catch(() => ({ series: [] }))).series);}));++ // This month's requests and egress, as Media reported them by the hour (media.usage.recorded).++ const monthLine = (id) => {++ const list = metered.get(id) || [];++ const sum = (pred) => list.filter(pred).reduce((n, x) => n + x.quantity, 0);++ const requests = sum((x) => x.unit === 'requests');++ const egress = sum((x) => x.unit === 'egress_bytes');++ return requests || egress ? `this month: ${plural(requests, 'request', 'requests')} · ${bytes(egress)} out` : '';++ };let keys = [];let keysError = null;if (admin) { try { keys = await clients.listKeys(project, me.subject); } catch (err) { keysError = err.message; } }@@@ -155,7 +166,7 @@ ${projects.length ? html`<ul class="zn-cards">${projects.map((p) => html`<li claconst rep = usage.get(z.id);const u = rep && rep.usage;return html`<tr id="${z.id}"><td><code>${z.name}</code>${z.default ? html` <span class="zn-badge">default</span>` : ''}</td><td>${z.state}</td>--<td class="zn-n">${u ? `${bytes(u.stored_bytes)} · ${plural(u.objects, 'object', 'objects')}` : '—'}</td>++<td class="zn-n">${u ? `${bytes(u.stored_bytes)} · ${plural(u.objects, 'object', 'objects')}` : '—'}${monthLine(z.id) ? html`<br><span class="zn-muted">${monthLine(z.id)}</span>` : ''}</td><td class="zn-n">${z.quota_bytes ? bytes(z.quota_bytes) : (z.default ? 'environment' : '—')}</td><td>${day(z.created_at)}</td><td>${admin && !z.default && z.state === 'active' ? html`<details><summary>Delete</summary><form method="post" action="/p/${project}/zones/${z.id}/delete"><p>This deletes every object in <code>${z.name}</code>, with no undo, and the name can never be used again.</p>@@@ -289,7 +300,7 @@ ${keysSection}<li><code>GET /zones/:zone_id</code> (<code>zone.zone.read</code>): the zone with its stored bytes and objects.</li><li><code>PATCH /zones/:zone_id</code> (<code>zone.zone.manage</code>): <code>{ description?, quota_bytes? }</code>.</li><li><code>DELETE /zones/:zone_id?recursive=true</code> (<code>zone.zone.delete</code>). Without <code>recursive</code>, a zone that still holds objects answers <code>zone.not_empty</code>. A zone holding an object under a retention hold answers <code>zone.held</code>.</li>--<li><code>GET /zones/:zone_id/usage</code> (<code>zone.usage.read</code>).</li>++<li><code>GET /zones/:zone_id/usage</code> (<code>zone.usage.read</code>): Media's live snapshot, and this month's metered series (requests, egress, byte-hours) as Media reports each closed hour.</li></ul><p>The objects of a zone are OpenVibe.Media objects. Media's own API (<code>/api/v2/:project_id/objects</code>, with <code>namespace</code> set to the zone's id) reads and writes the same objects.</p>`,}));@@@ -6,6 +6,7 @@*/const { createApp } = require('./app');const { gracefulStop } = require('openvibe-sdk/service');++const { startSubscriptions } = require('openvibe-sdk/account-data');/*** The process stop (openvibe-sdk/service): the HTTP drain runs, then the timers clear, the zone job stops, the JWKS@@@ -30,7 +31,14 @@ async function start() {ctx.keys.client.start();// The zone job (zone/zones.js tick): pushes what Media has not acknowledged and finishes deletions, every 15 s.ctx.zones.start();-- createLifecycle({ server, ctx, extra: [() => ctx.zones.stop()] });++ // Media's per-zone rollups (media.usage.recorded) at OpenVibe.Events, created when missing; off without EVENTS_URL,++ // ZONE_EVENTS_SECRET or the client secret.++ const subscriptions = startSubscriptions({++ eventsUrl: config.events.url, endpoint: `http://127.0.0.1:${config.port}/internal/events`, secret: config.events.secrets[0],++ topics: [ctx.usage.TOPIC],++ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,++ });++ createLifecycle({ server, ctx, extra: [() => ctx.zones.stop(), () => { if (subscriptions) subscriptions.stop(); }] });return { server, ctx };}nnew file mode 100644@@@ -0,0 +1,67 @@++'use strict';++/**++ * A zone's metered usage (ADR-031 section 8): the hourly rollups Media sends as media.usage.recorded++ * (common.usage-recorded@1), kept per zone, hour and series. Zone never counts: it stores what Media reports and sums++ * it for GET /zones/:zone/usage and the console. Media is the only source; a rollup names its zone twice (resource is++ * the zone's OVRN, dimension its id in lowercase) and both must name a zone of its project here.++ *++ * apply(event) 'recorded' or an 'ignored:*' outcome (a newer or equal revision replaces the hour)++ * series(zoneId, { from }) [{ capability, unit, quantity }] summed from `from` (ms), and the last reported hour's end++ * prune() hours older than KEEP_DAYS++ */++const contracts = require('openvibe-contracts');++++const TOPIC = 'media.usage.recorded';++const HOUR_MS = 3600 * 1000;++const KEEP_DAYS = 400;++const OVRN = /^ovrn:zone:(prj_[0-9A-HJKMNP-TV-Z]{26}):object-zone\/(zon_[0-9A-HJKMNP-TV-Z]{26})$/;++++function createUsage({ s, log = console }) {++ const { db } = s;++++ async function apply(event) {++ if (!event || event.event_type !== TOPIC) return 'ignored:type';++ if (event.source !== 'media') return 'ignored:source';++ const p = event.payload && typeof event.payload === 'object' ? event.payload : {};++ const v = contracts.validate(`${TOPIC}@${event.version || 1}`, p);++ if (!v.valid) {++ log.warn(`[Usage] ${event.event_id} does not match ${TOPIC}: ${v.errors.slice(0, 3).map((e) => `${e.path} ${e.message}`).join('; ')}`);++ return 'ignored:payload';++ }++ const subject = event.subject || {};++ if (subject.type !== 'project' || subject.id !== p.project_id) return 'ignored:subject';++ const m = OVRN.exec(String(p.resource || ''));++ if (!m || m[1] !== p.project_id || String(p.dimension || '') !== m[2].toLowerCase()) return 'ignored:resource';++ const zone = await db.maybe('SELECT id, project_id, env FROM zone_zones WHERE id = $1', [m[2]]);++ if (!zone || zone.project_id !== p.project_id || zone.env !== p.env) return 'ignored:zone';++ const start = Date.parse(p.window_start);++ if (p.window !== 'hour' || Date.parse(p.window_end) - start !== HOUR_MS || start % HOUR_MS !== 0) return 'ignored:window';++ if (start > s.now() + HOUR_MS) return 'ignored:future';++ const revision = p.revision || 1;++ const row = await db.maybe(`INSERT INTO zone_usage (zone_id, window_start, capability, unit, quantity, errors, revision, event_id, recorded_at)++ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)++ ON CONFLICT (zone_id, window_start, capability, unit) DO UPDATE SET quantity = excluded.quantity, errors = excluded.errors,++ revision = excluded.revision, event_id = excluded.event_id, recorded_at = excluded.recorded_at++ WHERE zone_usage.revision <= excluded.revision++ RETURNING zone_id`, [zone.id, start, p.capability, p.unit, p.quantity, p.errors, revision, event.event_id, s.now()]);++ return row ? 'recorded' : 'ignored:stale';++ }++++ async function series(zoneId, { from }) {++ const rows = await db.many(`SELECT capability, unit, SUM(quantity) AS quantity FROM zone_usage++ WHERE zone_id = $1 AND window_start >= $2 GROUP BY capability, unit ORDER BY capability, unit`, [zoneId, from]);++ const last = await db.value('SELECT MAX(window_start) FROM zone_usage WHERE zone_id = $1 AND window_start >= $2', [zoneId, from]);++ return {++ series: rows.slice(0, 32).map((r) => ({ capability: r.capability, unit: r.unit, quantity: Number(r.quantity) })),++ reportedUntil: last == null ? null : Number(last) + HOUR_MS,++ };++ }++++ async function prune() {++ await db.exec('DELETE FROM zone_usage WHERE window_start < $1', [s.now() - KEEP_DAYS * 24 * HOUR_MS]);++ }++++ return { apply, series, prune, TOPIC };++}++++module.exports = { createUsage, TOPIC };@@@ -28,7 +28,7 @@ class ZoneError extends Error {const fail = (status, code, detail) => { throw new ZoneError(status, code, detail); };const iso = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());--function createZones({ s, clients, config, log = console }) {++function createZones({ s, clients, config, log = console, metering = null }) {const { db } = s;/** The zone.object-zone@1 of a row (usage when given; deletion progress while it is deleting or deleted). */@@@ -198,15 +198,19 @@ function createZones({ s, clients, config, log = console }) {const now = new Date(s.now());const start = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), 1));const current = rep && rep.usage ? rep.usage : { stored_bytes: 0, objects: 0, as_of: now.toISOString() };++ // The series sums Media's hourly rollups (media.usage.recorded) of this month; window_end is the end of the++ // last hour Media reported (the month's start when none has arrived yet).++ const metered = metering ? await metering.series(row.id, { from: start.getTime() }) : { series: [], reportedUntil: null };const out = {zone_id: row.id, ovrn: contracts.zones.identity({ ...row, default: row.is_default }).ovrn,current: { stored_bytes: current.stored_bytes, objects: current.objects, as_of: current.as_of },-- window_start: start.toISOString(), window_end: now.toISOString(), series: [],++ window_start: start.toISOString(), window_end: new Date(metered.reportedUntil || start.getTime()).toISOString(), series: metered.series,};return out;}let ticking = false;++ let lastPrune = 0;async function tick() {if (ticking) return { pushed: 0, deleted: 0 };ticking = true;@@@ -224,6 +228,7 @@ function createZones({ s, clients, config, log = console }) {}}}++ if (metering && s.now() - lastPrune > 3600 * 1000) { lastPrune = s.now(); await metering.prune(); }} finally { ticking = false; }return { pushed, deleted };}@@@ -62,7 +62,8 @@ function fakeMedia(network) {const { startNetwork } = require('./helpers/mocks');const net = await startNetwork();let media;-- const t = await boot({ clients: new Proxy({}, { get: (_o, k) => (...a) => media[k](...a) }) });++ const EVENTS_SECRET = 'e'.repeat(40); // a test value: 32+ characters++ const t = await boot({ clients: new Proxy({}, { get: (_o, k) => (...a) => media[k](...a) }), env: { ZONE_EVENTS_SECRET: EVENTS_SECRET } });await net.close();media = fakeMedia(t.network);const owner = t.network.addUser('owen');@@@ -157,6 +158,46 @@ function fakeMedia(network) {assert.deepStrictEqual(z.usage.objects, 3);});++ check('Media\'s hourly rollups (media.usage.recorded) arrive signed and sum into the usage series and the console', async () => {++ const { signDeliveryHeaders } = require('openvibe-sdk/events');++ const { ids } = contracts;++ const HOUR = 3600 * 1000;++ const hour = Math.floor(Date.now() / HOUR) * HOUR - HOUR;++ const ovrn = `ovrn:zone:${project}:object-zone/${assets.id}`;++ const def = (await t.get(`/api/v1/projects/${project}/zones?env=${assets.env}`, { as: owner })).json().zones.find((z) => z.default);++ const rollup = (payload, over = {}) => ({++ event_id: ids.newId('event'), event_type: 'media.usage.recorded', version: 1, source: 'media', actor: { type: 'service', id: 'media' },++ timestamp: new Date().toISOString(), subject: { type: 'project', id: project }, visibility: 'internal', priority: 'low',++ payload: { project_id: project, env: assets.env, resource: ovrn, dimension: assets.id.toLowerCase(), window: 'hour',++ window_start: new Date(hour).toISOString(), window_end: new Date(hour + HOUR).toISOString(), errors: 0, ...payload }, ...over,++ });++ const deliver = async (ev, { secret = EVENTS_SECRET, headers = {} } = {}) => {++ const raw = JSON.stringify({ event: ev });++ const res = await fetch(`${t.base}/internal/events`, { method: 'POST', headers: { 'content-type': 'application/json', ...signDeliveryHeaders(raw, secret), ...headers }, body: raw });++ return { status: res.status, body: await res.json() };++ };++ let r = await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 10 }));++ assert.deepStrictEqual([r.status, r.body.outcome], [200, 'recorded'], JSON.stringify(r.body));++ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'egress_bytes', quantity: 2048 }))).body.outcome, 'recorded');++ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 12, revision: 2 }))).body.outcome, 'recorded', 'a corrected hour replaces it');++ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 1, revision: 1 }))).body.outcome, 'ignored:stale', 'an older revision never does');++ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 1 }, { source: 'tools' }))).body.outcome, 'ignored:source', 'only Media reports');++ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 1, dimension: def.id.toLowerCase() }))).body.outcome, 'ignored:resource', 'resource and dimension name one zone');++ const elsewhere = `zon_${ids.ulid()}`;++ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 1, resource: `ovrn:zone:${project}:object-zone/${elsewhere}`, dimension: elsewhere.toLowerCase() }))).body.outcome, 'ignored:zone', 'a zone Zone does not have');++ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 1, env: assets.env === 'production' ? 'sandbox' : 'production' }))).body.outcome, 'ignored:zone', 'or of another environment');++ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 1 }), { secret: 'x'.repeat(40) })).status, 401, 'unsigned deliveries are refused');++ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 1 }), { headers: { 'x-forwarded-for': '203.0.113.9' } })).status, 403, 'and so is anything through a proxy');++ r = await t.get(`/api/v1/zones/${assets.id}/usage`, { as: viewer });++ const u = r.json();++ assert.ok(contracts.zones.checkObjectZoneUsage(u).valid, JSON.stringify(contracts.zones.checkObjectZoneUsage(u).errors));++ const q = (unit) => (u.series.find((x) => x.capability === 'media.object.read' && x.unit === unit) || {}).quantity;++ assert.deepStrictEqual([q('requests'), q('egress_bytes')], [12, 2048], 'the series sums the month\'s hours');++ assert.strictEqual(u.window_end, new Date(hour + HOUR).toISOString(), 'up to the last hour Media reported');++ r = await t.get(`/p/${project}`, { as: owner });++ assert.ok(r.text.includes('this month: 12 requests · 2.0 KB out'), 'the console shows the month\'s requests and egress');++ });++check('deletion: not empty, held, then recursive; the job marks it deleted and its name is retired', async () => {let r = await t.get(`/api/v1/zones/${assets.id}`, { as: owner, method: 'DELETE', headers: same });assert.deepStrictEqual([r.status, r.json().code], [409, 'zone.not_empty']);
@@ -30,6 +30,10 @@ OV_OAUTH_CLIENT_SECRET=
@@ -30,6 +30,10 @@ OV_OAUTH_CLIENT_SECRET=
# OpenVibe.Media: Zone provisions each zone there and asks it for usage, deletion progress and access keys (loopback,
# OpenVibe.Media: Zone provisions each zone there and asks it for usage, deletion progress and access keys (loopback,
# with Zone's own client-credentials token: grants media.zone.provision and media.access_key.manage).
# with Zone's own client-credentials token: grants media.zone.provision and media.access_key.manage).
MEDIA_INTERNAL_URL=http://127.0.0.1:4100
MEDIA_INTERNAL_URL=http://127.0.0.1:4100
+# OpenVibe.Events: Media's per-zone rollups (media.usage.recorded) are delivered to POST /internal/events, signed with
+# ZONE_EVENTS_SECRET (32+ characters; comma-separated to rotate). The subscription is created at boot when both are set.
+EVENTS_URL=
+ZONE_EVENTS_SECRET=
# The S3 endpoint the console and the docs name (s3.openvibe.media once its DNS record exists).
# The S3 endpoint the console and the docs name (s3.openvibe.media once its DNS record exists).
ZONE_S3_ENDPOINT=https://s3.openvibe.media
ZONE_S3_ENDPOINT=https://s3.openvibe.media
# The zones one project environment may hold besides its default.
# The zones one project environment may hold besides its default.
@@ -19,6 +19,11 @@ bucket at `https://s3.openvibe.media`, path style.
@@ -19,6 +19,11 @@ bucket at `https://s3.openvibe.media`, path style.
Zone never lists keys and never holds bytes.
Zone never lists keys and never holds bytes.
+**Usage.** Zone never counts. Media sends each zone's closed hours as `media.usage.recorded`: requests by capability,
+read egress, and stored bytes as byte-hours. Zone keeps them (`zone_usage`, delivered signed to `POST /internal/events`,
+loopback only). `GET /zones/:zone_id/usage` sums the current UTC month beside Media's live snapshot, and the console
+shows each zone's requests and egress so far this month.
+
## What it does
## What it does
- **Default zones.** Every project environment has a `default` zone. Its namespace is the environment root, so it holds the environment's existing objects. Zone creates it the first time the environment is used, and it cannot be deleted on its own.
- **Default zones.** Every project environment has a `default` zone. Its namespace is the environment root, so it holds the environment's existing objects. Zone creates it the first time the environment is used, and it cannot be deleted on its own.
new file mode 100644
new file mode 100644
@@ -0,0 +1,15 @@
@@ -0,0 +1,15 @@
+-- phase: expand
+-- A zone's metered usage (ADR-031 section 8, server/zone/usage.js): the hourly rollups Media sends as
+-- media.usage.recorded, one row per zone, hour and series. Zone never counts; it keeps what Media reports and sums it.
+CREATE TABLE zone_usage (
+ zone_id text COLLATE "C" NOT NULL,
+ window_start bigint NOT NULL, -- ms since the epoch, on the hour (UTC)
+ capability text COLLATE "C" NOT NULL, -- media.object.upload | read | list | delete
+ unit text COLLATE "C" NOT NULL, -- byte_hours | requests | egress_bytes
+ quantity bigint NOT NULL,
+ errors bigint NOT NULL DEFAULT 0,
+ revision integer NOT NULL,
+ event_id text COLLATE "C" NOT NULL,
+ recorded_at bigint NOT NULL,
+ PRIMARY KEY (zone_id, window_start, capability, unit)
+);
@@ -16,7 +16,7 @@
@@ -16,7 +16,7 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.173.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.175.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.4",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.4",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -872,9 +872,9 @@
@@ -872,9 +872,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.173.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.173.0",
- "integrity": "sha512-6zHbU9zghIYDd3PPGtgMTsIgxrWwsoBHO/rvuLW1rghr6kRK6VwCmSMZRcjQbcPODPH1XrUAFuLg/T1K4nOCMQ==",
+ "version": "0.175.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.175.0",
+ "integrity": "sha512-AJNkb/g7mMQDi1oGPiKpVVEmk+oGvzJYLsgxCzd7bPkxN7aos3y/VKh7+tAlfnAOKTajpOTAz5cvLXcGJQHiiQ==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -23,7 +23,7 @@
@@ -23,7 +23,7 @@
"express-rate-limit": "^7.5.0",
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
"iovalkey": "^0.4.0",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.173.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.175.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.44.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.4",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.1.4",
"pg": "^8.23.0"
"pg": "^8.23.0"
@@ -26,6 +26,8 @@ const { createKeyStore } = require('./auth/keys');
@@ -26,6 +26,8 @@ const { createKeyStore } = require('./auth/keys');
const { createSso } = require('./auth/sso');
const { createSso } = require('./auth/sso');
const { createPrincipal } = require('./http/principal');
const { createPrincipal } = require('./http/principal');
const { createApi } = require('./http/api');
const { createApi } = require('./http/api');
+const { createUsage } = require('./zone/usage');
+const { eventsIntake } = require('./http/events');
const { createPageRoutes } = require('./http/pages');
const { createPageRoutes } = require('./http/pages');
const { createServiceReadiness } = require('./observability');
const { createServiceReadiness } = require('./observability');
const { createCallerLimits } = require('./http/caller-limits');
const { createCallerLimits } = require('./http/caller-limits');
@@ -51,8 +53,9 @@ async function createApp(opts = {}) {
@@ -51,8 +53,9 @@ async function createApp(opts = {}) {
const sso = createSso({ config, keys, fetchImpl, now: s.now, log });
const sso = createSso({ config, keys, fetchImpl, now: s.now, log });
const principal = createPrincipal({ config, keys });
const principal = createPrincipal({ config, keys });
const clients = opts.clients || createClients({ config, fetchImpl });
const clients = opts.clients || createClients({ config, fetchImpl });
- const zones = createZones({ s, clients, config, log });
- const ctx = { config, s, keys, sso, principal, log, clients, zones };
+ const usage = createUsage({ s, log });
+ const zones = createZones({ s, clients, config, log, metering: usage });
+ const ctx = { config, s, keys, sso, principal, log, clients, zones, usage };
const app = express();
const app = express();
app.disable('x-powered-by');
app.disable('x-powered-by');
@@ -91,6 +94,10 @@ async function createApp(opts = {}) {
@@ -91,6 +94,10 @@ async function createApp(opts = {}) {
}));
}));
app.use(cookieParser());
app.use(cookieParser());
+ // ── OpenVibe.Events → Zone (loopback only: nginx answers 404 for /internal/) ──
+ // media.usage.recorded: Media's hourly rollups of each zone, kept for the usage view (server/zone/usage.js).
+ app.post('/internal/events', eventsIntake({ secrets: config.events.secrets, usage, log }));
+
// ── Machine endpoints ───────────────────────────────────
// ── Machine endpoints ───────────────────────────────────
app.get('/api/health', (_req, res) => res.json({ status: 'ok', service: 'openvibe-zone', version: VERSION }));
app.get('/api/health', (_req, res) => res.json({ status: 'ok', service: 'openvibe-zone', version: VERSION }));
release.mount(app, { registry: metrics.registry });
release.mount(app, { registry: metrics.registry });
@@ -60,6 +60,13 @@ function load(env = process.env) {
@@ -60,6 +60,13 @@ function load(env = process.env) {
internalUrl: trim(env.MEDIA_INTERNAL_URL || 'http://127.0.0.1:4100'),
internalUrl: trim(env.MEDIA_INTERNAL_URL || 'http://127.0.0.1:4100'),
s3Endpoint: trim(env.ZONE_S3_ENDPOINT || 'https://s3.openvibe.media'),
s3Endpoint: trim(env.ZONE_S3_ENDPOINT || 'https://s3.openvibe.media'),
},
},
+ // OpenVibe.Events: ZONE_EVENTS_SECRET signs the deliveries to POST /internal/events (Media's per-zone rollups,
+ // media.usage.recorded; comma-separated for rotation, 32+ characters each). Unset, the route answers 503 and no
+ // subscription is created at boot.
+ events: {
+ url: trim(env.EVENTS_URL || ''),
+ secrets: String(env.ZONE_EVENTS_SECRET || '').split(',').map((x) => x.trim()).filter(Boolean),
+ },
// The zones one project environment may hold besides its default.
// The zones one project environment may hold besides its default.
maxZonesPerEnv: Math.max(1, int(env.ZONE_MAX_PER_ENV, 20)),
maxZonesPerEnv: Math.max(1, int(env.ZONE_MAX_PER_ENV, 20)),
};
};
new file mode 100644
new file mode 100644
@@ -0,0 +1,43 @@
@@ -0,0 +1,43 @@
+'use strict';
+/**
+ * POST /internal/events: OpenVibe.Events delivers Zone's subscriptions here (media.usage.recorded). No body parser runs
+ * before it: the v2 signature covers the raw body. Loopback only: a request that came through a proxy is refused.
+ * 2xx once applied (or ignored with a reason); anything else makes Events retry.
+ */
+const { parseDelivery } = require('openvibe-sdk/events');
+
+const LIMIT = 256 * 1024;
+
+function eventsIntake({ secrets = [], usage, log = console, now = () => Date.now() }) {
+ const keys = (secrets || []).filter((x) => typeof x === 'string' && x.length >= 32);
+ const reply = (res, status, body) => {
+ res.status(status).set('Cache-Control', 'no-store').type(status < 300 ? 'application/json' : 'application/problem+json').send(JSON.stringify(body));
+ };
+ const problem = (res, status, code, detail) => reply(res, status, { type: 'about:blank', title: code, status, code, detail });
+ const readRaw = (req) => new Promise((resolve, reject) => {
+ const chunks = [];
+ let size = 0;
+ req.on('data', (c) => { size += c.length; if (size > LIMIT) { reject(Object.assign(new Error('too large'), { status: 413 })); req.destroy(); } else chunks.push(c); });
+ req.on('end', () => resolve(Buffer.concat(chunks)));
+ req.on('error', reject);
+ });
+ return async function handle(req, res) {
+ if (req.headers['x-forwarded-for'] || req.headers['x-real-ip'] || req.headers['cf-connecting-ip']) return problem(res, 403, 'zone.internal_only', 'internal route');
+ if (!keys.length) return problem(res, 503, 'zone.events_disabled', 'no events secret is configured');
+ let raw;
+ try { raw = await readRaw(req); } catch (err) { return problem(res, err.status || 400, 'bad_request', 'unreadable body'); }
+ let delivery = null;
+ for (const k of keys) { delivery = parseDelivery(raw, req.headers, k, { requireV2: true, now: now() }); if (delivery) break; }
+ if (!delivery || !delivery.event || typeof delivery.event.event_type !== 'string') return problem(res, 401, 'zone.bad_signature', 'X-OpenVibe-Signature-V2 does not verify or is outside the replay window');
+ const ev = delivery.event;
+ try {
+ const outcome = ev.event_type === usage.TOPIC ? await usage.apply(ev) : 'ignored:type';
+ return reply(res, 200, { event_id: ev.event_id, outcome });
+ } catch (err) {
+ log.warn(`[Events] ${ev.event_type} ${ev.event_id} failed: ${(err && err.message) || err}`);
+ return problem(res, 503, 'zone.retry', 'not applied; Events will retry');
+ }
+ };
+}
+
+module.exports = { eventsIntake };
@@ -48,7 +48,7 @@ const plural = (n, one, many) => `${Number(n).toLocaleString('en-US')} ${n === 1
@@ -48,7 +48,7 @@ const plural = (n, one, many) => `${Number(n).toLocaleString('en-US')} ${n === 1
const day = (iso) => (iso ? String(iso).slice(0, 10) : '—');
const day = (iso) => (iso ? String(iso).slice(0, 10) : '—');
function createPageRoutes(ctx) {
function createPageRoutes(ctx) {
- const { config, clients, zones } = ctx;
+ const { config, clients, zones, usage: metering } = ctx;
const r = asyncRouter();
const r = asyncRouter();
const form = express.urlencoded({ extended: false, limit: '8kb' });
const form = express.urlencoded({ extended: false, limit: '8kb' });
const PUBLIC_CACHE = cache.htmlHeaders({ maxAge: 300 });
const PUBLIC_CACHE = cache.htmlHeaders({ maxAge: 300 });
@@ -144,10 +144,21 @@ ${projects.length ? html`<ul class="zn-cards">${projects.map((p) => html`<li cla
@@ -144,10 +144,21 @@ ${projects.length ? html`<ul class="zn-cards">${projects.map((p) => html`<li cla
for (const env of envs) await zones.ensureDefault(project, env);
for (const env of envs) await zones.ensureDefault(project, env);
const list = (await zones.list(project, {})).zones;
const list = (await zones.list(project, {})).zones;
const usage = new Map();
const usage = new Map();
+ const metered = new Map();
+ const monthStart = (() => { const d = new Date(); return Date.UTC(d.getUTCFullYear(), d.getUTCMonth(), 1); })();
await Promise.all(list.filter((z) => z.state !== 'deleted').map(async (z) => {
await Promise.all(list.filter((z) => z.state !== 'deleted').map(async (z) => {
const rep = await clients.report(z.id).catch(() => null);
const rep = await clients.report(z.id).catch(() => null);
if (rep) usage.set(z.id, rep);
if (rep) usage.set(z.id, rep);
+ if (metering) metered.set(z.id, (await metering.series(z.id, { from: monthStart }).catch(() => ({ series: [] }))).series);
}));
}));
+ // This month's requests and egress, as Media reported them by the hour (media.usage.recorded).
+ const monthLine = (id) => {
+ const list = metered.get(id) || [];
+ const sum = (pred) => list.filter(pred).reduce((n, x) => n + x.quantity, 0);
+ const requests = sum((x) => x.unit === 'requests');
+ const egress = sum((x) => x.unit === 'egress_bytes');
+ return requests || egress ? `this month: ${plural(requests, 'request', 'requests')} · ${bytes(egress)} out` : '';
+ };
let keys = [];
let keys = [];
let keysError = null;
let keysError = null;
if (admin) { try { keys = await clients.listKeys(project, me.subject); } catch (err) { keysError = err.message; } }
if (admin) { try { keys = await clients.listKeys(project, me.subject); } catch (err) { keysError = err.message; } }
@@ -155,7 +166,7 @@ ${projects.length ? html`<ul class="zn-cards">${projects.map((p) => html`<li cla
@@ -155,7 +166,7 @@ ${projects.length ? html`<ul class="zn-cards">${projects.map((p) => html`<li cla
const rep = usage.get(z.id);
const rep = usage.get(z.id);
const u = rep && rep.usage;
const u = rep && rep.usage;
return html`<tr id="${z.id}"><td><code>${z.name}</code>${z.default ? html` <span class="zn-badge">default</span>` : ''}</td><td>${z.state}</td>
return html`<tr id="${z.id}"><td><code>${z.name}</code>${z.default ? html` <span class="zn-badge">default</span>` : ''}</td><td>${z.state}</td>
-<td class="zn-n">${u ? `${bytes(u.stored_bytes)} · ${plural(u.objects, 'object', 'objects')}` : '—'}</td>
+<td class="zn-n">${u ? `${bytes(u.stored_bytes)} · ${plural(u.objects, 'object', 'objects')}` : '—'}${monthLine(z.id) ? html`<br><span class="zn-muted">${monthLine(z.id)}</span>` : ''}</td>
<td class="zn-n">${z.quota_bytes ? bytes(z.quota_bytes) : (z.default ? 'environment' : '—')}</td><td>${day(z.created_at)}</td>
<td class="zn-n">${z.quota_bytes ? bytes(z.quota_bytes) : (z.default ? 'environment' : '—')}</td><td>${day(z.created_at)}</td>
<td>${admin && !z.default && z.state === 'active' ? html`<details><summary>Delete</summary><form method="post" action="/p/${project}/zones/${z.id}/delete">
<td>${admin && !z.default && z.state === 'active' ? html`<details><summary>Delete</summary><form method="post" action="/p/${project}/zones/${z.id}/delete">
<p>This deletes every object in <code>${z.name}</code>, with no undo, and the name can never be used again.</p>
<p>This deletes every object in <code>${z.name}</code>, with no undo, and the name can never be used again.</p>
@@ -289,7 +300,7 @@ ${keysSection}
@@ -289,7 +300,7 @@ ${keysSection}
<li><code>GET /zones/:zone_id</code> (<code>zone.zone.read</code>): the zone with its stored bytes and objects.</li>
<li><code>GET /zones/:zone_id</code> (<code>zone.zone.read</code>): the zone with its stored bytes and objects.</li>
<li><code>PATCH /zones/:zone_id</code> (<code>zone.zone.manage</code>): <code>{ description?, quota_bytes? }</code>.</li>
<li><code>PATCH /zones/:zone_id</code> (<code>zone.zone.manage</code>): <code>{ description?, quota_bytes? }</code>.</li>
<li><code>DELETE /zones/:zone_id?recursive=true</code> (<code>zone.zone.delete</code>). Without <code>recursive</code>, a zone that still holds objects answers <code>zone.not_empty</code>. A zone holding an object under a retention hold answers <code>zone.held</code>.</li>
<li><code>DELETE /zones/:zone_id?recursive=true</code> (<code>zone.zone.delete</code>). Without <code>recursive</code>, a zone that still holds objects answers <code>zone.not_empty</code>. A zone holding an object under a retention hold answers <code>zone.held</code>.</li>
-<li><code>GET /zones/:zone_id/usage</code> (<code>zone.usage.read</code>).</li>
+<li><code>GET /zones/:zone_id/usage</code> (<code>zone.usage.read</code>): Media's live snapshot, and this month's metered series (requests, egress, byte-hours) as Media reports each closed hour.</li>
</ul>
</ul>
<p>The objects of a zone are OpenVibe.Media objects. Media's own API (<code>/api/v2/:project_id/objects</code>, with <code>namespace</code> set to the zone's id) reads and writes the same objects.</p>`,
<p>The objects of a zone are OpenVibe.Media objects. Media's own API (<code>/api/v2/:project_id/objects</code>, with <code>namespace</code> set to the zone's id) reads and writes the same objects.</p>`,
}));
}));
@@ -6,6 +6,7 @@
@@ -6,6 +6,7 @@
*/
*/
const { createApp } = require('./app');
const { createApp } = require('./app');
const { gracefulStop } = require('openvibe-sdk/service');
const { gracefulStop } = require('openvibe-sdk/service');
+const { startSubscriptions } = require('openvibe-sdk/account-data');
/**
/**
* The process stop (openvibe-sdk/service): the HTTP drain runs, then the timers clear, the zone job stops, the JWKS
* The process stop (openvibe-sdk/service): the HTTP drain runs, then the timers clear, the zone job stops, the JWKS
@@ -30,7 +31,14 @@ async function start() {
@@ -30,7 +31,14 @@ async function start() {
ctx.keys.client.start();
ctx.keys.client.start();
// The zone job (zone/zones.js tick): pushes what Media has not acknowledged and finishes deletions, every 15 s.
// The zone job (zone/zones.js tick): pushes what Media has not acknowledged and finishes deletions, every 15 s.
ctx.zones.start();
ctx.zones.start();
- createLifecycle({ server, ctx, extra: [() => ctx.zones.stop()] });
+ // Media's per-zone rollups (media.usage.recorded) at OpenVibe.Events, created when missing; off without EVENTS_URL,
+ // ZONE_EVENTS_SECRET or the client secret.
+ const subscriptions = startSubscriptions({
+ eventsUrl: config.events.url, endpoint: `http://127.0.0.1:${config.port}/internal/events`, secret: config.events.secrets[0],
+ topics: [ctx.usage.TOPIC],
+ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
+ });
+ createLifecycle({ server, ctx, extra: [() => ctx.zones.stop(), () => { if (subscriptions) subscriptions.stop(); }] });
return { server, ctx };
return { server, ctx };
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,67 @@
@@ -0,0 +1,67 @@
+'use strict';
+/**
+ * A zone's metered usage (ADR-031 section 8): the hourly rollups Media sends as media.usage.recorded
+ * (common.usage-recorded@1), kept per zone, hour and series. Zone never counts: it stores what Media reports and sums
+ * it for GET /zones/:zone/usage and the console. Media is the only source; a rollup names its zone twice (resource is
+ * the zone's OVRN, dimension its id in lowercase) and both must name a zone of its project here.
+ *
+ * apply(event) 'recorded' or an 'ignored:*' outcome (a newer or equal revision replaces the hour)
+ * series(zoneId, { from }) [{ capability, unit, quantity }] summed from `from` (ms), and the last reported hour's end
+ * prune() hours older than KEEP_DAYS
+ */
+const contracts = require('openvibe-contracts');
+
+const TOPIC = 'media.usage.recorded';
+const HOUR_MS = 3600 * 1000;
+const KEEP_DAYS = 400;
+const OVRN = /^ovrn:zone:(prj_[0-9A-HJKMNP-TV-Z]{26}):object-zone\/(zon_[0-9A-HJKMNP-TV-Z]{26})$/;
+
+function createUsage({ s, log = console }) {
+ const { db } = s;
+
+ async function apply(event) {
+ if (!event || event.event_type !== TOPIC) return 'ignored:type';
+ if (event.source !== 'media') return 'ignored:source';
+ const p = event.payload && typeof event.payload === 'object' ? event.payload : {};
+ const v = contracts.validate(`${TOPIC}@${event.version || 1}`, p);
+ if (!v.valid) {
+ log.warn(`[Usage] ${event.event_id} does not match ${TOPIC}: ${v.errors.slice(0, 3).map((e) => `${e.path} ${e.message}`).join('; ')}`);
+ return 'ignored:payload';
+ }
+ const subject = event.subject || {};
+ if (subject.type !== 'project' || subject.id !== p.project_id) return 'ignored:subject';
+ const m = OVRN.exec(String(p.resource || ''));
+ if (!m || m[1] !== p.project_id || String(p.dimension || '') !== m[2].toLowerCase()) return 'ignored:resource';
+ const zone = await db.maybe('SELECT id, project_id, env FROM zone_zones WHERE id = $1', [m[2]]);
+ if (!zone || zone.project_id !== p.project_id || zone.env !== p.env) return 'ignored:zone';
+ const start = Date.parse(p.window_start);
+ if (p.window !== 'hour' || Date.parse(p.window_end) - start !== HOUR_MS || start % HOUR_MS !== 0) return 'ignored:window';
+ if (start > s.now() + HOUR_MS) return 'ignored:future';
+ const revision = p.revision || 1;
+ const row = await db.maybe(`INSERT INTO zone_usage (zone_id, window_start, capability, unit, quantity, errors, revision, event_id, recorded_at)
+ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+ ON CONFLICT (zone_id, window_start, capability, unit) DO UPDATE SET quantity = excluded.quantity, errors = excluded.errors,
+ revision = excluded.revision, event_id = excluded.event_id, recorded_at = excluded.recorded_at
+ WHERE zone_usage.revision <= excluded.revision
+ RETURNING zone_id`, [zone.id, start, p.capability, p.unit, p.quantity, p.errors, revision, event.event_id, s.now()]);
+ return row ? 'recorded' : 'ignored:stale';
+ }
+
+ async function series(zoneId, { from }) {
+ const rows = await db.many(`SELECT capability, unit, SUM(quantity) AS quantity FROM zone_usage
+ WHERE zone_id = $1 AND window_start >= $2 GROUP BY capability, unit ORDER BY capability, unit`, [zoneId, from]);
+ const last = await db.value('SELECT MAX(window_start) FROM zone_usage WHERE zone_id = $1 AND window_start >= $2', [zoneId, from]);
+ return {
+ series: rows.slice(0, 32).map((r) => ({ capability: r.capability, unit: r.unit, quantity: Number(r.quantity) })),
+ reportedUntil: last == null ? null : Number(last) + HOUR_MS,
+ };
+ }
+
+ async function prune() {
+ await db.exec('DELETE FROM zone_usage WHERE window_start < $1', [s.now() - KEEP_DAYS * 24 * HOUR_MS]);
+ }
+
+ return { apply, series, prune, TOPIC };
+}
+
+module.exports = { createUsage, TOPIC };
@@ -28,7 +28,7 @@ class ZoneError extends Error {
@@ -28,7 +28,7 @@ class ZoneError extends Error {
const fail = (status, code, detail) => { throw new ZoneError(status, code, detail); };
const fail = (status, code, detail) => { throw new ZoneError(status, code, detail); };
const iso = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());
const iso = (ms) => (ms == null ? null : new Date(Number(ms)).toISOString());
-function createZones({ s, clients, config, log = console }) {
+function createZones({ s, clients, config, log = console, metering = null }) {
const { db } = s;
const { db } = s;
/** The zone.object-zone@1 of a row (usage when given; deletion progress while it is deleting or deleted). */
/** The zone.object-zone@1 of a row (usage when given; deletion progress while it is deleting or deleted). */
@@ -198,15 +198,19 @@ function createZones({ s, clients, config, log = console }) {
@@ -198,15 +198,19 @@ function createZones({ s, clients, config, log = console }) {
const now = new Date(s.now());
const now = new Date(s.now());
const start = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), 1));
const start = new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), 1));
const current = rep && rep.usage ? rep.usage : { stored_bytes: 0, objects: 0, as_of: now.toISOString() };
const current = rep && rep.usage ? rep.usage : { stored_bytes: 0, objects: 0, as_of: now.toISOString() };
+ // The series sums Media's hourly rollups (media.usage.recorded) of this month; window_end is the end of the
+ // last hour Media reported (the month's start when none has arrived yet).
+ const metered = metering ? await metering.series(row.id, { from: start.getTime() }) : { series: [], reportedUntil: null };
const out = {
const out = {
zone_id: row.id, ovrn: contracts.zones.identity({ ...row, default: row.is_default }).ovrn,
zone_id: row.id, ovrn: contracts.zones.identity({ ...row, default: row.is_default }).ovrn,
current: { stored_bytes: current.stored_bytes, objects: current.objects, as_of: current.as_of },
current: { stored_bytes: current.stored_bytes, objects: current.objects, as_of: current.as_of },
- window_start: start.toISOString(), window_end: now.toISOString(), series: [],
+ window_start: start.toISOString(), window_end: new Date(metered.reportedUntil || start.getTime()).toISOString(), series: metered.series,
};
};
return out;
return out;
}
}
let ticking = false;
let ticking = false;
+ let lastPrune = 0;
async function tick() {
async function tick() {
if (ticking) return { pushed: 0, deleted: 0 };
if (ticking) return { pushed: 0, deleted: 0 };
ticking = true;
ticking = true;
@@ -224,6 +228,7 @@ function createZones({ s, clients, config, log = console }) {
@@ -224,6 +228,7 @@ function createZones({ s, clients, config, log = console }) {
}
}
}
}
}
}
+ if (metering && s.now() - lastPrune > 3600 * 1000) { lastPrune = s.now(); await metering.prune(); }
} finally { ticking = false; }
} finally { ticking = false; }
return { pushed, deleted };
return { pushed, deleted };
}
}
@@ -62,7 +62,8 @@ function fakeMedia(network) {
@@ -62,7 +62,8 @@ function fakeMedia(network) {
const { startNetwork } = require('./helpers/mocks');
const { startNetwork } = require('./helpers/mocks');
const net = await startNetwork();
const net = await startNetwork();
let media;
let media;
- const t = await boot({ clients: new Proxy({}, { get: (_o, k) => (...a) => media[k](...a) }) });
+ const EVENTS_SECRET = 'e'.repeat(40); // a test value: 32+ characters
+ const t = await boot({ clients: new Proxy({}, { get: (_o, k) => (...a) => media[k](...a) }), env: { ZONE_EVENTS_SECRET: EVENTS_SECRET } });
await net.close();
await net.close();
media = fakeMedia(t.network);
media = fakeMedia(t.network);
const owner = t.network.addUser('owen');
const owner = t.network.addUser('owen');
@@ -157,6 +158,46 @@ function fakeMedia(network) {
@@ -157,6 +158,46 @@ function fakeMedia(network) {
assert.deepStrictEqual(z.usage.objects, 3);
assert.deepStrictEqual(z.usage.objects, 3);
});
});
+ check('Media\'s hourly rollups (media.usage.recorded) arrive signed and sum into the usage series and the console', async () => {
+ const { signDeliveryHeaders } = require('openvibe-sdk/events');
+ const { ids } = contracts;
+ const HOUR = 3600 * 1000;
+ const hour = Math.floor(Date.now() / HOUR) * HOUR - HOUR;
+ const ovrn = `ovrn:zone:${project}:object-zone/${assets.id}`;
+ const def = (await t.get(`/api/v1/projects/${project}/zones?env=${assets.env}`, { as: owner })).json().zones.find((z) => z.default);
+ const rollup = (payload, over = {}) => ({
+ event_id: ids.newId('event'), event_type: 'media.usage.recorded', version: 1, source: 'media', actor: { type: 'service', id: 'media' },
+ timestamp: new Date().toISOString(), subject: { type: 'project', id: project }, visibility: 'internal', priority: 'low',
+ payload: { project_id: project, env: assets.env, resource: ovrn, dimension: assets.id.toLowerCase(), window: 'hour',
+ window_start: new Date(hour).toISOString(), window_end: new Date(hour + HOUR).toISOString(), errors: 0, ...payload }, ...over,
+ });
+ const deliver = async (ev, { secret = EVENTS_SECRET, headers = {} } = {}) => {
+ const raw = JSON.stringify({ event: ev });
+ const res = await fetch(`${t.base}/internal/events`, { method: 'POST', headers: { 'content-type': 'application/json', ...signDeliveryHeaders(raw, secret), ...headers }, body: raw });
+ return { status: res.status, body: await res.json() };
+ };
+ let r = await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 10 }));
+ assert.deepStrictEqual([r.status, r.body.outcome], [200, 'recorded'], JSON.stringify(r.body));
+ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'egress_bytes', quantity: 2048 }))).body.outcome, 'recorded');
+ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 12, revision: 2 }))).body.outcome, 'recorded', 'a corrected hour replaces it');
+ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 1, revision: 1 }))).body.outcome, 'ignored:stale', 'an older revision never does');
+ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 1 }, { source: 'tools' }))).body.outcome, 'ignored:source', 'only Media reports');
+ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 1, dimension: def.id.toLowerCase() }))).body.outcome, 'ignored:resource', 'resource and dimension name one zone');
+ const elsewhere = `zon_${ids.ulid()}`;
+ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 1, resource: `ovrn:zone:${project}:object-zone/${elsewhere}`, dimension: elsewhere.toLowerCase() }))).body.outcome, 'ignored:zone', 'a zone Zone does not have');
+ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 1, env: assets.env === 'production' ? 'sandbox' : 'production' }))).body.outcome, 'ignored:zone', 'or of another environment');
+ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 1 }), { secret: 'x'.repeat(40) })).status, 401, 'unsigned deliveries are refused');
+ assert.strictEqual((await deliver(rollup({ capability: 'media.object.read', unit: 'requests', quantity: 1 }), { headers: { 'x-forwarded-for': '203.0.113.9' } })).status, 403, 'and so is anything through a proxy');
+ r = await t.get(`/api/v1/zones/${assets.id}/usage`, { as: viewer });
+ const u = r.json();
+ assert.ok(contracts.zones.checkObjectZoneUsage(u).valid, JSON.stringify(contracts.zones.checkObjectZoneUsage(u).errors));
+ const q = (unit) => (u.series.find((x) => x.capability === 'media.object.read' && x.unit === unit) || {}).quantity;
+ assert.deepStrictEqual([q('requests'), q('egress_bytes')], [12, 2048], 'the series sums the month\'s hours');
+ assert.strictEqual(u.window_end, new Date(hour + HOUR).toISOString(), 'up to the last hour Media reported');
+ r = await t.get(`/p/${project}`, { as: owner });
+ assert.ok(r.text.includes('this month: 12 requests · 2.0 KB out'), 'the console shows the month\'s requests and egress');
+ });
+
check('deletion: not empty, held, then recursive; the job marks it deleted and its name is retired', async () => {
check('deletion: not empty, held, then recursive; the job marks it deleted and its name is retired', async () => {
let r = await t.get(`/api/v1/zones/${assets.id}`, { as: owner, method: 'DELETE', headers: same });
let r = await t.get(`/api/v1/zones/${assets.id}`, { as: owner, method: 'DELETE', headers: same });
assert.deepStrictEqual([r.status, r.json().code], [409, 'zone.not_empty']);
assert.deepStrictEqual([r.status, r.json().code], [409, 'zone.not_empty']);