diff --git a/openvidu-test-integration/jest.config.ts b/openvidu-test-integration/jest.config.ts index d0a22e5a9..73847094d 100644 --- a/openvidu-test-integration/jest.config.ts +++ b/openvidu-test-integration/jest.config.ts @@ -1,7 +1,9 @@ import type { Config } from "jest"; const config: Config = { - preset: "ts-jest" + preset: "ts-jest", + // Every suite starts and removes the same local deployment: they run one at a time. + maxWorkers: 1 }; export default config; diff --git a/openvidu-test-integration/tests/analytics-migration.test.ts b/openvidu-test-integration/tests/analytics-migration.test.ts new file mode 100644 index 000000000..e9521b380 --- /dev/null +++ b/openvidu-test-integration/tests/analytics-migration.test.ts @@ -0,0 +1,77 @@ +import { Int32, Long } from "mongodb"; +import { waitFor } from "./utils/helper"; +import { legacyRoom, LEGACY_PARTICIPANTS, LEGACY_ROOM_NAME, LEGACY_ROOM_SID } from "./utils/legacy"; +import { LocalDeployment } from "./utils/local-deployment"; +import { MongoService } from "./utils/mongodb"; +import { Collection } from "./utils/types"; + +/** + * The first start of the OpenVidu server on a database of a previous version renames the legacy events to + * `legacy_events` and migrates them, and the legacy statistics, to the analytics data model in the background. + */ +describe("OpenVidu analytics legacy migration", () => { + const mongo = MongoService.getInstance(); + + beforeEach(async () => { + await LocalDeployment.start(); + }, 120000); + + afterEach(() => { + LocalDeployment.stop(); + }, 120000); + + afterAll(async () => { + await mongo.disconnect(); + }); + + it("migrates the analytics data of a previous version at the first start", async () => { + const start = Math.floor(Date.now() / 1000) - 3 * 3_600; + + LocalDeployment.stopContainer("openvidu"); + await mongo.database().dropDatabase(); + await mongo.database().collection("events").insertMany(legacyRoom(start)); + await mongo + .database() + .collection("last_alive") + .insertOne({ _id: "server" as never, last_alive: { seconds: Long.fromNumber(start), nanos: new Int32(0) } }); + const since = new Date(); + LocalDeployment.startContainer("openvidu"); + await LocalDeployment.waitForAnalytics(since); + + await waitFor( + async () => (await mongo.schemaMeta("migration"))?.state === "completed", + 180, + "the legacy migration to complete" + ); + + const collections = await mongo.collectionNames(); + expect(collections).toContain(Collection.LEGACY_EVENTS); + expect(collections).not.toContain("active_entities"); + expect(collections).not.toContain("last_alive"); + + const room = await mongo.findEntity(Collection.ROOMS, { _id: LEGACY_ROOM_SID }); + expect(room).toMatchObject({ name: LEGACY_ROOM_NAME, status: "closed", closedBy: "event", endReason: "IDLE_TIMEOUT" }); + expect((room!.createdAt as Date).getTime()).toBe(start * 1000); + expect((room!.endedAt as Date).getTime()).toBe((start + 1_820) * 1000); + expect(room!.participants).toMatchObject({ sessions: 2, unique: 2, active: 0 }); + + for (const p of LEGACY_PARTICIPANTS) { + const participant = await mongo.findEntity(Collection.PARTICIPANTS, { _id: p.sid }); + expect(participant).toMatchObject({ + identity: p.identity, + roomSid: LEGACY_ROOM_SID, + status: "left", + closedBy: "event", + disconnectReason: "CLIENT_INITIATED" + }); + expect(participant!.durationMs).toBe((p.left - p.joined) * 1000); + } + + const events = await mongo.findEvents({ roomSid: LEGACY_ROOM_SID }); + expect(events.map((event) => event.type)).toEqual( + expect.arrayContaining(["ROOM_CREATED", "PARTICIPANT_JOINED", "PARTICIPANT_LEFT", "ROOM_ENDED"]) + ); + expect(JSON.stringify(events)).not.toContain("legacy-turn-password"); + expect(await mongo.schemaMeta("migration")).toMatchObject({ processed: 8, total: 8 }); + }, 600000); +}); diff --git a/openvidu-test-integration/tests/analytics-mongo-outage.test.ts b/openvidu-test-integration/tests/analytics-mongo-outage.test.ts new file mode 100644 index 000000000..1f2ea3d0d --- /dev/null +++ b/openvidu-test-integration/tests/analytics-mongo-outage.test.ts @@ -0,0 +1,61 @@ +import { killProcess, sleep, waitFor } from "./utils/helper"; +import { Livekit } from "./utils/livekit"; +import { LocalDeployment } from "./utils/local-deployment"; +import { MongoService } from "./utils/mongodb"; +import { Collection } from "./utils/types"; + +/** + * While MongoDB is down the OpenVidu server keeps the analytics events in memory and retries the failed batches, so + * what happened during the outage is written once MongoDB is back. + */ +describe("OpenVidu analytics during a MongoDB outage", () => { + const mongo = MongoService.getInstance(); + let pids: number[] = []; + + beforeEach(async () => { + await LocalDeployment.start(); + }, 120000); + + afterEach(() => { + for (const pid of pids) { + killProcess(pid, "SIGKILL"); + } + pids = []; + LocalDeployment.stop(); + }, 120000); + + afterAll(async () => { + await mongo.disconnect(); + }); + + it("writes a session that joined and left while MongoDB was down", async () => { + const roomName = "OutageRoom"; + const identity = "outage-alice"; + + LocalDeployment.stopContainer("mongo"); + pids = [Livekit.joinParticipantToRoom(identity, roomName)]; + await sleep(20); + killProcess(pids[0]); + pids = []; + await sleep(20); + LocalDeployment.startContainer("mongo"); + + await waitFor( + async () => + (await mongo.count(Collection.PARTICIPANTS, { + roomName, + identity, + status: "left", + closedBy: "event" + })) === 1, + 180, + "the session of the outage to be written" + ); + const participant = await mongo.findEntity(Collection.PARTICIPANTS, { roomName, identity }); + expect(participant!.durationMs).toBeGreaterThan(0); + const types = new Set((await mongo.findEvents({ roomName })).map((event) => event.type as string)); + expect(types).toContain("ROOM_CREATED"); + expect(types).toContain("PARTICIPANT_JOINED"); + expect(types).toContain("PARTICIPANT_LEFT"); + }, 600000); +}); diff --git a/openvidu-test-integration/tests/analytics-reconciler.test.ts b/openvidu-test-integration/tests/analytics-reconciler.test.ts new file mode 100644 index 000000000..1e24c8560 --- /dev/null +++ b/openvidu-test-integration/tests/analytics-reconciler.test.ts @@ -0,0 +1,173 @@ +import { killProcess, sleep, waitFor } from "./utils/helper"; +import { Livekit } from "./utils/livekit"; +import { LocalDeployment } from "./utils/local-deployment"; +import { MongoService } from "./utils/mongodb"; +import { Collection } from "./utils/types"; + +/** + * The reconciler of the OpenVidu server closes the rooms and participant sessions whose end event never arrives, + * with their tracks and subscriptions: those of a server that crashed or could not flush on shutdown. Healthy + * sessions stay open, also while Redis restarts. + */ +describe("OpenVidu analytics reconciler", () => { + const mongo = MongoService.getInstance(); + /** A crashed session is closed within this time of the restart (reconciler period of 10 s in CI). */ + const CLOSURE_TIMEOUT_SECONDS = 180; + /** A room is closed once its empty timeout (300 s) and two reconciler periods passed since its last event. */ + const ROOM_CLOSURE_TIMEOUT_SECONDS = 420; + /** Statistics are flushed every 30 s: the estimated end time is at most this early. */ + const STATS_INTERVAL_MS = 30_000; + /** How long healthy sessions are watched after Redis restarts. */ + const REDIS_RESTART_WATCH_SECONDS = 240; + let pids: number[] = []; + + beforeEach(async () => { + await LocalDeployment.start(); + }, 120000); + + afterEach(() => { + for (const pid of pids) { + killProcess(pid, "SIGKILL"); + } + pids = []; + LocalDeployment.stop(); + }, 120000); + + afterAll(async () => { + await mongo.disconnect(); + }); + + /** Joins publishers to a room and waits until every session is active with statistics; returns their sids. */ + async function startRoomWithPublishers(roomName: string, identities: string[]): Promise { + pids = identities.map((identity) => Livekit.joinParticipantToRoom(identity, roomName)); + await waitFor( + async () => + (await mongo.count(Collection.PARTICIPANTS, { + roomName, + identity: { $in: identities }, + status: "active", + lastStatsAt: { $exists: true } + })) === identities.length, + 120, + `the sessions of '${roomName}' to be active with statistics` + ); + const participants = await mongo.findEntities(Collection.PARTICIPANTS, { roomName, status: "active" }); + return participants.map((participant) => participant._id); + } + + /** The clients are gone with the server: no leave event will ever arrive. */ + function dropClients() { + for (const pid of pids) { + killProcess(pid, "SIGKILL"); + } + pids = []; + } + + /** + * Every session closed by the reconciler: left with the reason RECONCILED, an end time at most one statistics + * interval before the failure, one synthetic event each, and no track or subscription of theirs left open. + */ + async function expectClosedByReconciler(sids: string[], failedAt: Date) { + for (const sid of sids) { + const participant = await mongo.findEntity(Collection.PARTICIPANTS, { _id: sid }); + expect(participant).toMatchObject({ status: "left", closedBy: "reconciler", disconnectReason: "RECONCILED" }); + const leftAt = (participant!.leftAt as Date).getTime(); + expect(leftAt).toBeLessThanOrEqual(failedAt.getTime()); + expect(leftAt).toBeGreaterThanOrEqual(failedAt.getTime() - STATS_INTERVAL_MS); + expect(await mongo.findSyntheticEvent(sid)).toMatchObject({ + type: "PARTICIPANT_LEFT", + participantSid: sid, + synthetic: true + }); + } + expect(await mongo.countOpen(Collection.TRACKS, { participantSid: { $in: sids } })).toBe(0); + expect( + await mongo.countOpen(Collection.TRACK_SUBSCRIPTIONS, { + $or: [{ subscriberSid: { $in: sids } }, { publisherSid: { $in: sids } }] + }) + ).toBe(0); + } + + it("closes the sessions and the room of a crashed server", async () => { + const roomName = "ReconcilerCrash"; + const sids = await startRoomWithPublishers(roomName, ["crash-alice", "crash-bob"]); + + const crashedAt = new Date(); + LocalDeployment.killContainer("openvidu"); + dropClients(); + await sleep(5); + LocalDeployment.startContainer("openvidu"); + + await waitFor( + async () => (await mongo.countOpen(Collection.PARTICIPANTS, { _id: { $in: sids } })) === 0, + CLOSURE_TIMEOUT_SECONDS, + "the sessions of the crashed server to be closed" + ); + await expectClosedByReconciler(sids, crashedAt); + await waitFor( + async () => (await mongo.countOpen(Collection.ROOMS, { name: roomName })) === 0, + ROOM_CLOSURE_TIMEOUT_SECONDS, + "the room of the crashed server to be closed" + ); + expect(await mongo.findEntity(Collection.ROOMS, { name: roomName })).toMatchObject({ + status: "closed", + closedBy: "reconciler", + endReason: "RECONCILED" + }); + }, 900000); + + it("closes the sessions that a server stopped before its flush could not write", async () => { + const sids = await startRoomWithPublishers("ReconcilerStop", ["stop-alice", "stop-bob"]); + + // SIGTERM, then SIGKILL one second later: the shutdown flush of the analytics queues (up to 5 s) is cut short. + const stoppedAt = new Date(); + LocalDeployment.stopContainerWithTimeout("openvidu", 1); + dropClients(); + LocalDeployment.startContainer("openvidu"); + + await waitFor( + async () => (await mongo.countOpen(Collection.PARTICIPANTS, { _id: { $in: sids } })) === 0, + CLOSURE_TIMEOUT_SECONDS, + "the sessions of the stopped server to be closed" + ); + // A leave event that made it before the kill closes its session itself. + const reconciled = ( + await mongo.findEntities(Collection.PARTICIPANTS, { _id: { $in: sids }, closedBy: "reconciler" }) + ).map((participant) => participant._id); + await expectClosedByReconciler(reconciled, stoppedAt); + expect(await mongo.countOpen(Collection.TRACKS, { participantSid: { $in: sids } })).toBe(0); + }, 900000); + + it("keeps healthy sessions open while Redis restarts, and records their real leave events", async () => { + const roomName = "ReconcilerRedis"; + const sids = await startRoomWithPublishers(roomName, ["redis-alice", "redis-bob"]); + const runBefore = await mongo.schemaMeta("reconciler"); + + const restartedAt = new Date(); + LocalDeployment.restartContainer("redis"); + + const deadline = Date.now() + REDIS_RESTART_WATCH_SECONDS * 1000; + while (Date.now() < deadline) { + expect(await mongo.countOpen(Collection.PARTICIPANTS, { _id: { $in: sids } })).toBe(sids.length); + await sleep(10); + } + expect(await mongo.countOpen(Collection.ROOMS, { name: roomName })).toBe(1); + const runAfter = await mongo.schemaMeta("reconciler"); + expect((runAfter!.lastRunAt as Date).getTime()).toBeGreaterThan( + Math.max(restartedAt.getTime(), (runBefore?.lastRunAt as Date | undefined)?.getTime() ?? 0) + ); + + for (const pid of pids) { + killProcess(pid); + } + pids = []; + await waitFor( + async () => (await mongo.countOpen(Collection.PARTICIPANTS, { _id: { $in: sids } })) === 0, + 120, + "the sessions to be closed by their leave events" + ); + expect( + await mongo.count(Collection.PARTICIPANTS, { _id: { $in: sids }, closedBy: "event" }) + ).toBe(sids.length); + }, 900000); +}); diff --git a/openvidu-test-integration/tests/analytics-writer.test.ts b/openvidu-test-integration/tests/analytics-writer.test.ts new file mode 100644 index 000000000..e53c61e4a --- /dev/null +++ b/openvidu-test-integration/tests/analytics-writer.test.ts @@ -0,0 +1,124 @@ +import { killProcess, waitFor } from "./utils/helper"; +import { Livekit } from "./utils/livekit"; +import { LocalDeployment } from "./utils/local-deployment"; +import { MongoService } from "./utils/mongodb"; +import { Collection } from "./utils/types"; + +/** + * The OpenVidu server writes the analytics data model that OpenVidu Dashboard reads: one document per room, + * participant session, track and subscription, the raw events, the statistics and the rollups. + */ +describe("OpenVidu analytics writer", () => { + const mongo = MongoService.getInstance(); + let pids: number[] = []; + + beforeEach(async () => { + await LocalDeployment.start(); + }, 120000); + + afterEach(() => { + for (const pid of pids) { + killProcess(pid, "SIGKILL"); + } + pids = []; + LocalDeployment.stop(); + }, 120000); + + afterAll(async () => { + await mongo.disconnect(); + }); + + it("records a room with its participants, tracks, subscriptions, statistics, events and rollups", async () => { + const roomName = "WriterRoom"; + const identities = ["writer-alice", "writer-bob"]; + pids = identities.map((identity) => Livekit.joinParticipantToRoom(identity, roomName)); + + await waitFor( + async () => + (await mongo.count(Collection.PARTICIPANTS, { + roomName, + status: "active", + lastStatsAt: { $exists: true } + })) === identities.length, + 90, + "both participants to be active with statistics" + ); + const room = await mongo.findEntity(Collection.ROOMS, { name: roomName, status: "active" }); + expect(room).not.toBeNull(); + expect(room!.participants).toMatchObject({ active: 2, sessions: 2, unique: 2 }); + const roomSid = room!._id; + + expect(await mongo.countOpen(Collection.TRACKS, { roomSid })).toBeGreaterThanOrEqual(2); + await waitFor( + async () => (await mongo.countOpen(Collection.TRACK_SUBSCRIPTIONS, { roomSid })) >= 2, + 60, + "the participants to subscribe to each other" + ); + await waitFor( + async () => (await mongo.count(Collection.TRACK_STATS, { "meta.roomSid": roomSid })) > 0, + 60, + "statistics of the room" + ); + + killProcess(pids[0]); + await waitFor( + async () => + (await mongo.count(Collection.PARTICIPANTS, { + roomSid, + identity: identities[0], + status: "left", + closedBy: "event" + })) === 1, + 90, + "the first participant to leave" + ); + expect( + await mongo.countOpen(Collection.TRACK_SUBSCRIPTIONS, { + $or: [{ subscriberIdentity: identities[0] }, { publisherIdentity: identities[0] }], + roomSid + }) + ).toBe(0); + + killProcess(pids[1]); + await waitFor( + async () => (await mongo.count(Collection.ROOMS, { _id: roomSid, status: "closed", closedBy: "event" })) === 1, + 120, + "the room to end after its departure timeout" + ); + const closed = await mongo.findEntity(Collection.ROOMS, { _id: roomSid }); + expect(closed!.endedAt).toBeInstanceOf(Date); + expect(closed!.durationMs).toBeGreaterThan(0); + expect(closed!.participants).toMatchObject({ active: 0, sessions: 2 }); + expect(await mongo.countOpen(Collection.PARTICIPANTS, { roomSid })).toBe(0); + expect(await mongo.countOpen(Collection.TRACKS, { roomSid })).toBe(0); + expect(await mongo.countOpen(Collection.TRACK_SUBSCRIPTIONS, { roomSid })).toBe(0); + + const events = await mongo.findEvents({ roomSid }); + const types = new Set(events.map((event) => event.type as string)); + for (const type of [ + "ROOM_CREATED", + "PARTICIPANT_JOINED", + "PARTICIPANT_ACTIVE", + "TRACK_PUBLISHED", + "PARTICIPANT_LEFT", + "ROOM_ENDED" + ]) { + expect(types).toContain(type); + } + // API calls are only counted in the rollups, and credentials never reach the database. + expect(types).not.toContain("API_CALL"); + expect(JSON.stringify(events)).not.toContain("turn_password"); + + const hourly = await mongo.findEntities(Collection.ANALYTICS_HOURLY, { + scope: "global", + bucket: { $gte: new Date(Date.now() - 2 * 3_600_000) } + }); + const total = (read: (doc: Record) => number | undefined) => + hourly.reduce((sum, doc) => sum + (read(doc) ?? 0), 0); + expect(total((doc) => doc.rooms?.created)).toBeGreaterThanOrEqual(1); + expect(total((doc) => doc.participants?.joined)).toBeGreaterThanOrEqual(2); + expect(total((doc) => doc.participants?.left)).toBeGreaterThanOrEqual(2); + + expect((await mongo.schemaMeta("schema"))?.version).toBeGreaterThanOrEqual(1); + }, 600000); +}); diff --git a/openvidu-test-integration/tests/openvidu-active-entities-fixer.test.ts b/openvidu-test-integration/tests/openvidu-active-entities-fixer.test.ts deleted file mode 100644 index b9f85258f..000000000 --- a/openvidu-test-integration/tests/openvidu-active-entities-fixer.test.ts +++ /dev/null @@ -1,95 +0,0 @@ -import { killProcess, sleep } from "./utils/helper"; -import { Livekit } from "./utils/livekit"; -import { LocalDeployment } from "./utils/local-deployment"; -import { MongoService } from "./utils/mongodb"; -import { EntityType } from "./utils/types"; - -describe("OpenVidu active entities fixer", () => { - const mongoService = MongoService.getInstance(); - - beforeEach(async () => { - await LocalDeployment.start(); - }, 60000); // 1 minute - - afterEach(() => { - LocalDeployment.stop(); - }, 60000); // 1 minute - - afterAll(async () => { - await mongoService.disconnect(); - }); - - it("should fix fake active entities in MongoDB", async () => { - console.log("Joining participant to room..."); - const roomName = "TestRoom"; - const participantIdentity = "TestParticipant1"; - const pid = Livekit.joinParticipantToRoom(participantIdentity, roomName); - await sleep(5); - - const roomStartedEvent = await mongoService.findStartEvent(EntityType.ROOM, roomName); - if (!roomStartedEvent) { - throw new Error("ROOM_CREATED event not found"); - } - - const participantActiveEvent = await mongoService.findStartEvent(EntityType.PARTICIPANT, participantIdentity); - if (!participantActiveEvent) { - throw new Error("PARTICIPANT_ACTIVE event not found"); - } - - const roomId = roomStartedEvent.room.sid; - const participantId = participantActiveEvent.participant_id; - - LocalDeployment.stopContainer("openvidu"); - killProcess(pid); - await sleep(5); - LocalDeployment.startContainer("openvidu"); - - // Check if there is a fake close event for room and participant in MongoDB - // and the active entities are removed - let roomEndedEvent; - let participantLeftEvent; - let isRoomActive: boolean; - let isParticipantActive: boolean; - - do { - console.log("Waiting for fake close events in MongoDB..."); - await sleep(10); - roomEndedEvent = await mongoService.findCloseEvent(EntityType.ROOM, roomId); - participantLeftEvent = await mongoService.findCloseEvent(EntityType.PARTICIPANT, participantId); - isRoomActive = await mongoService.isActiveEntity(roomId); - isParticipantActive = await mongoService.isActiveEntity(participantId); - } while (!roomEndedEvent || !participantLeftEvent || isRoomActive || isParticipantActive); - - // Check properties of fake ROOM_ENDED event - expect(isRoomActive).toBeFalsy(); - expect(roomEndedEvent).not.toBeNull(); - expect(roomEndedEvent).toMatchObject({ - room_id: roomId, - room: { - sid: roomId, - name: roomName, - creation_time: roomStartedEvent.room.creation_time - } - }); - expect(roomEndedEvent!.timestamp.seconds).toBeGreaterThan(roomStartedEvent.room.creation_time); - expect(roomEndedEvent!.openvidu_expire_at).toBeDefined(); - - // Check properties of fake PARTICIPANT_LEFT event - expect(isParticipantActive).toBeFalsy(); - expect(participantLeftEvent).not.toBeNull(); - expect(participantLeftEvent).toMatchObject({ - room_id: roomId, - room: { - sid: roomId - }, - participant_id: participantId, - participant: { - sid: participantId, - identity: participantIdentity, - joined_at: participantActiveEvent.participant.joined_at - } - }); - expect(participantLeftEvent!.timestamp.seconds).toBeGreaterThan(participantActiveEvent.participant.joined_at); - expect(participantLeftEvent!.openvidu_expire_at).toBeDefined(); - }, 300000); // 5 minutes -}); diff --git a/openvidu-test-integration/tests/utils/helper.ts b/openvidu-test-integration/tests/utils/helper.ts index 07cfc6d1b..8d73778d2 100644 --- a/openvidu-test-integration/tests/utils/helper.ts +++ b/openvidu-test-integration/tests/utils/helper.ts @@ -30,8 +30,12 @@ export const execCommandInBackground = (command: string, args: string[]): number return child.pid; }; -export const killProcess = (pid: number) => { - process.kill(pid); +export const killProcess = (pid: number, signal: NodeJS.Signals = "SIGTERM") => { + try { + process.kill(pid, signal); + } catch { + // Already exited. + } }; export const sleep = async (seconds: number) => { @@ -39,3 +43,19 @@ export const sleep = async (seconds: number) => { setTimeout(resolve, seconds * 1000); }); }; + +/** + * Polls a condition every 2 seconds until it holds, failing with the description after the timeout. + */ +export const waitFor = async (condition: () => Promise, timeoutSeconds: number, description: string) => { + const deadline = Date.now() + timeoutSeconds * 1000; + + while (Date.now() < deadline) { + if (await condition().catch(() => false)) { + return; + } + await sleep(2); + } + + throw new Error(`Timed out after ${timeoutSeconds} s waiting for ${description}`); +}; diff --git a/openvidu-test-integration/tests/utils/legacy.ts b/openvidu-test-integration/tests/utils/legacy.ts new file mode 100644 index 000000000..fb080fcb2 --- /dev/null +++ b/openvidu-test-integration/tests/utils/legacy.ts @@ -0,0 +1,71 @@ +import { Document, Double } from "mongodb"; + +export const LEGACY_ROOM_SID = "RM_LegacyRoom0001"; +export const LEGACY_ROOM_NAME = "legacy-room"; +export const LEGACY_NODE_ID = "ND_LegacyNode001"; +export const LEGACY_PARTICIPANTS = [ + { sid: "PA_LegacyAlice001", identity: "legacy-alice", joined: 60, left: 1_260 }, + { sid: "PA_LegacyBob00001", identity: "legacy-bob", joined: 120, left: 1_800 } +]; + +/** A legacy analytics event as the writer before 3.10 stored it: numbers as doubles, timestamps in seconds. */ +function legacyEvent(id: string, type: string, seconds: number, fields: Document): Document { + return { + _id: id, + id: `AE_${id}`, + type, + timestamp: { seconds: new Double(seconds), nanos: new Double(0) }, + room_id: LEGACY_ROOM_SID, + node_id: LEGACY_NODE_ID, + openvidu_expire_at: new Date((seconds + 32 * 86_400) * 1000), + ...fields + }; +} + +/** One room of two participants that ended an hour ago, in the legacy shape. */ +export function legacyRoom(start: number): Document[] { + const room = { + sid: LEGACY_ROOM_SID, + name: LEGACY_ROOM_NAME, + empty_timeout: new Double(300), + departure_timeout: new Double(20), + creation_time: new Double(start), + creation_time_ms: new Double(start * 1000), + turn_password: "legacy-turn-password" + }; + const participant = (p: (typeof LEGACY_PARTICIPANTS)[number], state: string) => ({ + sid: p.sid, + identity: p.identity, + state, + joined_at: new Double(start + p.joined), + joined_at_ms: new Double((start + p.joined) * 1000), + permission: { can_publish: true, can_subscribe: true, can_publish_data: true } + }); + const events = [legacyEvent("legacy-room-created", "ROOM_CREATED", start, { room })]; + + for (const p of LEGACY_PARTICIPANTS) { + events.push( + legacyEvent(`${p.sid}-joined`, "PARTICIPANT_JOINED", start + p.joined, { + room, + participant_id: p.sid, + participant: participant(p, "JOINED"), + client_info: { sdk: "JS", os: "linux", version: "2.15.0" } + }), + legacyEvent(`${p.sid}-active`, "PARTICIPANT_ACTIVE", start + p.joined + 1, { + room, + participant_id: p.sid, + participant: participant(p, "ACTIVE"), + client_meta: { client_connect_time: new Double(800), connection_type: "udp" } + }), + legacyEvent(`${p.sid}-left`, "PARTICIPANT_LEFT", start + p.left, { + room, + participant_id: p.sid, + participant: { ...participant(p, "DISCONNECTED"), disconnect_reason: new Double(1) } + }) + ); + } + + events.push(legacyEvent("legacy-room-ended", "ROOM_ENDED", start + 1_820, { room_end_reason: new Double(2) })); + + return events; +} diff --git a/openvidu-test-integration/tests/utils/livekit.ts b/openvidu-test-integration/tests/utils/livekit.ts index 1e8539e16..e64692964 100644 --- a/openvidu-test-integration/tests/utils/livekit.ts +++ b/openvidu-test-integration/tests/utils/livekit.ts @@ -7,6 +7,7 @@ const LIVEKIT_API_SECRET = "secret"; const LK_CLI = "lk"; export class Livekit { + /** Joins a participant that publishes the demo video and audio tracks; returns the pid of the CLI. */ static joinParticipantToRoom(participantIdentity: string, roomName: string): number { const args = [ "room", diff --git a/openvidu-test-integration/tests/utils/local-deployment.ts b/openvidu-test-integration/tests/utils/local-deployment.ts index 161669674..c2050aeb5 100644 --- a/openvidu-test-integration/tests/utils/local-deployment.ts +++ b/openvidu-test-integration/tests/utils/local-deployment.ts @@ -31,4 +31,37 @@ export class LocalDeployment { console.log(`Stopping container ${containerName}...`); execCommand(`docker compose -f ${LOCAL_DEPLOYMENT_PATH} stop ${containerName}`); } + + /** SIGTERM and, after the timeout, SIGKILL: a short timeout cuts the shutdown of the service short. */ + static stopContainerWithTimeout(containerName: string, timeoutSeconds: number) { + console.log(`Stopping container ${containerName} with a ${timeoutSeconds} s timeout...`); + execCommand(`docker stop --timeout=${timeoutSeconds} ${containerName}`); + } + + /** SIGKILL, as a crash: the service gets no chance to shut down. */ + static killContainer(containerName: string) { + console.log(`Killing container ${containerName}...`); + execCommand(`docker kill ${containerName}`); + } + + static restartContainer(containerName: string) { + console.log(`Restarting container ${containerName}...`); + execCommand(`docker restart ${containerName}`); + } + + /** Waits until the OpenVidu server logs that its analytics database is initialized after the given time. */ + static async waitForAnalytics(since: Date, timeoutSeconds = 60) { + const deadline = Date.now() + timeoutSeconds * 1000; + const sinceArg = since.toISOString(); + + while (Date.now() < deadline) { + const logs = execCommand(`docker logs --since ${sinceArg} openvidu 2>&1 || true`); + if (logs.includes("analytics database initialized")) { + return; + } + await sleep(1); + } + + throw new Error("The OpenVidu server did not initialize its analytics database"); + } } diff --git a/openvidu-test-integration/tests/utils/mongodb.ts b/openvidu-test-integration/tests/utils/mongodb.ts index e07b53531..688b79a4d 100644 --- a/openvidu-test-integration/tests/utils/mongodb.ts +++ b/openvidu-test-integration/tests/utils/mongodb.ts @@ -1,98 +1,86 @@ -import { Collection, Db, Document, MongoClient, WithId } from "mongodb"; -import { ActiveEntity, EntityType } from "./types"; +import { Db, Document, Filter, MongoClient } from "mongodb"; +import { Collection } from "./types"; const DATABASE_URL = "mongodb://mongoadmin:mongoadmin@localhost:27017/?replicaSet=rs0&readPreference=primaryPreferred&directConnection=true"; const DATABASE_NAME = "openvidu"; -const EVENTS_COLLECTION = "events"; -const ACTIVE_ENTITIES_COLLECTION = "active-entities"; -const entityStartEventTypesMap: Map = new Map([ - [EntityType.ROOM, "ROOM_CREATED"], - [EntityType.PARTICIPANT, "PARTICIPANT_ACTIVE"] -]); -const entityCloseEventTypesMap: Map = new Map([ - [EntityType.ROOM, "ROOM_ENDED"], - [EntityType.PARTICIPANT, "PARTICIPANT_LEFT"] -]); -const entityIdFieldMap: Map = new Map([ - [EntityType.ROOM, "room.sid"], - [EntityType.PARTICIPANT, "participant.sid"] -]); -const entityNameFieldMap: Map = new Map([ - [EntityType.ROOM, "room.name"], - [EntityType.PARTICIPANT, "participant.identity"] -]); +/** Statuses of the entities that are still open, per collection. */ +const OPEN_STATUSES: Partial> = { + [Collection.ROOMS]: ["active"], + [Collection.PARTICIPANTS]: ["active", "joining"], + [Collection.TRACKS]: ["published"], + [Collection.TRACK_SUBSCRIPTIONS]: ["requested", "subscribed"] +}; + +/** A document of the analytics data model, whose ids are strings. */ +export interface AnalyticsDocument extends Document { + _id: string; +} + +/** A query on documents whose ids are strings. */ +export type Query = Filter; export class MongoService { - private static instace: MongoService; + private static instance: MongoService; private client: MongoClient; private db: Db; - private eventsCollection: Collection; - private activeEntitiesCollection: Collection; private constructor() { - this.client = new MongoClient(DATABASE_URL); + this.client = new MongoClient(DATABASE_URL, { serverSelectionTimeoutMS: 30000 }); this.db = this.client.db(DATABASE_NAME); - this.eventsCollection = this.db.collection(EVENTS_COLLECTION); - this.activeEntitiesCollection = this.db.collection(ACTIVE_ENTITIES_COLLECTION); } public static getInstance(): MongoService { - if (!MongoService.instace) { - MongoService.instace = new MongoService(); + if (!MongoService.instance) { + MongoService.instance = new MongoService(); } - return MongoService.instace; - } - - public async connect(): Promise { - try { - await this.client.connect(); - } catch (error) { - console.error("Error connecting to MongoDB", error); - throw error; - } + return MongoService.instance; } public async disconnect(): Promise { - try { - await this.client.close(); - } catch (error) { - console.error("Error disconnecting from MongoDB", error); - throw error; - } + await this.client.close(); } - public async findStartEvent(entityType: EntityType, entityName: string): Promise | null> { - try { - const eventType = entityStartEventTypesMap.get(entityType)!; - const nameField = entityNameFieldMap.get(entityType)!; - return await this.eventsCollection.findOne({ type: eventType, [nameField]: entityName }); - } catch (error) { - console.error("Error finding start event", error); - throw error; - } + public database(): Db { + return this.db; } - public async findCloseEvent(entityType: EntityType, entityId: string): Promise | null> { - try { - const eventType = entityCloseEventTypesMap.get(entityType)!; - const idField = entityIdFieldMap.get(entityType)!; - return await this.eventsCollection.findOne({ type: eventType, [idField]: entityId }); - } catch (error) { - console.error("Error finding close event", error); - throw error; - } + public async findEntity(collection: Collection, filter: Query): Promise { + return this.db.collection(collection).findOne(filter); } - public async isActiveEntity(id: string): Promise { - try { - const result = await this.activeEntitiesCollection.findOne({ _id: id }); - return !!result; - } catch (error) { - console.error("Error finding active entity", error); - throw error; - } + public async findEntities(collection: Collection, filter: Query): Promise { + return this.db + .collection(collection) + .find(filter) + .toArray(); + } + + public async count(collection: Collection, filter: Query): Promise { + return this.db.collection(collection).countDocuments(filter); + } + + /** Documents of the collection matching the filter that are still open. */ + public async countOpen(collection: Collection, filter: Query): Promise { + return this.count(collection, { ...filter, status: { $in: OPEN_STATUSES[collection] ?? [] } }); + } + + public async findEvents(filter: Query): Promise { + return this.db.collection(Collection.EVENTS).find(filter).sort({ ts: 1 }).toArray(); + } + + /** The synthetic close event the reconciler inserted for an entity. */ + public async findSyntheticEvent(entityId: string): Promise { + return this.db.collection(Collection.EVENTS).findOne({ _id: `AE_SYN_${entityId}`, synthetic: true }); + } + + public async schemaMeta(id: string): Promise { + return this.db.collection(Collection.SCHEMA_META).findOne({ _id: id }); + } + + public async collectionNames(): Promise { + return (await this.db.listCollections({}, { nameOnly: true }).toArray()).map((info) => info.name); } } diff --git a/openvidu-test-integration/tests/utils/types.ts b/openvidu-test-integration/tests/utils/types.ts index cbeafbd76..b7ed3353c 100644 --- a/openvidu-test-integration/tests/utils/types.ts +++ b/openvidu-test-integration/tests/utils/types.ts @@ -1,9 +1,12 @@ -export const enum EntityType { - ROOM = "room", - PARTICIPANT = "participant" -} - -export interface ActiveEntity { - _id: string; - entity: EntityType; +/** Collections of the analytics data model written by the OpenVidu server. */ +export const enum Collection { + ROOMS = "rooms", + PARTICIPANTS = "participants", + TRACKS = "tracks", + TRACK_SUBSCRIPTIONS = "track_subscriptions", + EVENTS = "events", + TRACK_STATS = "track_stats", + ANALYTICS_HOURLY = "analytics_hourly", + SCHEMA_META = "schema_meta", + LEGACY_EVENTS = "legacy_events" }