import { writeFileSync } from 'node:fs'; import { generate } from '@openvidu-dashboard/data-generator/generator'; import { type Db, type FindCursor, MongoClient } from 'mongodb'; import { afterAll, beforeAll, describe, expect, inject, it } from 'vitest'; import { Collection, FeatureKey, ParticipantKind } from '@openvidu-dashboard/typings'; import { AnalyticsCache } from '../../src/analytics/analytics-cache.js'; import { CE_ANALYTICS_CATALOGUE, metricFields } from '../../src/analytics/analytics-catalogue.js'; import { AnalyticsFeatureGate } from '../../src/analytics/analytics-gate.js'; import { planSeries, type RollupCollection } from '../../src/analytics/analytics-planner.js'; import { AnalyticsRepository, ROLLUP_INDEX } from '../../src/analytics/analytics-repository.js'; import { analyticsBreakdownsQuery, analyticsExportQuery, analyticsSeriesQuery, analyticsSummaryQuery } from '../../src/analytics/analytics-schemas.js'; import { AnalyticsService } from '../../src/analytics/analytics-service.js'; import { expectIndexBacked, expectIndexes, p95 } from '../support/query-plan.js'; import { expectWithinBudgets } from '../support/latency-budget.js'; /** * Every analytics query explained on a database generated by the data generator with five-minute, node and room-sum * rollups, the consistency of the rollup sums with the entities, and the p95 of the performance budget, which fails the * suite when exceeded. ANALYTICS_PLAN_SCENARIO (S by default), ANALYTICS_PLAN_DAYS (2 by default) and * ANALYTICS_PLAN_REPORT (a file that also receives the report) work as in the rooms suite. */ const mongoUri = inject('mongoUri'); const scenario = (process.env.ANALYTICS_PLAN_SCENARIO ?? 'S') as 'S' | 'M' | 'L'; const days = Number(process.env.ANALYTICS_PLAN_DAYS ?? '2'); const DATABASE = 'dashboard_test_analytics_plan'; const HOUR = 3_600_000; const DAY = 24 * HOUR; const GENERATION_TIMEOUT_MS = scenario === 'S' ? 600_000 : 3_600_000; const SERIES_METRICS = 'roomsCreated,participantsJoined,connectionSuccessRate,bytesUp,bytesDown'; const BREAKDOWNS = 'sdk,os,browser,connectionType,kind,egressType,egressStatus'; describe.skipIf(mongoUri === '')(`analytics query plan at scenario ${scenario}`, () => { let client: MongoClient; let db: Db; let repository: AnalyticsRepository; let service: AnalyticsService; let node: string; const now = new Date(); const hour = new Date(Math.floor(now.getTime() / HOUR) * HOUR); const report: string[] = []; const clock = () => now; /** The rollup read of a series request, explained with the bound of the documents in its range. */ const explainSeries = async (query: Record, expected: RollupCollection) => { const params = analyticsSeriesQuery(clock).parse(query); const plan = planSeries({ ...params, now, retention: { statsMs: 7 * DAY, entitiesMs: 90 * DAY, rollups5MinMs: 7 * DAY, rollupsHourlyMs: 400 * DAY }, info: CE_ANALYTICS_CATALOGUE.info }); const from = plan.starts[0] ?? params.from; const filter = { scope: params.scope, bucket: { $gte: from, $lt: params.to } }; const documents = await db.collection(plan.collection).countDocuments(filter); const cursor = repository.rollups( plan.collection, params.scope, from, params.to, metricFields(CE_ANALYTICS_CATALOGUE, params.metrics) ) as unknown as FindCursor; const summary = await expectIndexBacked(db, cursor, { maxDocsExamined: documents }); expect(plan.collection).toBe(expected); expectIndexes(summary, [ROLLUP_INDEX]); expect(summary.totalKeysExamined).toBeLessThanOrEqual(documents + 1); return documents; }; beforeAll(async () => { client = new MongoClient(mongoUri); db = client.db(DATABASE); await db.dropDatabase(); await generate(db, { scenario, days, statsMode: 'participant', seed: 1, retentionDays: { events: 7, stats: 7, tracks: 7, entities: 90 }, timeseries: true, rollups: { fiveMinute: true, nodes: true, roomSums: true } }); repository = new AnalyticsRepository(db); service = new AnalyticsService(db, { now: clock, cache: new AnalyticsCache(0, clock), gate: new AnalyticsFeatureGate(() => [FeatureKey.PER_NODE_BREAKDOWN]), catalogue: CE_ANALYTICS_CATALOGUE }); const scopes = ((await db.collection(Collection.ANALYTICS_HOURLY).distinct('scope')) as string[]).filter( (scope) => scope.startsWith('node:') ); if (scopes[0] === undefined) { throw new Error('the generated database has no node scope'); } node = scopes[0]; }, GENERATION_TIMEOUT_MS); afterAll(async () => { if (report.length > 0) { const text = [`Analytics p95 of 20 runs, scenario ${scenario}, ${String(days)} days`, ...report].join('\n'); console.log(text); if (process.env.ANALYTICS_PLAN_REPORT !== undefined) { writeFileSync(process.env.ANALYTICS_PLAN_REPORT, `${text}\n`); } } await db.dropDatabase(); await client.close(); }); it('series: bounded scans of { scope, bucket } in the collection of each granularity', async () => { expect(await explainSeries({ metrics: SERIES_METRICS, from: 'now-24h' }, 'analytics_5min')).toBeGreaterThan(0); expect( await explainSeries({ metrics: SERIES_METRICS, from: 'now-7d', scope: node }, 'analytics_hourly') ).toBeGreaterThan(0); await explainSeries( { metrics: SERIES_METRICS, from: 'now-400d', granularity: '1d', tz: 'Europe/Madrid' }, 'analytics_hourly' ); expect( await explainSeries({ metrics: SERIES_METRICS, from: 'now-400d', granularity: '1d' }, 'analytics_daily') ).toBeGreaterThan(0); }); it('sums of the rollups equal the entity counts over whole hours', async () => { const from = new Date(hour.getTime() - Math.min(days, 2) * DAY); const summary = await service.summary({ from, to: hour, tz: 'UTC', scope: 'global' }); const count = (collection: Collection, field: string, filter: Record = {}) => db.collection(collection).countDocuments({ ...filter, [field]: { $gte: from, $lt: hour } }); expect(summary.roomsCreated).toBe(await count(Collection.ROOMS, 'createdAt')); // Rollups count standard participant sessions only. expect(summary.participantSessions).toBe( await count(Collection.PARTICIPANTS, 'joinedAt', { kind: ParticipantKind.STANDARD }) ); expect(summary.egressStarted).toBe(await count(Collection.EGRESSES, 'startedAt')); expect(summary.roomsCreated).toBeGreaterThan(0); }); it('reports the p95 of every budgeted query and fails above its budget', async () => { const series = (query: Record) => () => service.series(analyticsSeriesQuery(clock).parse(query)); const rows: [string, number, () => Promise][] = [ ['Series, 5m over 24 h', 50, series({ metrics: SERIES_METRICS, from: 'now-24h' })], ['Series, 1h over 7 d', 50, series({ metrics: SERIES_METRICS, from: 'now-7d', granularity: '1h' })], [ 'Series, 1d over 400 d, Europe/Madrid', 400, series({ metrics: SERIES_METRICS, from: 'now-400d', granularity: '1d', tz: 'Europe/Madrid' }) ], [ 'Breakdowns, 90 d', 150, () => service.breakdowns( analyticsBreakdownsQuery(clock).parse({ dimensions: BREAKDOWNS, from: 'now-90d' }) ) ], ['Summary, 90 d', 150, () => service.summary(analyticsSummaryQuery(clock).parse({ from: 'now-90d' }))], [ 'CSV, series 5m over 24 h', 70, () => service.export( analyticsExportQuery(clock).parse({ kind: 'series', metrics: SERIES_METRICS, from: 'now-24h' }) ) ] ]; report.push('| Query | Budget | p95 |', '| --- | --- | --- |'); const over: string[] = []; for (const [name, budget, run] of rows) { const measured = await p95(run); report.push(`| ${name} | ${String(budget)} ms | ${measured.toFixed(1)} ms |`); if (measured > budget) { over.push(`${name}: ${measured.toFixed(1)} ms > ${String(budget)} ms`); } } expectWithinBudgets(over); }); });