diff --git a/.changeset/client-telemetry-pipeline.md b/.changeset/client-telemetry-pipeline.md new file mode 100644 index 0000000000..a7ebf08690 --- /dev/null +++ b/.changeset/client-telemetry-pipeline.md @@ -0,0 +1,5 @@ +--- +'livekit-client': minor +--- + +Client telemetry: the SDK can now report its own sessions as OTLP to a collector — `lk.connect` and `lk.reconnect` spans with their checkpoints, `lk.publish` and `lk.subscribe`, `lk.rtc.stats.sample` windows folded from the readings the track monitors already take, `lk.room.disconnected`, and the device state a page can observe. Records are batched write-ahead into a cache and uploaded under a policy that never lets telemetry win over media: one request in flight, holds while connecting, a 429 that keeps its batch, and a per-tick budget for replaying a backlog. Off unless `Telemetry.configure({ endpoint })` is called or the room is on LiveKit Cloud, which derives the route and the token from the connect. See TELEMETRY.md. diff --git a/.size-limit.cjs b/.size-limit.cjs index 81a6ad3fb3..1546b14e5f 100644 --- a/.size-limit.cjs +++ b/.size-limit.cjs @@ -7,6 +7,9 @@ module.exports = [ { path: 'dist/livekit-client.umd.js', import: '{ Room }', - limit: '130 kB', + // Telemetry costs +8.95 kB brotli, and UMD cannot shake it out: 130 kB no longer fits. + // Raising it is one of two answers — the other is a separate UMD entry point, as the + // e2ee and frame-metadata workers already have. See TELEMETRY.md. + limit: '135 kB', }, ]; diff --git a/TELEMETRY.md b/TELEMETRY.md new file mode 100644 index 0000000000..9c46b91648 --- /dev/null +++ b/TELEMETRY.md @@ -0,0 +1,282 @@ +# Client telemetry — JS ecosystem design + +How `livekit-client` (browser) and `@livekit/react-native` get the pipeline that the Swift, Kotlin +and Dart SDKs get from the Rust core. `livekit-telemetry/SPEC.md` in `rust-sdks` stays the source of +truth for event names, attributes and cadences; this document is only about what carries them here. + +## 1. OpenTelemetry SDK, the Rust core via WASM, or our own? + +**Our own policy, OpenTelemetry's encoder.** One new runtime dependency — +`@opentelemetry/otlp-transformer`, pinned — turns plain record objects into an OTLP request body in +protobuf or JSON. Nothing else from OpenTelemetry ships. + +The budget first, from the repo's own `pnpm size-limit` (webpack, minified, **brotli**): + +| `pnpm size-limit` entry | before | with the encoder | limit | +|---|---|---|---| +| `{ Room }` from `dist/livekit-client.esm.mjs` | 114.24 kB | 118.75 kB | 150 kB | +| `dist/livekit-client.umd.js` (not tree-shakeable — every app pays) | 123.58 kB | **128.01 kB** | 130 kB | + +So the encoder costs ~4.4 kB brotli where it cannot be shaken out, and leaves **2 kB** under the +UMD limit. That is the number every option has to fit in. Measured against it with esbuild +(`--bundle --minify`, `gzip -9`), `@opentelemetry/*` 0.222.0 / 2.11.0: + +| Bundle | minified | gzipped | +|---|---|---| +| `otlp-transformer` proto serializers, logs + traces | 17.8 KB | **5.2 KB** | +| logs SDK + OTLP/HTTP **JSON** exporter | 42.9 KB | 13.5 KB | +| logs SDK + OTLP/HTTP **proto** exporter | 49.4 KB | 15.2 KB | +| logs + traces SDKs + proto exporters | 78.2 KB | **23.0 KB** | + +What the full SDK adds over the serializers is a batch processor, a fetch transport and the +provider/context plumbing: ~18 KB gzipped on top, four times the UMD headroom, for three things +that do not fit. + +- **The trace model is different.** Our trace is a *scope* — one Room connection across reconnects, + its id minted by the pipeline and stamped on every span and log record. OTel's tracer wants a + context tree with a propagator and an active-span mechanism; we would be fighting it to get one + long-lived id onto records that have no active span. +- **The batching is the upload policy.** `BatchLogRecordProcessor` is a timer and a queue. + SPEC's policy is holds while `lk.connect` is open, one request in flight, a per-tick budget, a + 60 s cap, a flood guard and a self-report. That is written either way; the processor would sit + underneath it doing a second, conflicting round of batching. +- **Logs are pre-1.0.** `@opentelemetry/sdk-logs` is `0.222.0` and takes breaking changes across + minors. `livekit-client` is a library on a hard size budget that many apps pin; the smaller the + exposed surface of an experimental dependency, the better. The serializers are the smallest + useful unit, and their input is structural — plain objects, no classes to construct. + +Two things we take from the SDK as knowledge rather than code, because it has been through this: +the retryable status set (`429`, `502`, `503`, `504`, honouring `Retry-After`), which is already +what the Rust transport does; and the keepalive accounting — browsers cap *all* in-flight +`fetch(keepalive)` bodies at 64 KiB together and Chromium at 9 concurrent requests, so the last +flush of a page has to be small. + +**Not the Rust core via WASM.** Hermes has no WebAssembly, so React Native could never share it, +and sharing is the entire reason to consider it. A browser-only WASM core would be a third +implementation, not a second. + +## 2. Less custom logic than the device SDKs + +Yes — roughly half of the core does not exist here. + +| Dropped | Why | +|---|---| +| Write-ahead file cache, gzip-on-disk, 24 h age prune, next-launch replay | no disk (§3) | +| The FFI layer — UniFFI types, callback interfaces, the transport trait | the pipeline is in the same language as the SDK | +| `record_stats_report` raw-entry parsing | `RTCStatsReport` is already the SDK's own shape (`src/room/stats.ts`, `monitorFrequency = 2000`) | + +Device state is not dropped, but it is not uniform either — the pipeline never measures anything +itself, so an `lk.device.*` record exists only where the platform will say so: + +| SPEC event | Browser | React Native | +|---|---|---| +| `app_state.changed` | **yes**, `document.visibilityState` | **yes**, `AppState` | +| `network.changed` | **Chromium**, `navigator.connection` — and `type` is populated on Android only, so a desktop page reports `unknown` | via `@react-native-community/netinfo`, an app-owned dependency — not wired | +| `capture.failed` | the `getUserMedia` DOMException names, per SPEC | the same errors through `react-native-webrtc` — not wired | +| `thermal.changed` | `PressureObserver` is Chromium **desktop** only — not wired | **yes**, native | +| `low_power.changed` | no web API | **yes**, native | +| `memory.changed` | no web API (`deviceMemory` is a static figure) | **yes**, native | +| `battery.changed` | Chromium only; Firefox removed it, Safari never shipped it — not wired | native, not wired | +| `audio_route.changed`, `audio.interruption` | `devicechange` says the set changed, not that it went speaker → bluetooth | the package already owns the audio session — not wired | + +React Native can reach parity with the Swift and Kotlin SDKs because `@livekit/react-native` already +ships a native module; a page structurally cannot. The cadence factors those signals drive are +implemented (`device.ts`, capped at 4× as SPEC says), so a page stretches on background and Data +Saver, and an app stretches on everything. + +| Kept | Why | +|---|---| +| Scope = trace per Room, `session.id` on every record | the whole query story depends on it | +| `lk.rtc.stats.sample` windowing (15 s, counters + min/max/avg) | the reason the project exists | +| Bounded queue, batch size/byte caps, oldest-first eviction, counted | §3 makes the queue the *only* bound | +| Upload holds during `lk.connect` / `lk.reconnect`, 60 s cap, one request in flight | telemetry never wins over media | +| Flood guard, `lk.telemetry.report` | fleet-wide denominators, same shape as everywhere else | + +## 3. Caching: in-memory in a tab, whatever the platform has elsewhere + +The pipeline is **write-ahead**, as the Rust core is: a batch is stored before the network is tried +and removed only once the collector has taken it, so a refused upload, a lost connection or a +process that dies costs nothing. Where it is stored is `TelemetryStorage` — five synchronous +operations, the same five the core's `BatchCache` has — and the default keeps batches in memory, +bounded by 4 MiB and 512 batches, oldest evicted first and counted. + +A batch id carries everything needed to send a batch an earlier run of the app wrote — its route, +its record count, and the encoding it was written with. That last one is not hypothetical: the +first React Native replay lost 40 records because cached JSON batches were re-sent with the +protobuf content type, and the only reason anyone noticed is that `lk.telemetry.report` said +`dropped.rejected: 40` instead of quietly succeeding. + +The in-memory default is right for a tab, and nobody in this ecosystem persists there either. + +- **OpenTelemetry JS** caches nothing: `BatchLogRecordProcessor` is a bounded in-memory queue and + the spec puts retry on the exporter, explicitly not on the processor. +- **Sentry browser** is in-memory; offline caching is opt-in, by wrapping the transport in + `makeBrowserOfflineTransport` (IndexedDB). +- **Datadog browser-SDK** keeps batches in memory and flushes on `visibilitychange`. +- **Grafana Faro** is in-memory. + +A tab's lifetime is the call's lifetime; there is no "app killed in the background, replay at next +launch". The last-gasp flush is best effort: `fetch(keepalive)` under 64 KiB on `pagehide`, an +ordinary `fetch` on React Native's `AppState` → background. Neither turns a disappearing page into +a durable queue, which is why the flush happens at `visibilitychange → hidden` and not at `unload`. + +React Native is the case where it is not right, because an app really can be killed holding a +backlog and really can be offline for hours. There it supplies a store backed by files — the same +shape the Rust core's `FileCache` has, written natively in the package that already ships native +code rather than pulled in as a dependency. Nothing about that reaches this package: it hands over +`storage` and knows nothing else. A browser could do the same over IndexedDB if the field ever +shows it is worth it. + +## 4. Protobuf or JSON? + +**Protobuf by default, JSON behind a switch.** Both come from the same package, so this is one +line, and the PoC posts both. + +- Smaller: one `lk.ping` with a resource and three attributes is **326 bytes of protobuf against + 846 bytes of JSON** (measured in the React Native PoC below, same record, same encoder package). + JSON repeats every key as a string, which is what the OTel issue tracker cites for browsers in + the first place. +- **JSON's ids do not survive LiveKit Cloud today.** OTLP/JSON mandates *hex* `traceId`/`spanId` + (`otlp-transformer`'s `JSON_ENCODER` passes them through as hex; `PROTOBUF_ENCODER` converts to + bytes). Posting hex ids to staging landed records whose trace and span ids were **zero** — the + ingest reads them as stock protojson `bytes`, i.e. base64. Protobuf is unaffected. Filing this is + worth it regardless, because every standard OTLP/JSON client hits it. +- `Content-Encoding: gzip` is *not* in the v1 design: `CompressionStream` is Chrome 80+, Safari + 16.4+, Firefox 113+ and absent in Hermes, so it would have to be feature-detected for a win + protobuf already mostly delivers. + +## 5. Reusable by React Native + +By construction: `@livekit/react-native` depends on `livekit-client`, so telemetry that ships in +this package is in RN with no second implementation. The rules that keep it that way: + +- No DOM or `document`/`navigator` access outside the one lifecycle seam. Everything else — + records, serialization, batching, transport — is plain JS and `fetch`. +- No `crypto.getRandomValues` requirement: ids need to be unique, not unguessable, so `Math.random` + is the fallback and RN needs no `react-native-get-random-values`. +- `TextEncoder` is only on the JSON path (the protobuf serializer writes its own bytes); RN + polyfills it in `registerGlobals` anyway. +- A binary body is fine: RN's `convertRequestBody` base64-encodes an `ArrayBuffer`/`ArrayBufferView` + for the bridge. It costs ~33 % in-process, nothing on the wire. +- The seam: `pagehide` / `visibilitychange` in a page, `AppState` in an app; `navigator.connection` + in Chromium, nothing (or `@react-native-community/netinfo`, an app-owned dependency) in RN. + +**React Native follows the phones, and reuses this package's instrumentation.** The decision is +that `@livekit/react-native` binds the Rust core through UniFFI, the way the Swift, Kotlin and Dart +SDKs do, so that an app on a phone behaves identically whichever SDK it used — same windowing code, +same upload policy, same write-ahead file cache, same bytes on the wire. What it does *not* do is +reimplement the instrumentation: when a connect span starts, which checkpoints it carries, how a +subscribe ends at first media, which `getStats` fields become a window — all of that stays here and +is reused. + +**React Native reuses all of it and replaces one thing: where batches wait.** The caching +semantics of the native SDKs — a batch on disk that survives the process and replays at next launch +— do not need the Rust core, only a directory of files. `@livekit/react-native` supplies a +`TelemetryStorage` backed by `LKBatchStore.swift` and `BatchStore.kt`, a native store mirroring +the core's `FileCache`, and this package learns nothing about it beyond five synchronous calls. + +``` +Room, LocalParticipant, the four track monitors ← the instrumentation, one copy + │ + Pipeline ← the policy, one copy + │ + TelemetryStorage ← the one seam with two implementations + ╱ ╲ + MemoryStorage (a tab) a directory of files (@livekit/react-native) +``` + +An earlier round put a whole `Backend` interface here so React Native could bind the Rust core for +the entire pipeline. It is gone: it had one implementation and one hypothetical one, which is the +definition of an abstraction to write later. If that question ever resolves the other way, the git +history has the shape and `Pipeline` is small enough to sit behind an interface again in an hour. + +**Nothing mobile appears on this side either.** `DeviceState` carries only what a page can +answer — visibility and, on Chromium, the connection. Thermal state, low power mode and memory +pressure are not absent because they are unimportant; they are absent because this package cannot +observe them and must not pretend to. React Native's native monitors (`LKDeviceState.swift`, +`DeviceStateMonitor.kt`) map them onto SPEC's event names and onto a cadence factor *in that +package*, and hand the result over through two verbs that name no platform: `Telemetry.emit`, which +takes a record this package never interprets, and `Telemetry.setCadenceFactor`, which takes a +number without a reason attached. + +One thing the PoC found, which is about this package rather than telemetry: `livekit-client` +evaluates `class … extends DOMException` and `new TextDecoder()` at **module scope**, and Hermes has +neither, so anything importing it must come after the React Native polyfills. That is why +`registerTelemetry` is imported below them in `src/index.tsx` — the order is load-bearing. + +## Shape + +``` +src/telemetry/ + index.ts the Telemetry facade: configure/setServer, the scope factory, the track registry + pipeline.ts queue, flush timer, holds, 429/5xx, one request in flight, the self-report + scope.ts one per Room connection: trace id, attributes, spans, stats windows + storage.ts the seam: where batches wait between being made and being accepted (see §3) + otlp.ts the records and their wire form (the only OpenTelemetry import lives here) + webrtc.ts the SDK's typed sender/receiver stats → one SPEC reading +``` + +Where the SDK calls it — eleven places, all of them one line except the connect span: + +| Site | What it says | +|---|---| +| `Room.connect` | a Cloud URL names the destination; a scope is opened and `lk.connect` starts, holding uploads | +| `Room.attemptConnection` | `signal`, `join_recv`, `pc_connected`, `room_connected` checkpoints, then the span ends | +| `Room.applyJoinResponse` / `handleRoomUpdate` | `lk.room.*`, `lk.participant.*` on every record of the call | +| `EngineEvent.Resuming` / `Restarting` / `Resumed` / `handleSignalRestarted` | one `lk.reconnect` span per reconnect, one checkpoint per attempt, mode = whichever won | +| `Room.handleDisconnect` | open spans fail, `lk.room.disconnected`, the open windows close, one last upload | +| `ParticipantEvent.TrackSubscribed` / `Unsubscribed` | `lk.subscribe`, ended by the first inbound window with bytes | +| `Room.onLocalTrackPublished` / `Unpublished` | which scope and direction a track sid belongs to | +| `LocalParticipant.publishTrack` | `lk.publish` | +| the four track monitors | the reading they already took, every 2 s | + +`RTCEngine` gained one field: the reconnect reason, so `Resuming`/`Restarting` can carry it. +`RemoteAudioTrack.getReceiverStats` gained `packetsReceived` / `packetsLost`, which its own +`ReceiverStats` type already declared and nothing was filling in. + +## What this design still owes an answer + +- **The UMD budget — now measured, and over.** The finished integration costs **+8.95 kB brotli**: + ESM `{ Room }` goes 114.24 → 123.19 kB (limit 150 kB, comfortable), and the UMD bundle, which + cannot shake anything out, goes 123.58 → **132.70 kB against a 130 kB limit**. This branch raises + the UMD limit to 135 kB so the build passes; the alternative is a separate UMD entry point, the + way the e2ee and frame-metadata workers already have one. That is a call for the SDK's owners. +- **What the SDK's typed stats do not carry.** The windows are folded from the readings the track + monitors already take, so they cost no extra `getStats()` — but those readings are a subset of + SPEC. Missing: inbound RTT and jitter-buffer counters, video freeze and pause counts, audio + level and interruptions. Each is a field to add to `getSenderStats` / `getReceiverStats`, which + parse the raw report already; `packetsReceived` / `packetsLost` on inbound audio were added here + as the first of them. +- **The OTLP/JSON id bug on LiveKit Cloud** (see §4) — worth filing whichever encoding we ship. + +## Tests + +`pnpm test` covers the policy against a stubbed collector, reading the JSON bodies it would have +sent: a window's counters and gauges, a hold that stops uploads and not collection, a 429 that keeps +its batch, the queue evicting oldest-first and saying so, a span's checkpoints and outcome. + +`pnpm vitest run --config vitest.telemetry.config.mts` is the session: a real Chromium with fake +media devices, a real `livekit-server --dev`, two Rooms in one page, both reconnect paths, and the +collector that fans out to the same Grafana LGTM stack the mobile harness writes to. + +```sh +livekit-server --dev +otelcol-contrib --config src/telemetry/otelcol-web.yaml # :4320, CORS, fans out to :4318 LGTM +pnpm vitest run --config vitest.telemetry.config.mts +``` + +One run puts this in the collector: two `lk.connect` spans with all four checkpoints, two +`lk.publish`, four `lk.subscribe` (`subscribed` → `first_media`), two `lk.reconnect` +(`attempt 1 quick` → `attempt 2 full` at `signal_disconnected`, then a full one), six +`lk.rtc.stats.sample` windows across both directions and both kinds, and two `lk.room.disconnected` +at `client_initiated`. + +A page cannot POST at an OTLP receiver that does not answer the preflight, which is the one +difference from the mobile harness config: `receivers.otlp.protocols.http.cors`. + +React Native runs the identical file — `telemetry-poc/` in the `client-sdk-react-native` worktree +copies `src/telemetry/index.ts` verbatim, no edits. Both pings arrive from a bare RN 0.82.1 app on +the iOS simulator at the same collector and show up in Loki next to the browser's, and the module +also runs under the bare Hermes VM with `fetch` stubbed, which is the cheap check when no simulator +is around. There it reported the size difference that settles §4: **326 bytes of protobuf against +846 bytes of JSON** for the same record. diff --git a/package.json b/package.json index 4c3fa3e1e0..794804a2dd 100644 --- a/package.json +++ b/package.json @@ -74,6 +74,7 @@ "dependencies": { "@livekit/mutex": "1.1.1", "@livekit/protocol": "1.50.4", + "@opentelemetry/otlp-transformer": "0.222.0", "events": "^3.3.0", "jose": "^6.1.0", "loglevel": "^1.9.2", @@ -94,6 +95,7 @@ "@eslint/js": "10.0.1", "@livekit/changesets-changelog-github": "^0.2.0", "@livekit/throws-transformer": "^0.1.3", + "@nbilyk/downlevel-dts": "^0.15.2", "@rollup/plugin-babel": "7.1.0", "@rollup/plugin-commonjs": "29.0.3", "@rollup/plugin-json": "6.1.0", @@ -111,7 +113,6 @@ "@typescript/native": "catalog:", "@vitest/browser": "^4.1.10", "@vitest/browser-playwright": "^4.1.10", - "@nbilyk/downlevel-dts": "^0.15.2", "eslint": "10.7.0", "eslint-config-airbnb-extended": "^2.3.2", "eslint-config-prettier": "10.1.8", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 955bc77eb4..06bb41d338 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -26,6 +26,9 @@ importers: '@livekit/protocol': specifier: 1.50.4 version: 1.50.4 + '@opentelemetry/otlp-transformer': + specifier: 0.222.0 + version: 0.222.0(@opentelemetry/api@1.9.1) '@types/dom-mediacapture-record': specifier: ^1 version: 1.0.22 @@ -215,7 +218,7 @@ importers: version: 8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0) vitest: specifier: ^4.1.10 - version: 4.1.11(@types/node@26.1.2)(@vitest/browser-playwright@4.1.10)(happy-dom@20.11.0)(jsdom@26.1.0)(vite@8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0)) + version: 4.1.11(@opentelemetry/api@1.9.1)(@types/node@26.1.2)(@vitest/browser-playwright@4.1.10)(happy-dom@20.11.0)(jsdom@26.1.0)(vite@8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0)) packages: @@ -1174,6 +1177,54 @@ packages: resolution: {integrity: sha512-oGB+UxlgWcgQkgwo8GcEGwemoTFt3FIO9ababBmaGwXIoBKZ+GTy0pP185beGg7Llih/NSHSV2XAs1lnznocSg==} engines: {node: '>= 8'} + '@opentelemetry/api-logs@0.222.0': + resolution: {integrity: sha512-9mb1If+IF6u0ZVXkHQ6ogEae5HwA6ajIVUgpSDQyRASxft6BSXHvBvPooRle3yFN/fKnCdSOnuu0OC3PLcF6+g==} + engines: {node: '>=8.0.0'} + + '@opentelemetry/api@1.9.1': + resolution: {integrity: sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q==} + engines: {node: '>=8.0.0'} + + '@opentelemetry/core@2.11.0': + resolution: {integrity: sha512-7YP44XH0tV6+Mb54x2YGf84i7yi+31MBZlE8JwvozkxyTvXbSp10X7cI7YE49ChJ3shMJoBmCJF3+1QFBJctGA==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': '>=1.0.0 <1.10.0' + + '@opentelemetry/otlp-transformer@0.222.0': + resolution: {integrity: sha512-/F3BZ89+CJQnZkMh2tCrtcdB+XT2Dxhj4FFE+WPQ//413hmFL0/RfEX6vgOIWGhiSzrkHWTK3+6SiT7K5/g/jQ==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': ^1.3.0 + + '@opentelemetry/resources@2.11.0': + resolution: {integrity: sha512-Ie7+8q8MDF4FAEQCKVMTx3ReUvxiIAgIiiW3c9JdmP8+HMcDy20puT+AHjexnExgnbvBxjQ9fjkFDWrikJ2jQA==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': '>=1.3.0 <1.10.0' + + '@opentelemetry/sdk-logs@0.222.0': + resolution: {integrity: sha512-+19YHODIjaUCArxleaJtuufFZVpz/xvvK+VllQqE+W8hHolxdoRwHfK/s667zezwh1hkx6FFF+oYzetYgqK+Bg==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': '>=1.4.0 <1.10.0' + + '@opentelemetry/sdk-metrics@2.11.0': + resolution: {integrity: sha512-7GXXcObyHyDUUSG+L+kJoquty01bzm7ivE7+SSgXXJcHuPzGviptxwARmI2c+bnnxjexGQbJnyNlN8HxBP/Y7A==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': '>=1.9.0 <1.10.0' + + '@opentelemetry/sdk-trace@2.11.0': + resolution: {integrity: sha512-fFnTqGm8/G73GQVnxYi7LXa1ZVYEUvgL6XI1LpvV0bPC7WQ/ZGgKxCSl8FnlZBKto9JHHEFTO6s6CUpvvtwFrA==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': '>=1.3.0 <1.10.0' + + '@opentelemetry/semantic-conventions@1.43.0': + resolution: {integrity: sha512-eSYWTm620tTk45EKSedaUL8MFYI8hW164hIXsgIHyxu3VobUB3fFCu5t0hQby6OoWRPsG1KkKUG2M5UadiLiVg==} + engines: {node: '>=14'} + '@oxc-project/types@0.139.0': resolution: {integrity: sha512-r9gHphtCs+1M7J0pw6Sn/hh/Wpa/iQrOOkrNAlVLF/gHq+/CJmHIWKKUUhdWjcD6CIa8idarspCsASiXCXvFUw==} @@ -5800,6 +5851,56 @@ snapshots: '@nodelib/fs.scandir': 2.1.5 fastq: 1.15.0 + '@opentelemetry/api-logs@0.222.0': + dependencies: + '@opentelemetry/api': 1.9.1 + + '@opentelemetry/api@1.9.1': {} + + '@opentelemetry/core@2.11.0(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/semantic-conventions': 1.43.0 + + '@opentelemetry/otlp-transformer@0.222.0(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/api-logs': 0.222.0 + '@opentelemetry/core': 2.11.0(@opentelemetry/api@1.9.1) + '@opentelemetry/resources': 2.11.0(@opentelemetry/api@1.9.1) + '@opentelemetry/sdk-logs': 0.222.0(@opentelemetry/api@1.9.1) + '@opentelemetry/sdk-metrics': 2.11.0(@opentelemetry/api@1.9.1) + '@opentelemetry/sdk-trace': 2.11.0(@opentelemetry/api@1.9.1) + + '@opentelemetry/resources@2.11.0(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/core': 2.11.0(@opentelemetry/api@1.9.1) + '@opentelemetry/semantic-conventions': 1.43.0 + + '@opentelemetry/sdk-logs@0.222.0(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/api-logs': 0.222.0 + '@opentelemetry/core': 2.11.0(@opentelemetry/api@1.9.1) + '@opentelemetry/resources': 2.11.0(@opentelemetry/api@1.9.1) + '@opentelemetry/semantic-conventions': 1.43.0 + + '@opentelemetry/sdk-metrics@2.11.0(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/core': 2.11.0(@opentelemetry/api@1.9.1) + '@opentelemetry/resources': 2.11.0(@opentelemetry/api@1.9.1) + + '@opentelemetry/sdk-trace@2.11.0(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/core': 2.11.0(@opentelemetry/api@1.9.1) + '@opentelemetry/resources': 2.11.0(@opentelemetry/api@1.9.1) + '@opentelemetry/semantic-conventions': 1.43.0 + + '@opentelemetry/semantic-conventions@1.43.0': {} + '@oxc-project/types@0.139.0': {} '@pkgr/core@0.3.6': {} @@ -6495,7 +6596,7 @@ snapshots: '@vitest/mocker': 4.1.10(vite@8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0)) playwright: 1.61.1 tinyrainbow: 3.1.0 - vitest: 4.1.11(@types/node@26.1.2)(@vitest/browser-playwright@4.1.10)(happy-dom@20.11.0)(jsdom@26.1.0)(vite@8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0)) + vitest: 4.1.11(@opentelemetry/api@1.9.1)(@types/node@26.1.2)(@vitest/browser-playwright@4.1.10)(happy-dom@20.11.0)(jsdom@26.1.0)(vite@8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0)) transitivePeerDependencies: - bufferutil - msw @@ -6511,7 +6612,7 @@ snapshots: pngjs: 7.0.0 sirv: 3.0.2 tinyrainbow: 3.1.0 - vitest: 4.1.11(@types/node@26.1.2)(@vitest/browser-playwright@4.1.10)(happy-dom@20.11.0)(jsdom@26.1.0)(vite@8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0)) + vitest: 4.1.11(@opentelemetry/api@1.9.1)(@types/node@26.1.2)(@vitest/browser-playwright@4.1.10)(happy-dom@20.11.0)(jsdom@26.1.0)(vite@8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0)) ws: 8.20.0 transitivePeerDependencies: - bufferutil @@ -8324,7 +8425,7 @@ snapshots: machina: 7.0.1 machina-inspect: 4.0.0(machina@7.0.1) optionalDependencies: - vitest: 4.1.11(@types/node@26.1.2)(@vitest/browser-playwright@4.1.10)(happy-dom@20.11.0)(jsdom@26.1.0)(vite@8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0)) + vitest: 4.1.11(@opentelemetry/api@1.9.1)(@types/node@26.1.2)(@vitest/browser-playwright@4.1.10)(happy-dom@20.11.0)(jsdom@26.1.0)(vite@8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0)) machina@7.0.1: {} @@ -9263,7 +9364,7 @@ snapshots: tsx: 4.23.1 yaml: 2.9.0 - vitest@4.1.11(@types/node@26.1.2)(@vitest/browser-playwright@4.1.10)(happy-dom@20.11.0)(jsdom@26.1.0)(vite@8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0)): + vitest@4.1.11(@opentelemetry/api@1.9.1)(@types/node@26.1.2)(@vitest/browser-playwright@4.1.10)(happy-dom@20.11.0)(jsdom@26.1.0)(vite@8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0)): dependencies: '@vitest/expect': 4.1.11 '@vitest/mocker': 4.1.11(vite@8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0)) @@ -9286,6 +9387,7 @@ snapshots: vite: 8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0) why-is-node-running: 2.3.0 optionalDependencies: + '@opentelemetry/api': 1.9.1 '@types/node': 26.1.2 '@vitest/browser-playwright': 4.1.10(playwright@1.61.1)(vite@8.1.5(@types/node@26.1.2)(esbuild@0.28.1)(jiti@2.4.2)(terser@5.39.2)(tsx@4.23.1)(yaml@2.9.0))(vitest@4.1.11) happy-dom: 20.11.0 diff --git a/src/index.ts b/src/index.ts index 4fa1329baa..afbee3ee1f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -201,3 +201,13 @@ export { type SerializerOutput, serializers, } from './utils/serializer'; + +export { + Telemetry, + type TelemetryOptions, + // The one seam a platform SDK implements: where batches wait between being made and accepted. + type TelemetryStorage, + type TelemetryScope, + type TelemetrySpan, + type DeviceState as TelemetryDeviceState, +} from './telemetry'; diff --git a/src/room/RTCEngine.ts b/src/room/RTCEngine.ts index fd0fc1fd61..86c3626153 100644 --- a/src/room/RTCEngine.ts +++ b/src/room/RTCEngine.ts @@ -221,6 +221,9 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit private reconnectAttempts: number = 0; + /** Why the current reconnect started — the two events below carry it to telemetry. */ + private reconnectReason?: ReconnectReason; + private reconnectStart: number = 0; private clientConfiguration?: ClientConfiguration; @@ -1307,6 +1310,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit if (this._isClosed) { return; } + this.reconnectReason = reason; // guard for attempting reconnection multiple times while one attempt is still not finished if (this.attemptingReconnect) { this.log.warn('already attempting reconnect, returning early'); @@ -1404,7 +1408,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } this.log.info(`reconnecting, attempt: ${this.reconnectAttempts}`); - this.emit(EngineEvent.Restarting); + this.emit(EngineEvent.Restarting, this.reconnectReason); if (!this.client.isDisconnected) { await this.client.sendLeave(); @@ -1477,7 +1481,7 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit } this.log.info(`resuming signal connection, attempt ${this.reconnectAttempts}`); - this.emit(EngineEvent.Resuming); + this.emit(EngineEvent.Resuming, this.reconnectReason); let res: ReconnectResponse | undefined; try { this.setupSignalClientCallbacks(); @@ -2068,9 +2072,9 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit export type EngineEventCallbacks = { connected: (joinResp: JoinResponse) => void; disconnected: (reason?: DisconnectReason) => void; - resuming: () => void; + resuming: (reason?: ReconnectReason) => void; resumed: () => void; - restarting: () => void; + restarting: (reason?: ReconnectReason) => void; restarted: () => void; signalResumed: () => void; signalRestarted: (joinResp: JoinResponse) => void; diff --git a/src/room/Room.ts b/src/room/Room.ts index d48ace5cb9..adffcd9190 100644 --- a/src/room/Room.ts +++ b/src/room/Room.ts @@ -14,6 +14,7 @@ import { ParticipantInfo, ParticipantInfo_State, ParticipantPermission, + ReconnectReason, Room as RoomModel, ServerInfo, SimulateScenario, @@ -46,6 +47,7 @@ import type { RoomConnectOptions, RoomOptions, } from '../options'; +import { SpanKind, Telemetry, type TelemetryScope, type TelemetrySpan } from '../telemetry'; import type { NonSharedUint8Array } from '../type-polyfills/non-shared-typed-arrays'; import TypedPromise from '../utils/TypedPromise'; import { getBrowser } from '../utils/browserParser'; @@ -200,6 +202,16 @@ class Room extends (EventEmitter as new () => TypedEmitter) /** future holding client initiated connection attempt */ private connectFuture?: Future; + /** One telemetry scope per connect: a trace id and the attributes every record of it carries. */ + private telemetry?: TelemetryScope; + + private connectSpan?: TelemetrySpan; + + private reconnectSpan?: TelemetrySpan; + + /** Attempts inside the *current* reconnect; the engine's own counter spans several. */ + private reconnectAttempts = 0; + private disconnectLock: Mutex; private e2eeManager: BaseE2EEManager | undefined; @@ -632,7 +644,8 @@ class Room extends (EventEmitter as new () => TypedEmitter) }) .on(EngineEvent.ActiveSpeakersUpdate, this.handleActiveSpeakersUpdate) .on(EngineEvent.DataPacketReceived, this.handleDataPacket) - .on(EngineEvent.Resuming, () => { + .on(EngineEvent.Resuming, (reason?: ReconnectReason) => { + this.startReconnectSpan('quick', reason); this.clearConnectionReconcile(); this.isResuming = true; this.log.debug('Resuming signal connection'); @@ -641,6 +654,7 @@ class Room extends (EventEmitter as new () => TypedEmitter) } }) .on(EngineEvent.Resumed, () => { + this.endReconnectSpan('ok'); this.registerConnectionReconcile(); this.isResuming = false; this.log.debug('Resumed signal connection'); @@ -858,6 +872,17 @@ class Room extends (EventEmitter as new () => TypedEmitter) } this.setAndEmitConnectionState(ConnectionState.Connecting); + if (isCloud(new URL(url))) { + // Cloud names its own destination: this host's client OTLP route, this token (SPEC). + Telemetry.setServer(url, token); + } + this.telemetry = Telemetry.scope(); + this.localParticipant.telemetry = this.telemetry; + this.connectSpan = this.telemetry.start('lk.connect', { + kind: SpanKind.client, + attributes: { 'lk.connect.attempt': 1 }, + }); + Telemetry.hold(true); if (this.regionUrlProvider?.getServerUrl().toString() !== ensureTrailingSlash(url)) { this.regionUrl = undefined; this.regionUrlProvider = undefined; @@ -1022,6 +1047,7 @@ class Room extends (EventEmitter as new () => TypedEmitter) this.localParticipant.sid = pi.sid; this.localParticipant.identity = pi.identity; + this.telemetry?.setRoom({ participantSid: pi.sid, participantIdentity: pi.identity }); this.localParticipant.setEnabledPublishCodecs(joinResponse.enabledPublishCodecs); if (this.e2eeManager) { @@ -1076,6 +1102,7 @@ class Room extends (EventEmitter as new () => TypedEmitter) } try { + this.connectSpan?.step('signal'); const joinResponse = await this.connectSignal( url, token, @@ -1084,6 +1111,7 @@ class Room extends (EventEmitter as new () => TypedEmitter) this.options, abortController, ); + this.connectSpan?.step('join_recv'); this.applyJoinResponse(joinResponse); // forward metadata changed for the local participant @@ -1121,6 +1149,7 @@ class Room extends (EventEmitter as new () => TypedEmitter) this.connOptions.peerConnectionTimeout, abortController, ); + this.connectSpan?.step('pc_connected'); } catch (e) { await this.engine.close(); this.recreateEngine(); @@ -1137,6 +1166,8 @@ class Room extends (EventEmitter as new () => TypedEmitter) window.addEventListener('freeze', this.onPageLeave); } this.setAndEmitConnectionState(ConnectionState.Connected); + this.connectSpan?.step('room_connected'); + this.endConnectSpan('ok'); this.emit(RoomEvent.Connected); BackOffStrategy.getInstance().resetFailedConnectionAttempts(url); this.registerConnectionReconcile(); @@ -1781,7 +1812,8 @@ class Room extends (EventEmitter as new () => TypedEmitter) this.emitWhenConnected(RoomEvent.LocalTrackSubscribed, trackPublication, this.localParticipant); } - private handleRestarting = () => { + private handleRestarting = (reason?: ReconnectReason) => { + this.startReconnectSpan('full', reason); this.clearConnectionReconcile(); // in case we went from resuming to full-reconnect, make sure to reflect it on the isResuming flag this.isResuming = false; @@ -1824,11 +1856,61 @@ class Room extends (EventEmitter as new () => TypedEmitter) return; } this.setAndEmitConnectionState(ConnectionState.Connected); + this.endReconnectSpan('ok'); this.emit(RoomEvent.Reconnected); this.registerConnectionReconcile(); this.emitBufferedEvents(); }; + /** `lk.connect` ends once per connect, hold and all; ending it twice is a no-op. */ + private endConnectSpan(outcome: 'ok' | 'cancelled', error?: unknown) { + if (!this.connectSpan) return; + if (error !== undefined) { + this.connectSpan.fail(error); + } else { + this.connectSpan.end(outcome); + } + this.connectSpan = undefined; + Telemetry.hold(false); + } + + private startReconnectSpan(mode: 'quick' | 'full', reason?: ReconnectReason) { + this.reconnectAttempts += 1; + const attempts = this.reconnectAttempts; + if (this.reconnectSpan) { + // A resume that turned into a restart is the same reconnect, one attempt later. + this.reconnectSpan.setAttribute('lk.reconnect.mode', mode); + this.reconnectSpan.setAttribute('lk.reconnect.attempts', attempts); + } else { + this.reconnectSpan = this.telemetry?.start('lk.reconnect', { + kind: SpanKind.client, + attributes: { + 'lk.reconnect.mode': mode, + 'lk.reconnect.attempts': attempts, + 'lk.reconnect.reason': ( + ReconnectReason[reason ?? ReconnectReason.RR_UNKNOWN] ?? 'RR_UNKNOWN' + ) + .replace(/^RR_/, '') + .toLowerCase(), + }, + }); + Telemetry.hold(true); + } + this.reconnectSpan?.step(`attempt ${attempts} ${mode}`); + } + + private endReconnectSpan(outcome: 'ok' | 'cancelled', error?: unknown) { + if (!this.reconnectSpan) return; + if (error !== undefined) { + this.reconnectSpan.fail(error); + } else { + this.reconnectSpan.end(outcome); + } + this.reconnectSpan = undefined; + this.reconnectAttempts = 0; + Telemetry.hold(false); + } + private handleDisconnect(shouldStopTracks = true, reason?: DisconnectReason) { this.clearConnectionReconcile(); this.isResuming = false; @@ -1841,6 +1923,16 @@ class Room extends (EventEmitter as new () => TypedEmitter) return; } + const reasonName = ( + DisconnectReason[reason ?? DisconnectReason.UNKNOWN_REASON] ?? 'UNKNOWN_REASON' + ).toLowerCase(); + // The call never came up: then the connect attempt is what failed, and it says why. + this.endConnectSpan('cancelled', new Error(`disconnected: ${reasonName}`)); + this.endReconnectSpan('cancelled', new Error(`disconnected: ${reasonName}`)); + this.telemetry?.disconnected(reasonName); + this.telemetry?.close(); + Telemetry.flush().catch(() => {}); + this.regionUrl = undefined; // Notify region provider about disconnect to potentially stop auto-refetch @@ -2320,6 +2412,7 @@ class Room extends (EventEmitter as new () => TypedEmitter) private handleRoomUpdate = (room: RoomModel) => { const oldRoom = this.roomInfo; this.roomInfo = room; + this.telemetry?.setRoom({ sid: room.sid, name: room.name }); if (oldRoom && oldRoom.metadata !== room.metadata) { this.emitWhenConnected(RoomEvent.RoomMetadataChanged, room.metadata); } @@ -2450,6 +2543,19 @@ class Room extends (EventEmitter as new () => TypedEmitter) track.on(TrackEvent.VideoPlaybackFailed, this.handleVideoPlaybackFailed); track.on(TrackEvent.VideoPlaybackStarted, this.handleVideoPlaybackStarted); } + if (this.telemetry) { + Telemetry.registerTrack( + publication.trackSid, + this.telemetry, + track.kind === Track.Kind.Audio ? 'audio' : 'video', + 'inbound', + ); + this.telemetry.subscribeStarted(publication.trackSid, { + 'lk.track.kind': track.kind, + 'lk.track.source': publication.source, + 'lk.participant.remote_identity': participant.identity, + }); + } this.emitWhenConnected(RoomEvent.TrackSubscribed, track, publication, participant); }, ) @@ -2460,6 +2566,8 @@ class Room extends (EventEmitter as new () => TypedEmitter) .on( ParticipantEvent.TrackUnsubscribed, (track: RemoteTrack, publication: RemoteTrackPublication) => { + this.telemetry?.subscribeEnded(publication.trackSid, 'cancelled'); + Telemetry.unregisterTrack(publication.trackSid, 'inbound'); this.emit(RoomEvent.TrackUnsubscribed, track, publication, participant); }, ) @@ -2763,6 +2871,14 @@ class Room extends (EventEmitter as new () => TypedEmitter) pub.track?.on(TrackEvent.Restarted, this.onLocalTrackRestarted); pub.track?.getProcessor()?.onPublish?.(this); + if (this.telemetry && pub.track) { + Telemetry.registerTrack( + pub.trackSid, + this.telemetry, + pub.kind === Track.Kind.Audio ? 'audio' : 'video', + 'outbound', + ); + } this.emit(RoomEvent.LocalTrackPublished, pub, this.localParticipant); if (isLocalAudioTrack(pub.track)) { @@ -2786,6 +2902,7 @@ class Room extends (EventEmitter as new () => TypedEmitter) private onLocalTrackUnpublished = (pub: LocalTrackPublication) => { pub.track?.off(TrackEvent.TrackProcessorUpdate, this.onTrackProcessorUpdate); pub.track?.off(TrackEvent.Restarted, this.onLocalTrackRestarted); + Telemetry.unregisterTrack(pub.trackSid, 'outbound'); this.emit(RoomEvent.LocalTrackUnpublished, pub, this.localParticipant); }; diff --git a/src/room/data-track/depacketizer.test.ts b/src/room/data-track/depacketizer.test.ts index a369a830b6..7ac36dc369 100644 --- a/src/room/data-track/depacketizer.test.ts +++ b/src/room/data-track/depacketizer.test.ts @@ -1,4 +1,3 @@ -/* eslint-disable @typescript-eslint/no-unused-vars */ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import DataTrackDepacketizer from './depacketizer'; import { DataTrackHandle } from './handle'; diff --git a/src/room/data-track/handle.test.ts b/src/room/data-track/handle.test.ts index c486c4a8c0..34e46408ee 100644 --- a/src/room/data-track/handle.test.ts +++ b/src/room/data-track/handle.test.ts @@ -1,4 +1,3 @@ -/* eslint-disable @typescript-eslint/no-unused-vars */ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { DataTrackHandle } from './handle'; diff --git a/src/room/data-track/incoming/IncomingDataTrackManager.test.ts b/src/room/data-track/incoming/IncomingDataTrackManager.test.ts index 4c866179c5..e9935c2bc5 100644 --- a/src/room/data-track/incoming/IncomingDataTrackManager.test.ts +++ b/src/room/data-track/incoming/IncomingDataTrackManager.test.ts @@ -1,4 +1,3 @@ -/* eslint-disable @typescript-eslint/no-unused-vars */ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { subscribeToEvents } from '../../../utils/subscribeToEvents'; import { type DataTrackFrame } from '../frame'; diff --git a/src/room/data-track/outgoing/OutgoingDataTrackManager.test.ts b/src/room/data-track/outgoing/OutgoingDataTrackManager.test.ts index aac62db8c0..35bf9e24d5 100644 --- a/src/room/data-track/outgoing/OutgoingDataTrackManager.test.ts +++ b/src/room/data-track/outgoing/OutgoingDataTrackManager.test.ts @@ -1,4 +1,3 @@ -/* eslint-disable @typescript-eslint/no-unused-vars */ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { type DecryptDataResponseMessage, diff --git a/src/room/data-track/packet/index.test.ts b/src/room/data-track/packet/index.test.ts index f918acb721..daddfa8316 100644 --- a/src/room/data-track/packet/index.test.ts +++ b/src/room/data-track/packet/index.test.ts @@ -1,4 +1,3 @@ -/* eslint-disable @typescript-eslint/no-unused-vars */ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { DataTrackPacket, DataTrackPacketHeader, FrameMarker } from '.'; import { DataTrackHandle } from '../handle'; diff --git a/src/room/data-track/packetizer.test.ts b/src/room/data-track/packetizer.test.ts index 452f765bc2..7e6e333a32 100644 --- a/src/room/data-track/packetizer.test.ts +++ b/src/room/data-track/packetizer.test.ts @@ -1,4 +1,3 @@ -/* eslint-disable @typescript-eslint/no-unused-vars */ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { DataTrackFrameInternal } from './frame'; import { DataTrackHandle } from './handle'; diff --git a/src/room/data-track/utils.test.ts b/src/room/data-track/utils.test.ts index dedf0e41bd..ce951a4cc8 100644 --- a/src/room/data-track/utils.test.ts +++ b/src/room/data-track/utils.test.ts @@ -1,4 +1,3 @@ -/* eslint-disable @typescript-eslint/no-unused-vars */ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { U16_MAX_SIZE, WrapAroundUnsignedInt } from './utils'; diff --git a/src/room/participant/LocalParticipant.ts b/src/room/participant/LocalParticipant.ts index 645482c224..7d5d6b4922 100644 --- a/src/room/participant/LocalParticipant.ts +++ b/src/room/participant/LocalParticipant.ts @@ -30,6 +30,7 @@ import { isFrameMetadataSupported, } from '../../frameMetadata/utils'; import type { InternalRoomOptions } from '../../options'; +import type { TelemetryScope } from '../../telemetry'; import type { NonSharedUint8Array } from '../../type-polyfills/non-shared-typed-arrays'; import TypedPromise from '../../utils/TypedPromise'; import { PCTransportState } from '../PCTransportManager'; @@ -138,6 +139,9 @@ export default class LocalParticipant extends Participant { /** @internal */ activeDeviceMap: Map; + /** @internal — the Room's telemetry scope, set at connect; publishing is a span on it. */ + telemetry?: TelemetryScope; + private pendingPublishing = new Set(); private pendingPublishPromises = new Map>(); @@ -778,7 +782,22 @@ export default class LocalParticipant extends Participant { * @param options */ async publishTrack(track: LocalTrack | MediaStreamTrack, options?: TrackPublishOptions) { - return this.publishOrRepublishTrack(track, options); + const span = this.telemetry?.start('lk.publish', { + attributes: { + 'lk.track.kind': track.kind, + 'lk.track.source': 'source' in track ? track.source : undefined, + }, + }); + try { + const publication = await this.publishOrRepublishTrack(track, options); + span?.setAttribute('lk.track.sid', publication.trackSid); + span?.setAttribute('lk.track.source', publication.source); + span?.end('ok'); + return publication; + } catch (error) { + span?.fail(error); + throw error; + } } /** diff --git a/src/room/token-source/TokenSource.test.ts b/src/room/token-source/TokenSource.test.ts index 709e76ca46..a44c69ed87 100644 --- a/src/room/token-source/TokenSource.test.ts +++ b/src/room/token-source/TokenSource.test.ts @@ -1,4 +1,3 @@ -/* eslint-disable @typescript-eslint/no-unused-vars */ import { describe, expect, it, vi } from 'vitest'; import { sleep } from '../utils'; import { TokenSource } from './TokenSource'; diff --git a/src/room/token-source/types.ts b/src/room/token-source/types.ts index 1750f00615..3e51638903 100644 --- a/src/room/token-source/types.ts +++ b/src/room/token-source/types.ts @@ -3,7 +3,7 @@ import type { JWTPayload } from 'jose'; import type { ValueToSnakeCase } from '../../utils/camelToSnakeCase'; // The below imports are being linked in tsdoc comments, so they have to be imported even if they // aren't being used. -// eslint-disable-next-line @typescript-eslint/no-unused-vars + import type { TokenSourceCustom, TokenSourceEndpoint, TokenSourceLiteral } from './TokenSource'; export type TokenSourceRequestObject = Required< diff --git a/src/room/track/LocalAudioTrack.ts b/src/room/track/LocalAudioTrack.ts index 05c6db4dc6..ff18cfd42b 100644 --- a/src/room/track/LocalAudioTrack.ts +++ b/src/room/track/LocalAudioTrack.ts @@ -1,4 +1,5 @@ import { AudioTrackFeature } from '@livekit/protocol'; +import { Telemetry } from '../../telemetry'; import { TrackEvent } from '../events'; import { computeBitrate, monitorFrequency } from '../stats'; import type { AudioSenderStats } from '../stats'; @@ -158,6 +159,9 @@ export default class LocalAudioTrack extends LocalTrack { if (stats && this.prevStats) { this._currentBitrate = computeBitrate(stats, this.prevStats); } + if (stats) { + Telemetry.senderStats(this.sid, [stats]); + } this.prevStats = stats; }; diff --git a/src/room/track/LocalVideoTrack.ts b/src/room/track/LocalVideoTrack.ts index 31683db86a..684c601236 100644 --- a/src/room/track/LocalVideoTrack.ts +++ b/src/room/track/LocalVideoTrack.ts @@ -7,6 +7,7 @@ import { } from '@livekit/protocol'; import type { SignalClient } from '../../api/SignalClient'; import type { StructuredLogger } from '../../logger'; +import { Telemetry } from '../../telemetry'; import { TrackEvent } from '../events'; import { ScalabilityMode, @@ -644,6 +645,7 @@ export default class LocalVideoTrack extends LocalTrack { }); this._currentBitrate = totalBitrate; } + Telemetry.senderStats(this.sid, stats); this.prevStats = statsMap; }; diff --git a/src/room/track/RemoteAudioTrack.ts b/src/room/track/RemoteAudioTrack.ts index 7d48caa7a2..665cfa0bac 100644 --- a/src/room/track/RemoteAudioTrack.ts +++ b/src/room/track/RemoteAudioTrack.ts @@ -1,3 +1,4 @@ +import { Telemetry } from '../../telemetry'; import { TrackEvent } from '../events'; import type { AudioReceiverStats } from '../stats'; import { computeBitrate } from '../stats'; @@ -230,6 +231,7 @@ export default class RemoteAudioTrack extends RemoteTrack { if (stats && this.prevStats && this.receiver) { this._currentBitrate = computeBitrate(stats, this.prevStats); } + Telemetry.receiverStats(this.sid, stats); this.prevStats = stats; }; @@ -249,6 +251,8 @@ export default class RemoteAudioTrack extends RemoteTrack { timestamp: v.timestamp, jitter: v.jitter, bytesReceived: v.bytesReceived, + packetsReceived: v.packetsReceived, + packetsLost: v.packetsLost, concealedSamples: v.concealedSamples, concealmentEvents: v.concealmentEvents, silentConcealedSamples: v.silentConcealedSamples, diff --git a/src/room/track/RemoteVideoTrack.ts b/src/room/track/RemoteVideoTrack.ts index 7c511afd5b..7143adc614 100644 --- a/src/room/track/RemoteVideoTrack.ts +++ b/src/room/track/RemoteVideoTrack.ts @@ -1,4 +1,5 @@ import type { FrameMetadata } from '../../frameMetadata/types'; +import { Telemetry } from '../../telemetry'; import { debounce } from '../debounce'; import { TrackEvent } from '../events'; import type { VideoReceiverStats } from '../stats'; @@ -186,6 +187,7 @@ export default class RemoteVideoTrack extends RemoteTrack { if (stats && this.prevStats && this.receiver) { this._currentBitrate = computeBitrate(stats, this.prevStats); } + Telemetry.receiverStats(this.sid, stats); this.prevStats = stats; }; diff --git a/src/telemetry/device.ts b/src/telemetry/device.ts new file mode 100644 index 0000000000..9554ef05ab --- /dev/null +++ b/src/telemetry/device.ts @@ -0,0 +1,113 @@ +/** + * Device state, and the cadence factor it drives — but only what a page can answer about itself. + * + * The pipeline never measures anything: every value here is the platform saying so. A browser can + * say whether it is visible and (on Chromium) what it is connected through. It cannot say anything + * about heat, power or memory pressure, and this package does not pretend otherwise: a platform + * that knows those reports them to its own backend, not through here. + */ +import type { Attributes } from './otlp'; + +export type AppStateName = 'foreground' | 'background'; +export type NetworkType = + 'wifi' | 'cell' | 'wired' | 'vpn' | 'bluetooth' | 'other' | 'unavailable' | 'unknown'; + +export interface DeviceState { + appState?: AppStateName; + networkType?: NetworkType; + /** Cellular or hotspot. */ + networkExpensive?: boolean; + /** Low Data Mode, Data Saver, `navigator.connection.saveData`. */ + networkConstrained?: boolean; +} + +/** SPEC's cadence table, for the rows a page can fill in. Factors multiply, capped at 4×. */ +const MAX_FACTOR = 4; + +export function cadenceFactor(state: DeviceState): number { + let factor = 1; + if (state.appState === 'background') factor *= 2; + if (state.networkConstrained) factor *= 2; + return Math.min(factor, MAX_FACTOR); +} + +/** `NetworkInformation.type` → SPEC's enum. */ +export function networkType(type: string | undefined): NetworkType { + switch (type) { + case 'wifi': + case 'bluetooth': + return type; + case 'cellular': + return 'cell'; + case 'ethernet': + return 'wired'; + case 'none': + return 'unavailable'; + case 'wimax': + case 'mixed': + case 'other': + return 'other'; + default: + return 'unknown'; + } +} + +interface Change { + event: string; + attributes: Attributes; +} + +/** What changed since the last update, as the records SPEC names — one per group, on change only. */ +export function changes(previous: DeviceState, next: DeviceState): Change[] { + const out: Change[] = []; + if (next.appState !== undefined && next.appState !== previous.appState) { + out.push({ + event: 'lk.device.app_state.changed', + attributes: { 'lk.device.app_state': next.appState }, + }); + } + const networkChanged = + (next.networkType !== undefined && next.networkType !== previous.networkType) || + (next.networkExpensive !== undefined && next.networkExpensive !== previous.networkExpensive) || + (next.networkConstrained !== undefined && + next.networkConstrained !== previous.networkConstrained); + if (networkChanged) { + out.push({ + event: 'lk.device.network.changed', + attributes: { + 'network.connection.type': next.networkType ?? previous.networkType, + 'lk.device.network.expensive': next.networkExpensive ?? previous.networkExpensive, + 'lk.device.network.constrained': next.networkConstrained ?? previous.networkConstrained, + }, + }); + } + return out; +} + +/** Everything a page can answer: the tab's visibility everywhere, the connection on Chromium. */ +export function observeBrowser(report: (state: DeviceState) => void): void { + if (typeof document !== 'undefined') { + const visibility = () => + report({ appState: document.visibilityState === 'hidden' ? 'background' : 'foreground' }); + document.addEventListener('visibilitychange', visibility); + visibility(); + } + const connection = (globalThis.navigator as { connection?: NetworkInformationLike } | undefined) + ?.connection; + if (connection) { + const network = () => + report({ + networkType: networkType(connection.type), + networkExpensive: connection.type === 'cellular', + networkConstrained: connection.saveData === true, + }); + connection.addEventListener?.('change', network); + network(); + } +} + +interface NetworkInformationLike { + type?: string; + saveData?: boolean; + addEventListener?: (event: 'change', listener: () => void) => void; +} diff --git a/src/telemetry/index.ts b/src/telemetry/index.ts new file mode 100644 index 0000000000..a0bd422974 --- /dev/null +++ b/src/telemetry/index.ts @@ -0,0 +1,173 @@ +/** + * Client telemetry. One pipeline per page (or per app), one scope per Room connection; see + * TELEMETRY.md for the design and `livekit-telemetry/SPEC.md` in rust-sdks for what the records + * mean — the Swift, Kotlin and Dart SDKs emit the same ones from the Rust core. + * + * Nothing is collected until a destination exists: `Telemetry.configure({ endpoint })` for your own + * collector, or the first connect to LiveKit Cloud, which derives the route and the token itself. + * + * Where batches wait between being made and being accepted is the one replaceable part + * (`TelemetryOptions.storage`): a browser keeps them in memory, a platform with a filesystem + * keeps them on disk. Everything else is this package's. + */ +import { type DeviceState, observeBrowser } from './device'; +import { type Attributes, Severity, hrTime, randomHex } from './otlp'; +import { Pipeline, type TelemetryOptions } from './pipeline'; +import type { StatsSample, TelemetryScope, TrackDirection } from './scope'; +import { receiverSample, senderSample } from './webrtc'; + +export type { TelemetryOptions } from './pipeline'; +export type { DeviceState, AppStateName, NetworkType } from './device'; +export type { + Outcome, + RoomIdentity, + SeverityName, + StatsSample, + TrackDirection, + TraceContext, +} from './scope'; +export type { TelemetryScope, TelemetrySpan } from './scope'; +export { SpanKind } from './otlp'; +export type { Attributes } from './otlp'; +export type { TelemetryStorage } from './storage'; + +const pipeline = new Pipeline(); + +/** Which scope a track's stats belong to — the monitors know a sid, not a Room. */ +interface TrackRegistration { + scope: TelemetryScope; + kind: 'audio' | 'video'; + direction: TrackDirection; +} + +const tracks = new Map(); + +let lifecycleAttached = false; + +function attachLifecycle() { + if (lifecycleAttached || typeof document === 'undefined') return; + lifecycleAttached = true; + // The page's last chance: `pagehide` and a hidden tab, never `unload` — by then a request has + // no chance of leaving. `fetch(keepalive)` makes it best effort, not durable (TELEMETRY.md §3). + const flush = () => { + pipeline.flush().catch(() => {}); + }; + document.addEventListener('visibilitychange', () => { + if (document.visibilityState === 'hidden') flush(); + }); + window.addEventListener('pagehide', flush); + observeBrowser((state) => Telemetry.deviceState(state)); +} + +export const Telemetry = { + /** Point the default pipeline at a collector and start it. Safe to call more than once. */ + configure(options: TelemetryOptions) { + pipeline.configure(options); + attachLifecycle(); + }, + + /** LiveKit Cloud: the server URL and the connect token are the destination (SPEC). */ + setServer(serverUrl: string, token: string) { + pipeline.setServer(serverUrl, token); + attachLifecycle(); + }, + + get enabled(): boolean { + return pipeline.enabled; + }, + + scope(): TelemetryScope { + return pipeline.scope(); + }, + + /** Uploads stop, collection does not — spans that own the uplink raise a hold (SPEC). */ + hold(up: boolean) { + pipeline.hold(up); + }, + + /** What the platform can say about the device it runs on; see `DeviceState` for the limits. */ + deviceState(state: DeviceState) { + pipeline.deviceState(state); + }, + + /** + * A record belonging to the pipeline rather than to any call. A platform SDK that observes + * something this package has no vocabulary for — a phone's thermal state — names it here. + */ + emit(event: string, attributes?: Attributes, severity?: 'info' | 'warn' | 'error') { + pipeline.emit(event, attributes, severity); + }, + + /** + * Stretch the flush interval and the stats window by this much, 1–4. The platform that can see + * pressure this package cannot reports the number, not the reason. + */ + setCadenceFactor(factor: number) { + pipeline.setCadenceFactor(factor); + }, + + registerTrack( + sid: string, + scope: TelemetryScope, + kind: 'audio' | 'video', + direction: TrackDirection, + ) { + // Nothing to route when nobody is listening: an SDK without a collector keeps no map. + if (!pipeline.enabled) return; + tracks.set(`${sid}:${direction}`, { scope, kind, direction }); + }, + + unregisterTrack(sid: string, direction: TrackDirection) { + tracks.delete(`${sid}:${direction}`); + }, + + /** What a local track's monitor just read; simulcast layers arrive together. */ + senderStats(sid: string | undefined, stats: Parameters[0]) { + if (sid) Telemetry.trackStats(sid, 'outbound', senderSample(stats)); + }, + + /** What a remote track's monitor just read. */ + receiverStats(sid: string | undefined, stats: Parameters[0] | undefined) { + if (sid && stats) Telemetry.trackStats(sid, 'inbound', receiverSample(stats)); + }, + + /** Called from the SDK's existing per-track monitors: no extra `getStats()` anywhere. */ + trackStats(sid: string, direction: TrackDirection, sample: StatsSample) { + if (!pipeline.enabled) return; + const registration = tracks.get(`${sid}:${direction}`); + if (!registration) return; + registration.scope.recordStats(sid, registration.kind, direction, sample); + }, + + flush(): Promise { + return pipeline.flush(); + }, + + diagnostics(): string { + return pipeline.diagnostics(); + }, + + /** A pipeline smoke test: one record, one request, whatever the collector answers. */ + ping(seq = 1) { + const now = hrTime(); + pipeline.record({ + hrTime: now, + hrTimeObserved: now, + eventName: 'lk.ping', + severityNumber: Severity.info, + severityText: 'INFO', + body: 'lk.ping', + attributes: { 'lk.ping.seq': seq, 'otel.event.name': 'lk.ping' }, + droppedAttributesCount: 0, + resource: { attributes: {} }, + instrumentationScope: { name: 'livekit-telemetry', version: '0.1.0' }, + spanContext: { traceId: randomHex(16), spanId: randomHex(8), traceFlags: 1 }, + }); + return pipeline.flush(true); + }, + + async shutdown() { + tracks.clear(); + await pipeline.shutdown(); + }, +}; diff --git a/src/telemetry/otelcol-web.yaml b/src/telemetry/otelcol-web.yaml new file mode 100644 index 0000000000..8cecc9cb30 --- /dev/null +++ b/src/telemetry/otelcol-web.yaml @@ -0,0 +1,30 @@ +# Browser variant of the mobile harness collector (Tests/.../otelcol-lgtm.yaml in client-sdk-swift): +# same file sink + Grafana LGTM fan-out, on its own port, with CORS — a page cannot POST to an +# OTLP receiver that does not answer the preflight. +# Run: otelcol-contrib --config src/telemetry/otelcol-web.yaml +receivers: + otlp: + protocols: + http: + endpoint: 127.0.0.1:4320 + cors: + allowed_origins: ['http://localhost:*', 'http://127.0.0.1:*'] + allowed_headers: ['*'] +exporters: + file: + path: /tmp/livekit-telemetry-web.jsonl + otlphttp/lgtm: + endpoint: http://127.0.0.1:4318 + retry_on_failure: + enabled: false +service: + telemetry: + metrics: + level: none + pipelines: + logs: + receivers: [otlp] + exporters: [file, otlphttp/lgtm] + traces: + receivers: [otlp] + exporters: [file, otlphttp/lgtm] diff --git a/src/telemetry/otlp.ts b/src/telemetry/otlp.ts new file mode 100644 index 0000000000..a596120c6b --- /dev/null +++ b/src/telemetry/otlp.ts @@ -0,0 +1,120 @@ +/** + * OTLP records and their wire form. The only piece of OpenTelemetry the SDK ships is + * `@opentelemetry/otlp-transformer`, whose serializers are structural: they read the fields below + * off plain objects and nothing else. See TELEMETRY.md for why the rest of the SDK is not here. + */ +import { + JsonLogsSerializer, + JsonTraceSerializer, + ProtobufLogsSerializer, + ProtobufTraceSerializer, +} from '@opentelemetry/otlp-transformer'; + +export type Encoding = 'protobuf' | 'json'; + +export type AttributeValue = string | number | boolean; + +export type Attributes = { [key: string]: AttributeValue | undefined }; + +export interface Resource { + attributes: { [key: string]: unknown }; +} + +export interface SpanContext { + traceId: string; + spanId: string; + traceFlags: number; +} + +/** OTel severity numbers; only these three are emitted (SPEC: nothing below `warn` leaves). */ +export const Severity = { info: 9, warn: 13, error: 17 } as const; + +export const INSTRUMENTATION_SCOPE = { name: 'livekit-telemetry', version: '0.1.0' }; + +export interface LogRecord { + hrTime: [number, number]; + hrTimeObserved: [number, number]; + eventName?: string; + severityNumber: number; + severityText: string; + body?: string; + attributes: Attributes; + droppedAttributesCount: number; + resource: Resource; + instrumentationScope: typeof INSTRUMENTATION_SCOPE; + spanContext?: SpanContext; +} + +export interface SpanEvent { + name: string; + time: [number, number]; + attributes: Attributes; + droppedAttributesCount: number; +} + +export interface SpanRecord { + name: string; + /** OTel SpanKind: 2 = client, 1 = internal. */ + kind: number; + spanContext: () => SpanContext; + parentSpanContext?: SpanContext; + startTime: [number, number]; + endTime: [number, number]; + duration: [number, number]; + status: { code: number; message?: string }; + attributes: Attributes; + links: never[]; + events: SpanEvent[]; + ended: boolean; + resource: Resource; + instrumentationScope: typeof INSTRUMENTATION_SCOPE; + droppedAttributesCount: number; + droppedEventsCount: number; + droppedLinksCount: number; +} + +export const SpanKind = { internal: 1, client: 3 } as const; +/** OTel status codes: cancellation is `Unset` like success — only a failure is `Error`. */ +export const SpanStatus = { unset: 0, error: 2 } as const; + +// ponytail: unique, not unguessable — ids need no entropy guarantees, and this keeps React Native +// from needing `react-native-get-random-values`. +export function randomHex(bytes: number): string { + const buffer = new Uint8Array(bytes); + if (globalThis.crypto?.getRandomValues) { + globalThis.crypto.getRandomValues(buffer); + } else { + for (let i = 0; i < bytes; i += 1) { + buffer[i] = Math.floor(Math.random() * 256); + } + } + return Array.from(buffer, (byte) => byte.toString(16).padStart(2, '0')).join(''); +} + +export function hrTime(milliseconds: number = Date.now()): [number, number] { + return [Math.trunc(milliseconds / 1000), Math.round((milliseconds % 1000) * 1e6)]; +} + +export function hrDuration(start: [number, number], end: [number, number]): [number, number] { + let seconds = end[0] - start[0]; + let nanos = end[1] - start[1]; + if (nanos < 0) { + seconds -= 1; + nanos += 1e9; + } + return [seconds, nanos]; +} + +export function serializeLogs(records: LogRecord[], encoding: Encoding): Uint8Array { + const serializer = encoding === 'json' ? JsonLogsSerializer : ProtobufLogsSerializer; + return serializer.serializeRequest(records as never)!; +} + +export function serializeSpans(records: SpanRecord[], encoding: Encoding): Uint8Array { + const serializer = encoding === 'json' ? JsonTraceSerializer : ProtobufTraceSerializer; + return serializer.serializeRequest(records as never)!; +} + +export function contentType(encoding: Encoding): string { + return encoding === 'json' ? 'application/json' : 'application/x-protobuf'; +} diff --git a/src/telemetry/pipeline.ts b/src/telemetry/pipeline.ts new file mode 100644 index 0000000000..987fa55b2d --- /dev/null +++ b/src/telemetry/pipeline.ts @@ -0,0 +1,545 @@ +/** + * The upload policy: one queue, one request in flight, holds while the uplink belongs to media. + * Everything a browser cannot do is absent by design — no disk cache, no replay across launches + * (TELEMETRY.md §3), so the queue is the only bound and every eviction is counted. + */ +import { version } from '../version'; +import { type DeviceState, cadenceFactor, changes } from './device'; +import { + type AttributeValue, + type Attributes, + type Encoding, + INSTRUMENTATION_SCOPE, + type LogRecord, + type Resource, + Severity, + type SpanRecord, + contentType, + hrTime, + serializeLogs, + serializeSpans, +} from './otlp'; +import { TelemetryScope } from './scope'; +import { + MemoryStorage, + type TelemetryStorage, + batchEncoding, + batchId, + batchKind, + batchRecords, +} from './storage'; + +export interface TelemetryOptions { + /** OTLP logs route. Cloud derives it from the server URL instead; see `setServer`. */ + endpoint?: string; + headers?: Record; + encoding?: Encoding; + /** Seconds between uploads (default 15). */ + flushInterval?: number; + /** Seconds per `lk.rtc.stats.sample` window (default 15). */ + statsWindow?: number; + maxQueueSize?: number; + resource?: Record; + /** + * Where batches wait between being made and being accepted. A platform with a filesystem + * supplies one that survives the process; the default keeps them in memory. + */ + storage?: TelemetryStorage; +} + +interface Destination { + logs: string; + traces: string; + headers: Record; +} + +const FLUSH_INTERVAL = 15; +const STATS_WINDOW = 15; +const MAX_QUEUE = 2048; +const MAX_BATCH = 512; +/** What the cache may hold: 4 MiB across at most 512 batches, oldest evicted first. */ +const MAX_CACHE_BYTES = 4 * 1024 * 1024; +const MAX_CACHE_BATCHES = 512; +/** A backlog replays beside a live call at this many batches per tick, never faster (SPEC). */ +const MAX_BATCHES_PER_UPLOAD = 4; +/** Browsers cap every in-flight keepalive body at 64 KiB together; stay under it. */ +const KEEPALIVE_LIMIT = 60 * 1024; +/** A hold never outlasts this, whatever the signal that raised it claims (SPEC). */ +const HOLD_CAP_MS = 60_000; +/** LiveKit Cloud's quota answer names no delay, so a bare 429 means "this minute". */ +const THROTTLE_DEFAULT_MS = 60_000; +const FLOOD_LIMIT = 300; +const FLOOD_WINDOW_MS = 10 * 60_000; + +export interface Counters { + sent: number; + bytes: number; + failed: number; + holdsCapped: number; + droppedCacheFull: number; + droppedCacheError: number; + droppedQueueFull: number; + droppedRejected: number; + droppedThrottled: number; + droppedRateLimited: number; +} + +function emptyCounters(): Counters { + return { + sent: 0, + bytes: 0, + failed: 0, + holdsCapped: 0, + droppedCacheFull: 0, + droppedCacheError: 0, + droppedQueueFull: 0, + droppedRejected: 0, + droppedThrottled: 0, + droppedRateLimited: 0, + }; +} + +/** `wss://host/…` → the client OTLP routes on the same host. */ +export function cloudEndpoints(serverUrl: string): { logs: string; traces: string } { + const url = new URL(serverUrl); + const host = `${url.protocol === 'ws:' || url.protocol === 'http:' ? 'http' : 'https'}://${url.host}`; + return { + logs: `${host}/observability/client/logs/otlp/v0`, + traces: `${host}/observability/client/traces/otlp/v0`, + }; +} + +/** A collector's logs route implies its traces route, both for Cloud and for plain OTLP. */ +export function tracesEndpointFor(logs: string): string { + if (logs.includes('/logs/otlp/')) return logs.replace('/logs/otlp/', '/traces/otlp/'); + if (logs.endsWith('/v1/logs')) return `${logs.slice(0, -'/v1/logs'.length)}/v1/traces`; + return logs; +} + +export class Pipeline { + private logs: LogRecord[] = []; + + private spans: SpanRecord[] = []; + + private destination?: Destination; + + private encoding: Encoding = 'protobuf'; + + private timer?: ReturnType; + + private inFlight = false; + + private throttledUntil = 0; + + private holds = 0; + + private holdSince = 0; + + private counters = emptyCounters(); + + private reportDue = false; + + private floodCount = 0; + + private floodWindowStart = 0; + + private disabled = false; + + /** Set by whoever says telemetry is wanted, which is not the same as knowing where to send it. */ + private collecting = false; + + /** Device state belongs to no call: it is filed under the pipeline's own scope (SPEC). */ + private processScope?: TelemetryScope; + + private device: DeviceState = {}; + + private storage: TelemetryStorage = new MemoryStorage(MAX_CACHE_BYTES, MAX_CACHE_BATCHES); + + private sequence = 0; + + resource: Resource = { attributes: {} }; + + private baseFlushInterval = FLUSH_INTERVAL; + + private baseStatsWindow = STATS_WINDOW; + + /** SPEC's cadence policy: pressure stretches both periods, never past 4×. Two sources + * multiply — what this package observes, and what a platform reports through + * `setCadenceFactor` for the pressure it can see and this one cannot. */ + private cadence = 1; + + private deviceFactor = 1; + + private platformFactor = 1; + + maxQueueSize = MAX_QUEUE; + + get flushInterval(): number { + return this.baseFlushInterval * this.cadence; + } + + get statsWindow(): number { + return this.baseStatsWindow * this.cadence; + } + + scope(): TelemetryScope { + return new TelemetryScope(this); + } + + /** + * What the platform now says about the device — only the rows a page can fill in; see + * `DeviceState`. Each group becomes an `lk.device.*` record the first time it is seen and on + * every change after, and the whole state sets the cadence factor. + */ + deviceState(state: DeviceState) { + const next = { ...this.device, ...state }; + const records = changes(this.device, next); + this.device = next; + if (this.enabled) { + this.processScope ??= new TelemetryScope(this); + for (const record of records) { + this.processScope.emit(record.event, record.attributes); + } + } + this.deviceFactor = cadenceFactor(next); + this.applyCadence(); + } + + setCadenceFactor(factor: number) { + this.platformFactor = factor; + this.applyCadence(); + } + + private applyCadence() { + const factor = Math.min(4, this.deviceFactor * this.platformFactor); + if (factor === this.cadence) return; + this.cadence = factor; + // Restarting the timer is what makes a *shorter* period apply at once — pressure relieved + // should not mean waiting out a stretched interval. + if (this.timer) { + this.stop(); + this.start(); + } + } + + /** + * Collection starts as soon as anything asked for telemetry — `configure` in an app that sets its + * own resource or collector, `setServer` at the first Cloud connect. Records made before the + * destination is known wait in the queue rather than being thrown away (SPEC: the pipeline may + * start without a destination). An SDK nobody asked stays inert and costs nothing. + */ + get enabled(): boolean { + return !this.disabled && this.collecting; + } + + configure(options: TelemetryOptions) { + this.disabled = false; + this.collecting = true; + this.encoding = options.encoding ?? this.encoding; + this.baseFlushInterval = options.flushInterval ?? this.baseFlushInterval; + this.baseStatsWindow = options.statsWindow ?? this.baseStatsWindow; + this.maxQueueSize = options.maxQueueSize ?? this.maxQueueSize; + this.resource = { + attributes: { + // The platform SDK owns `service.*` and `os.*`; these are what this package can say + // about itself. React Native overrides the name and adds the device (SPEC). + 'service.name': 'livekit-client-js', + 'service.version': version, + 'telemetry.sdk.name': INSTRUMENTATION_SCOPE.name, + 'telemetry.sdk.language': 'webjs', + 'telemetry.sdk.version': INSTRUMENTATION_SCOPE.version, + ...this.resource.attributes, + ...options.resource, + }, + }; + if (options.storage) { + this.storage = options.storage; + } + if (options.endpoint) { + this.destination = { + logs: options.endpoint, + traces: tracesEndpointFor(options.endpoint), + headers: options.headers ?? {}, + }; + } + this.start(); + } + + /** The first connect names the destination: the server's host, the connect token (SPEC). */ + setServer(serverUrl: string, token: string) { + this.collecting = true; + if (this.destination) return; // an explicit endpoint wins + const { logs, traces } = cloudEndpoints(serverUrl); + this.destination = { logs, traces, headers: { Authorization: `Bearer ${token}` } }; + this.start(); + } + + private start() { + if (this.timer || !this.enabled) return; + this.timer = setInterval(() => { + this.flush().catch(() => {}); + }, this.flushInterval * 1000); + } + + /** While a hold is up nothing is sent; records keep arriving. Capped, so a lying signal cannot + * silence the pipeline for good. */ + hold(up: boolean) { + if (up) { + if (this.holds === 0) this.holdSince = Date.now(); + this.holds += 1; + } else if (this.holds > 0) { + this.holds -= 1; + } + } + + private held(): boolean { + if (this.holds === 0) return false; + if (Date.now() - this.holdSince >= HOLD_CAP_MS) { + this.counters.holdsCapped += 1; + this.reportDue = true; + this.holdSince = Date.now(); + return false; // one batch goes out, then the hold starts over + } + return true; + } + + /** Discrete events are rationed; stats windows and the self-report are not (SPEC flood guard). */ + private floodOk(): boolean { + const now = Date.now(); + if (now - this.floodWindowStart > FLOOD_WINDOW_MS) { + this.floodWindowStart = now; + this.floodCount = 0; + } + this.floodCount += 1; + if (this.floodCount > FLOOD_LIMIT) { + this.counters.droppedRateLimited += 1; + this.reportDue = true; + return false; + } + return true; + } + + /** A record that belongs to the pipeline rather than to a call — what a platform reports. */ + emit(event: string, attributes: Attributes = {}, severity: 'info' | 'warn' | 'error' = 'info') { + if (!this.enabled) return; + this.processScope ??= new TelemetryScope(this); + this.processScope.emit(event, attributes, severity); + } + + record(record: LogRecord, options: { exemptFromFlood?: boolean } = {}) { + if (!this.enabled) return; + if (!options.exemptFromFlood && !this.floodOk()) return; + record.resource = this.resource; + this.push(this.logs, record); + } + + endSpan(span: SpanRecord) { + if (!this.enabled) return; + span.resource = this.resource; + this.push(this.spans, span); + } + + private push(queue: T[], record: T) { + queue.push(record); + const total = this.logs.length + this.spans.length; + if (total > this.maxQueueSize) { + const overflow = total - this.maxQueueSize; + const fromLogs = Math.min(overflow, this.logs.length); + this.logs.splice(0, fromLogs); // oldest first, at every level + this.spans.splice(0, overflow - fromLogs); + this.counters.droppedQueueFull += overflow; + this.reportDue = true; + } + } + + async flush(force = false): Promise { + if (!this.enabled || this.inFlight) return; + if (!force && (this.held() || Date.now() < this.throttledUntil)) return; + if (this.reportDue) this.appendReport(); + // Write-ahead: what is queued becomes a stored batch whether or not anything can be sent yet. + this.persist(); + const destination = this.destination; + if (!destination) return; + + this.inFlight = true; + try { + // A backlog from a previous launch replays beside the call at 4 batches a tick; a shutdown + // or a page leaving drains without the budget. + const budget = force ? Number.POSITIVE_INFINITY : MAX_BATCHES_PER_UPLOAD; + for (const id of this.storage.pending().slice(0, budget)) { + const body = this.storage.read(id); + if (body === undefined) { + this.storage.remove(id); + continue; + } + const url = batchKind(id) === 'logs' ? destination.logs : destination.traces; + const verdict = await this.send(url, body, batchRecords(id), batchEncoding(id)); + // Throttled or offline: this batch keeps its place and so does everything behind it. + if (verdict === 'keep') break; + this.storage.remove(id); + } + } finally { + this.inFlight = false; + } + } + + private persist() { + if (this.logs.length > 0) { + const batch = this.logs.splice(0, MAX_BATCH); + this.store(batchId('logs', (this.sequence += 1), batch.length, this.encoding), () => + serializeLogs(batch, this.encoding), + ); + } + if (this.spans.length > 0) { + const batch = this.spans.splice(0, MAX_BATCH); + this.store(batchId('traces', (this.sequence += 1), batch.length, this.encoding), () => + serializeSpans(batch, this.encoding), + ); + } + } + + private store(id: string, body: () => Uint8Array) { + try { + const evicted = this.storage.put(id, body()); + if (evicted.length > 0) { + // The id says how many records were in the batch, so an eviction costs a known number. + this.counters.droppedCacheFull += evicted.reduce((sum, key) => sum + batchRecords(key), 0); + this.reportDue = true; + } + } catch { + // A store that cannot store (disk full, quota) must not take the session down with it. + this.counters.droppedCacheError += batchRecords(id); + this.reportDue = true; + } + } + + private async send( + url: string, + body: Uint8Array, + records: number, + encoding: Encoding, + ): Promise<'sent' | 'drop' | 'keep'> { + const destination = this.destination!; + try { + const response = await fetch(url, { + method: 'POST', + headers: { + 'Content-Type': contentType(encoding), + // RFC 9218 lowest urgency: telemetry never wins over media on a shared uplink. + Priority: 'u=7', + ...destination.headers, + }, + body: body as BodyInit, + keepalive: body.byteLength <= KEEPALIVE_LIMIT, + }); + if (response.status >= 200 && response.status < 300) { + this.counters.sent += 1; + this.counters.bytes += body.byteLength; + return 'sent'; + } + if (response.status === 429 || response.status >= 500) { + const retryAfter = Number.parseInt(response.headers.get('Retry-After') ?? '', 10); + this.throttledUntil = + Date.now() + (Number.isFinite(retryAfter) ? retryAfter * 1000 : THROTTLE_DEFAULT_MS); + this.counters.failed += 1; + this.reportDue = true; + return 'keep'; + } + // A 4xx is the collector's verdict on the payload: retrying cannot fix it. + this.counters.droppedRejected += records; + this.reportDue = true; + return 'drop'; + } catch { + this.counters.failed += 1; + this.reportDue = true; + return 'keep'; + } + } + + private appendReport() { + this.reportDue = false; + const counters = this.counters; + const attributes: Attributes = { + 'lk.telemetry.uploads.sent': counters.sent, + 'lk.telemetry.uploads.bytes': counters.bytes, + 'lk.telemetry.uploads.failed': counters.failed, + 'lk.telemetry.queue.records': this.logs.length + this.spans.length, + 'lk.telemetry.cache.batches': this.storage.pending().length, + }; + if (counters.holdsCapped) attributes['lk.telemetry.holds.capped'] = counters.holdsCapped; + if (counters.droppedCacheFull) { + attributes['lk.telemetry.dropped.cache_full'] = counters.droppedCacheFull; + } + if (counters.droppedCacheError) { + attributes['lk.telemetry.dropped.cache_error'] = counters.droppedCacheError; + } + if (counters.droppedQueueFull) { + attributes['lk.telemetry.dropped.queue_full'] = counters.droppedQueueFull; + } + if (counters.droppedRejected) { + attributes['lk.telemetry.dropped.rejected'] = counters.droppedRejected; + } + if (counters.droppedThrottled) { + attributes['lk.telemetry.dropped.throttled'] = counters.droppedThrottled; + } + if (counters.droppedRateLimited) { + attributes['lk.telemetry.dropped.rate_limited'] = counters.droppedRateLimited; + } + const now = hrTime(); + this.logs.push({ + hrTime: now, + hrTimeObserved: now, + eventName: 'lk.telemetry.report', + severityNumber: Severity.info, + severityText: 'INFO', + body: 'lk.telemetry.report', + attributes, + droppedAttributesCount: 0, + resource: this.resource, + instrumentationScope: INSTRUMENTATION_SCOPE, + }); + } + + /** The collector said stop: throw the backlog away and never speak again (SPEC `Disabled`). */ + disable() { + this.disabled = true; + this.logs = []; + this.spans = []; + this.storage.clear(); + this.stop(); + } + + diagnostics(): string { + const counters = this.counters; + const state = this.disabled + ? 'disabled' + : !this.destination + ? 'no destination' + : Date.now() < this.throttledUntil + ? 'throttled' + : this.holds > 0 + ? 'held' + : 'ready'; + const lost = + counters.droppedCacheFull + + counters.droppedCacheError + + counters.droppedQueueFull + + counters.droppedRejected + + counters.droppedThrottled + + counters.droppedRateLimited; + return `telemetry ${state}: sent ${counters.sent} batches, ${counters.bytes} bytes, failed ${counters.failed}, queued ${this.logs.length + this.spans.length}, lost ${lost}`; + } + + stop() { + if (this.timer) { + clearInterval(this.timer); + this.timer = undefined; + } + } + + async shutdown() { + this.stop(); + this.processScope = undefined; + await this.flush(true); + } +} diff --git a/src/telemetry/scope.ts b/src/telemetry/scope.ts new file mode 100644 index 0000000000..27b082fbbe --- /dev/null +++ b/src/telemetry/scope.ts @@ -0,0 +1,366 @@ +/** + * One scope per Room connection: a trace id, the attributes every record of that call carries, the + * spans, and the stats windows. The scope is not ended — a call's last record is simply its last. + */ +import { + type Attributes, + INSTRUMENTATION_SCOPE, + type LogRecord, + Severity, + SpanKind, + type SpanRecord, + SpanStatus, + hrDuration, + hrTime, + randomHex, +} from './otlp'; +import type { Pipeline } from './pipeline'; + +export type Outcome = 'ok' | 'error' | 'cancelled'; + +export type TrackDirection = 'inbound' | 'outbound'; + +export type SeverityName = 'info' | 'warn' | 'error'; + +export interface RoomIdentity { + sid?: string; + name?: string; + participantSid?: string; + participantIdentity?: string; +} + +/** One track's reading, already in SPEC units: milliseconds, not the WebRTC seconds. */ +export interface StatsSample { + codec?: string; + bytes?: number; + packets?: number; + packetsLost?: number; + framesDropped?: number; + concealedSamples?: number; + concealmentEvents?: number; + silentConcealedSamples?: number; + jitterBufferDelayMs?: number; + qualityLimitationBandwidthMs?: number; + qualityLimitationCpuMs?: number; + qualityLimitationOtherMs?: number; + jitterMs?: number; + rttMs?: number; + fps?: number; + audioLevel?: number; +} + +export interface TraceContext { + traceId: string; + spanId: string; + traceFlags: number; +} + +/** An attribute nobody set is not an attribute: `undefined` would ship as an empty value. */ +function defined(attributes: Attributes): Attributes { + const out: Attributes = {}; + for (const [key, value] of Object.entries(attributes)) { + if (value !== undefined) out[key] = value; + } + return out; +} + +const COUNTERS = [ + ['bytes', 'lk.rtc.bytes'], + ['packets', 'lk.rtc.packets'], + ['packetsLost', 'lk.rtc.packets_lost'], + ['framesDropped', 'lk.rtc.frames_dropped'], + ['concealedSamples', 'lk.rtc.concealed_samples'], + ['concealmentEvents', 'lk.rtc.concealment_events'], + ['silentConcealedSamples', 'lk.rtc.silent_concealed_samples'], + ['jitterBufferDelayMs', 'lk.rtc.jitter_buffer_delay_ms'], + ['qualityLimitationBandwidthMs', 'lk.rtc.quality_limitation.bandwidth_ms'], + ['qualityLimitationCpuMs', 'lk.rtc.quality_limitation.cpu_ms'], + ['qualityLimitationOtherMs', 'lk.rtc.quality_limitation.other_ms'], +] as const; + +const GAUGES = [ + ['jitterMs', 'lk.rtc.jitter_ms'], + ['rttMs', 'lk.rtc.rtt_ms'], + ['fps', 'lk.rtc.fps'], + ['audioLevel', 'lk.rtc.audio_level'], +] as const; + +interface Gauge { + min: number; + max: number; + sum: number; + count: number; +} + +/** Counters are reported as the window's last reading (the W3C webrtc-stats model: monotonic); + * gauges as min/max/avg over it. */ +class Window { + started = Date.now(); + + samples = 0; + + codec?: string; + + last: StatsSample = {}; + + gauges = new Map(); + + add(sample: StatsSample) { + this.samples += 1; + this.codec = sample.codec ?? this.codec; + for (const [field] of COUNTERS) { + const value = sample[field]; + if (value !== undefined) this.last[field] = value; + } + for (const [field] of GAUGES) { + const value = sample[field]; + if (value === undefined || Number.isNaN(value)) continue; + const gauge = this.gauges.get(field); + if (!gauge) { + this.gauges.set(field, { min: value, max: value, sum: value, count: 1 }); + } else { + gauge.min = Math.min(gauge.min, value); + gauge.max = Math.max(gauge.max, value); + gauge.sum += value; + gauge.count += 1; + } + } + } + + attributes(): Attributes { + const attributes: Attributes = { + 'lk.rtc.window_ms': Date.now() - this.started, + 'lk.rtc.samples': this.samples, + }; + if (this.codec) attributes['lk.rtc.codec'] = this.codec; + for (const [field, key] of COUNTERS) { + const value = this.last[field]; + if (value !== undefined) attributes[key] = Math.round(value); + } + for (const [field, key] of GAUGES) { + const gauge = this.gauges.get(field); + if (!gauge) continue; + attributes[`${key}.min`] = gauge.min; + attributes[`${key}.max`] = gauge.max; + attributes[`${key}.avg`] = gauge.sum / gauge.count; + } + return attributes; + } +} + +export class TelemetrySpan { + private events: SpanRecord['events'] = []; + + private attributes: Attributes; + + private startTime = hrTime(); + + private finished = false; + + private spanId = randomHex(8); + + constructor( + private scope: TelemetryScope, + private pipeline: Pipeline, + private name: string, + private kind: number, + attributes: Attributes, + private parent?: { traceId: string; spanId: string; traceFlags: number }, + ) { + this.attributes = { ...attributes }; + } + + /** A checkpoint inside the attempt — `ws_open`, `join_recv`, `first_media`, `attempt 2 full`. */ + step(name: string) { + if (this.finished) return; + this.events.push({ name, time: hrTime(), attributes: {}, droppedAttributesCount: 0 }); + } + + setAttribute(key: string, value: Attributes[string]) { + this.attributes[key] = value; + } + + context() { + return { traceId: this.scope.traceId, spanId: this.spanId, traceFlags: 1 }; + } + + /** Ending twice is a no-op, as on every other platform. */ + end(outcome: Outcome = 'ok', errorType?: string, message?: string) { + if (this.finished) return; + this.finished = true; + const endTime = hrTime(); + this.attributes['lk.outcome'] = outcome; + if (errorType) this.attributes['error.type'] = errorType; + this.pipeline.endSpan({ + name: this.name, + kind: this.kind, + spanContext: () => this.context(), + parentSpanContext: this.parent, + startTime: this.startTime, + endTime, + duration: hrDuration(this.startTime, endTime), + // Cancellation is `Unset` like success: only a failure is an error (SPEC). + status: + outcome === 'error' ? { code: SpanStatus.error, message } : { code: SpanStatus.unset }, + attributes: defined({ ...this.scope.attributes(), ...this.attributes }), + links: [], + events: this.events, + ended: true, + resource: { attributes: {} }, + instrumentationScope: INSTRUMENTATION_SCOPE, + droppedAttributesCount: 0, + droppedEventsCount: 0, + droppedLinksCount: 0, + }); + } + + fail(error: unknown) { + const type = error instanceof Error ? error.name : 'unknown'; + const message = error instanceof Error ? error.message : String(error); + this.end('error', type, message); + } + + cancel() { + this.end('cancelled'); + } +} + +export class TelemetryScope { + readonly traceId = randomHex(16); + + private room: RoomIdentity = {}; + + private windows = new Map(); + + private pendingSubscribes = new Map(); + + constructor(private pipeline: Pipeline) {} + + attributes(): Attributes { + return { + 'session.id': this.traceId, + 'lk.room.sid': this.room.sid, + 'lk.room.name': this.room.name, + 'lk.participant.sid': this.room.participantSid, + 'lk.participant.identity': this.room.participantIdentity, + }; + } + + setRoom(identity: RoomIdentity) { + this.room = { ...this.room, ...identity }; + } + + start( + name: string, + options: { kind?: number; attributes?: Attributes; parent?: TelemetrySpan } = {}, + ): TelemetrySpan { + return new TelemetrySpan( + this, + this.pipeline, + name, + options.kind ?? SpanKind.internal, + options.attributes ?? {}, + options.parent?.context(), + ); + } + + emit( + eventName: string, + attributes: Attributes = {}, + severity: SeverityName = 'info', + span?: TelemetrySpan, + ) { + const now = hrTime(); + const record: LogRecord = { + hrTime: now, + hrTimeObserved: now, + eventName, + severityNumber: Severity[severity], + severityText: severity.toUpperCase(), + // Log viewers key their line on the body, and not every backend surfaces event_name yet. + body: eventName, + attributes: defined({ + ...this.attributes(), + // semconv 1.39, duplicating the record's own event name for backends that do not surface + // the field yet (Loki included) — the Rust core writes it too, so a query works either way. + 'otel.event.name': eventName || undefined, + ...attributes, + }), + droppedAttributesCount: 0, + resource: { attributes: {} }, + instrumentationScope: INSTRUMENTATION_SCOPE, + spanContext: span?.context() ?? { traceId: this.traceId, spanId: '', traceFlags: 1 }, + }; + this.pipeline.record(record, { exemptFromFlood: eventName === 'lk.rtc.stats.sample' }); + } + + disconnected(reason: string) { + this.emit( + 'lk.room.disconnected', + { 'lk.disconnect.reason': reason }, + reason === 'client_initiated' ? 'info' : 'warn', + ); + } + + /** The intent to subscribe exists; the window that sees the first inbound bytes ends the span. */ + subscribeStarted(sid: string, attributes: Attributes) { + if (this.pendingSubscribes.has(sid)) return; + const span = this.start('lk.subscribe', { attributes: { 'lk.track.sid': sid, ...attributes } }); + span.step('subscribed'); + this.pendingSubscribes.set(sid, span); + } + + subscribeEnded(sid: string, outcome: Outcome, errorType?: string) { + const span = this.pendingSubscribes.get(sid); + if (!span) return; + this.pendingSubscribes.delete(sid); + span.end(outcome, errorType); + } + + recordStats( + sid: string, + kind: 'audio' | 'video', + direction: TrackDirection, + sample: StatsSample, + ) { + const key = `${sid}:${direction}`; + let entry = this.windows.get(key); + if (!entry) { + entry = { window: new Window(), kind, direction }; + this.windows.set(key, entry); + } + entry.window.add(sample); + if (direction === 'inbound' && (sample.bytes ?? 0) > 0) { + const span = this.pendingSubscribes.get(sid); + if (span) { + span.step('first_media'); + this.subscribeEnded(sid, 'ok'); + } + } + if (Date.now() - entry.window.started >= this.pipeline.statsWindow * 1000) { + this.closeWindow(sid, key); + } + } + + private closeWindow(sid: string, key: string) { + const entry = this.windows.get(key); + if (!entry || entry.window.samples === 0) return; + this.windows.delete(key); + this.emit('lk.rtc.stats.sample', { + 'lk.track.sid': sid, + 'lk.track.kind': entry.kind, + 'lk.track.direction': entry.direction, + ...entry.window.attributes(), + }); + } + + /** Closes every open window early — the call is ending and a partial window beats none. */ + close() { + for (const key of Array.from(this.windows.keys())) { + this.closeWindow(key.slice(0, key.lastIndexOf(':')), key); + } + for (const sid of Array.from(this.pendingSubscribes.keys())) { + this.subscribeEnded(sid, 'cancelled'); + } + } +} diff --git a/src/telemetry/storage.ts b/src/telemetry/storage.ts new file mode 100644 index 0000000000..ad1d25bc9a --- /dev/null +++ b/src/telemetry/storage.ts @@ -0,0 +1,92 @@ +/** + * Where batches wait between being made and being accepted. This is the write-ahead cache the Rust + * core calls `BatchCache`: every batch is stored *before* the network is tried, and removed only + * when the collector has taken it, so a crash, a kill or an offline hour costs nothing. + * + * The interface is deliberately the same five operations as the core's, and deliberately + * synchronous — the queue path has no await in it, and a store that forces one turns the pipeline + * into a state machine. A platform with a filesystem implements this; the default keeps batches in + * memory, which is all a browser tab needs (TELEMETRY.md §3). + */ +import type { Encoding } from './otlp'; + +export interface TelemetryStorage { + /** Store a batch. Returns the ids evicted to stay inside the store's own bounds. */ + put(id: string, body: Uint8Array): string[]; + /** Stored ids, oldest first. */ + pending(): string[]; + read(id: string): Uint8Array | undefined; + remove(id: string): void; + clear(): void; +} + +/** + * Ids sort oldest-first as plain strings, so a store never has to parse or stat anything — and + * they carry everything needed to send a batch that was written by an earlier run of the app: + * which route it belongs to, how it was encoded, and how many records it cost. + */ +export function batchId( + kind: 'logs' | 'traces', + sequence: number, + records: number, + encoding: Encoding, +): string { + const now = String(Date.now()).padStart(15, '0'); + const suffix = `${kind === 'logs' ? 'l' : 't'}${encoding === 'json' ? 'j' : 'p'}`; + return `${now}-${String(sequence).padStart(6, '0')}-${records}-${suffix}`; +} + +export function batchKind(id: string): 'logs' | 'traces' { + return id.split('-').pop()?.startsWith('t') === true ? 'traces' : 'logs'; +} + +/** A cached batch keeps the encoding it was written with: the collector is told the truth. */ +export function batchEncoding(id: string): Encoding { + return id.endsWith('j') ? 'json' : 'protobuf'; +} + +/** How many records a batch holds, so an eviction costs a known number and not a guess. */ +export function batchRecords(id: string): number { + return Number.parseInt(id.split('-')[2] ?? '0', 10) || 0; +} + +/** Lost with the tab, bounded by bytes and by count; the oldest goes first, and one always stays. */ +export class MemoryStorage implements TelemetryStorage { + private batches: Array<[string, Uint8Array]> = []; + + constructor( + private maxBytes: number, + private maxBatches: number, + ) {} + + put(id: string, body: Uint8Array): string[] { + this.batches.push([id, body]); + const evicted: string[] = []; + let total = this.batches.reduce((sum, [, batch]) => sum + batch.byteLength, 0); + while ( + (total > this.maxBytes || this.batches.length > this.maxBatches) && + this.batches.length > 1 + ) { + const [oldest, batch] = this.batches.shift()!; + total -= batch.byteLength; + evicted.push(oldest); + } + return evicted; + } + + pending(): string[] { + return this.batches.map(([id]) => id); + } + + read(id: string): Uint8Array | undefined { + return this.batches.find(([key]) => key === id)?.[1]; + } + + remove(id: string) { + this.batches = this.batches.filter(([key]) => key !== id); + } + + clear() { + this.batches = []; + } +} diff --git a/src/telemetry/telemetry.browser.test.ts b/src/telemetry/telemetry.browser.test.ts new file mode 100644 index 0000000000..4b33111f3b --- /dev/null +++ b/src/telemetry/telemetry.browser.test.ts @@ -0,0 +1,75 @@ +import { expect, inject, test } from 'vitest'; +import { Telemetry } from '.'; +import Room, { ConnectionState } from '../room/Room'; +import { RoomEvent } from '../room/events'; + +/** + * One session through the whole integrated path: a real Chromium with fake media devices, a real + * `livekit-server --dev`, and the collector that fans out to the same Grafana LGTM stack the mobile + * harness writes to. Two Rooms in one page, so the session has an inbound and an outbound side. + * + * pnpm vitest run --config vitest.telemetry.config.mts + */ +const { url, endpoint, headers, publisherToken, subscriberToken } = inject('telemetry'); + +async function poll(what: string, condition: () => boolean, timeout = 20_000) { + const deadline = Date.now() + timeout; + while (Date.now() < deadline) { + if (condition()) return; + await new Promise((resolve) => setTimeout(resolve, 200)); + } + throw new Error(`timed out waiting for ${what}`); +} + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +test('a session reports itself: connect span, stats windows, disconnect', async () => { + Telemetry.configure({ + endpoint, + headers, + flushInterval: 1, + statsWindow: 2, + resource: { + 'service.name': 'livekit-client-js', + 'service.version': '2.22.3-telemetry', + 'os.name': 'browser', + }, + }); + expect(Telemetry.enabled).toBe(true); + + const publisher = new Room(); + const subscriber = new Room(); + await publisher.connect(url, publisherToken); + await subscriber.connect(url, subscriberToken); + expect(publisher.state).toBe(ConnectionState.Connected); + + const media = await navigator.mediaDevices.getUserMedia({ audio: true, video: true }); + await publisher.localParticipant.publishTrack(media.getAudioTracks()[0]); + await publisher.localParticipant.publishTrack(media.getVideoTracks()[0]); + + await poll('the subscriber to receive both tracks', () => { + const remote = Array.from(subscriber.remoteParticipants.values())[0]; + return !!remote?.audioTrackPublications.size && !!remote?.videoTrackPublications.size; + }); + + // Two stats windows on both sides… + await sleep(6000); + + // …then both reconnect paths, which are their own spans. + for (const scenario of ['resume-reconnect', 'full-reconnect'] as const) { + const reconnected = new Promise((resolve) => publisher.once(RoomEvent.Reconnected, resolve)); + await publisher.simulateScenario(scenario); + await reconnected; + await sleep(1000); + } + + await publisher.disconnect(); + await subscriber.disconnect(); + await sleep(1500); + await Telemetry.flush(); + + const diagnostics = Telemetry.diagnostics(); + console.log('telemetry:', diagnostics); + expect(diagnostics).toContain('lost 0'); + expect(diagnostics).not.toContain('sent 0 batches'); +}, 90_000); diff --git a/src/telemetry/telemetry.test.ts b/src/telemetry/telemetry.test.ts new file mode 100644 index 0000000000..b6d02ef0d4 --- /dev/null +++ b/src/telemetry/telemetry.test.ts @@ -0,0 +1,336 @@ +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; +import { Telemetry } from '.'; +import type { Backend, Scope, Span } from './backend'; +import { cadenceFactor, changes, networkType } from './device'; +import { Pipeline } from './pipeline'; +import { TelemetryScope } from './scope'; +import { MemoryStorage, batchId } from './storage'; + +/** The JSON encoding is the readable one: every assertion here reads the body the collector gets. */ +function bodyOf(call: unknown[]): any { + const init = call[1] as RequestInit; + return JSON.parse(new TextDecoder().decode(init.body as Uint8Array)); +} + +function recordsOf(call: unknown[]): any[] { + return bodyOf(call).resourceLogs.flatMap((r: any) => + r.scopeLogs.flatMap((s: any) => s.logRecords), + ); +} + +function attributesOf(record: any): Record { + return Object.fromEntries(record.attributes.map((a: any) => [a.key, Object.values(a.value)[0]])); +} + +describe('telemetry pipeline', () => { + let pipeline: Pipeline; + let fetchMock: ReturnType; + + beforeEach(() => { + fetchMock = vi.fn(async () => new Response(null, { status: 200 })); + vi.stubGlobal('fetch', fetchMock); + pipeline = new Pipeline(); + pipeline.configure({ + endpoint: 'http://collector.test/v1/logs', + encoding: 'json', + flushInterval: 3600, + statsWindow: 0.01, + }); + }); + + afterEach(() => { + pipeline.stop(); + vi.unstubAllGlobals(); + }); + + test('one window of readings becomes one record, counters last and gauges summarised', async () => { + const scope = new TelemetryScope(pipeline); + scope.setRoom({ sid: 'RM_1', name: 'harness', participantIdentity: 'publisher' }); + scope.recordStats('TR_1', 'video', 'outbound', { + bytes: 1000, + packets: 10, + rttMs: 20, + fps: 30, + }); + await new Promise((resolve) => setTimeout(resolve, 20)); + scope.recordStats('TR_1', 'video', 'outbound', { + bytes: 3000, + packets: 30, + rttMs: 40, + fps: 24, + }); + await pipeline.flush(true); + + const window = recordsOf(fetchMock.mock.calls[0]).find( + (r) => r.eventName === 'lk.rtc.stats.sample', + ); + const attributes = attributesOf(window); + expect(attributes['lk.track.sid']).toBe('TR_1'); + expect(attributes['lk.track.direction']).toBe('outbound'); + expect(attributes['lk.room.name']).toBe('harness'); + // A counter is the window's last reading, not a sum of them (the W3C webrtc-stats model). + expect(attributes['lk.rtc.bytes']).toBe(3000); + expect(attributes['lk.rtc.samples']).toBe(2); + expect(attributes['lk.rtc.rtt_ms.min']).toBe(20); + expect(attributes['lk.rtc.rtt_ms.max']).toBe(40); + expect(attributes['lk.rtc.rtt_ms.avg']).toBe(30); + }); + + test('a hold stops uploads, never collection', async () => { + const scope = new TelemetryScope(pipeline); + pipeline.hold(true); + scope.emit('lk.test.one'); + await pipeline.flush(); + expect(fetchMock).not.toHaveBeenCalled(); + + scope.emit('lk.test.two'); + pipeline.hold(false); + await pipeline.flush(); + // Everything collected during the hold went out together: a pause is not a hole. + expect(recordsOf(fetchMock.mock.calls[0]).map((r) => r.eventName)).toEqual([ + 'lk.test.one', + 'lk.test.two', + ]); + }); + + test('a 429 holds the pipeline and keeps the batch', async () => { + fetchMock.mockResolvedValueOnce(new Response(null, { status: 429 })); + const scope = new TelemetryScope(pipeline); + scope.emit('lk.test.throttled'); + await pipeline.flush(true); + expect(pipeline.diagnostics()).toContain('throttled'); + expect(pipeline.diagnostics()).toContain('lost 0'); + + // The next tick is not spent on a request the collector already refused. + await pipeline.flush(); + expect(fetchMock).toHaveBeenCalledTimes(1); + + // …and the record is still there when the hold lifts. + await pipeline.flush(true); + expect(recordsOf(fetchMock.mock.calls[1])[0].eventName).toBe('lk.test.throttled'); + }); + + test('the queue evicts the oldest and says how many', async () => { + pipeline.maxQueueSize = 3; + const scope = new TelemetryScope(pipeline); + for (let i = 0; i < 5; i += 1) { + scope.emit(`lk.test.${i}`); + } + await pipeline.flush(true); + + const records = recordsOf(fetchMock.mock.calls[0]); + expect(records.map((r) => r.eventName)).toContain('lk.test.4'); + expect(records.map((r) => r.eventName)).not.toContain('lk.test.0'); + const report = records.find((r) => r.eventName === 'lk.telemetry.report'); + expect(attributesOf(report)['lk.telemetry.dropped.queue_full']).toBe(2); + }); + + test('a span carries its checkpoints and its outcome', async () => { + const scope = new TelemetryScope(pipeline); + const span = scope.start('lk.connect', { attributes: { 'lk.connect.attempt': 1 } }); + span.step('ws_open'); + span.step('join_recv'); + span.end('ok'); + span.end('error'); // ending twice is a no-op + await pipeline.flush(true); + + const traceCall = fetchMock.mock.calls.find((call) => String(call[0]).endsWith('/v1/traces'))!; + const spans = bodyOf(traceCall).resourceSpans.flatMap((r: any) => + r.scopeSpans.flatMap((s: any) => s.spans), + ); + expect(spans).toHaveLength(1); + expect(spans[0].name).toBe('lk.connect'); + expect(spans[0].events.map((e: any) => e.name)).toEqual(['ws_open', 'join_recv']); + expect(attributesOf(spans[0])['lk.outcome']).toBe('ok'); + }); + + test('an SDK nobody asked for telemetry collects nothing', async () => { + const idle = new Pipeline(); + const scope = new TelemetryScope(idle); + scope.emit('lk.test.void'); + await idle.flush(true); + expect(fetchMock).not.toHaveBeenCalled(); + expect(idle.diagnostics()).toContain('no destination'); + }); + + test('records made before the destination is known are kept, not dropped', async () => { + // `registerGlobals` configures the resource long before a connect names the collector. + const early = new Pipeline(); + early.configure({ encoding: 'json', flushInterval: 3600 }); + const scope = new TelemetryScope(early); + scope.emit('lk.test.early'); + await early.flush(true); + expect(fetchMock).not.toHaveBeenCalled(); + + early.setServer('wss://project.livekit.cloud', 'token'); + await early.flush(true); + const call = fetchMock.mock.calls[0]; + expect(String(call[0])).toBe('https://project.livekit.cloud/observability/client/logs/otlp/v0'); + expect(recordsOf(call)[0].eventName).toBe('lk.test.early'); + early.stop(); + }); +}); + +describe('device state', () => { + test('only a change is a record, and the connection is one record', () => { + expect(changes({}, { appState: 'foreground' }).map((c) => c.event)).toEqual([ + 'lk.device.app_state.changed', + ]); + expect(changes({ appState: 'foreground' }, { appState: 'foreground' })).toEqual([]); + + // type, expensive and constrained are one event, not three. + const network = changes( + { networkType: 'wifi', networkExpensive: false, networkConstrained: false }, + { networkType: 'cell', networkExpensive: true, networkConstrained: false }, + ); + expect(network).toHaveLength(1); + expect(network[0].event).toBe('lk.device.network.changed'); + expect(network[0].attributes['network.connection.type']).toBe('cell'); + expect(network[0].attributes['lk.device.network.expensive']).toBe(true); + }); + + test('factors multiply and stop at 4x', () => { + expect(cadenceFactor({})).toBe(1); + expect(cadenceFactor({ appState: 'background' })).toBe(2); + // The rows a page cannot fill in are not here at all: a platform that knows heat, power or + // memory pressure reports them to its own backend (TELEMETRY.md §5). + expect(cadenceFactor({ appState: 'background', networkConstrained: true })).toBe(4); + }); + + test('NetworkInformation names become SPEC names', () => { + expect(networkType('cellular')).toBe('cell'); + expect(networkType('ethernet')).toBe('wired'); + expect(networkType('none')).toBe('unavailable'); + expect(networkType(undefined)).toBe('unknown'); + }); + + test('the factor stretches both periods, and relief applies at once', () => { + const pipeline = new Pipeline(); + pipeline.configure({ + endpoint: 'http://collector.test/v1/logs', + flushInterval: 15, + statsWindow: 15, + }); + expect(pipeline.statsWindow).toBe(15); + pipeline.setCadenceFactor(4); + expect(pipeline.flushInterval).toBe(60); + expect(pipeline.statsWindow).toBe(60); + pipeline.setCadenceFactor(1); + expect(pipeline.statsWindow).toBe(15); + pipeline.stop(); + }); +}); + +describe('what a platform reports', () => { + test('an event it names and a factor it chose both land', async () => { + // React Native sees heat and power; this package has no vocabulary for either, so the platform + // names the record and sets the number. Nothing mobile appears on this side. + const fetchMock = vi.fn(async () => new Response(null, { status: 200 })); + vi.stubGlobal('fetch', fetchMock); + const pipeline = new Pipeline(); + pipeline.configure({ + endpoint: 'http://collector.test/v1/logs', + encoding: 'json', + flushInterval: 15, + statsWindow: 15, + }); + + pipeline.emit('lk.device.thermal.changed', { 'lk.device.thermal.state': 'serious' }); + pipeline.setCadenceFactor(2); + await pipeline.flush(true); + + const record = recordsOf(fetchMock.mock.calls[0])[0]; + expect(record.eventName).toBe('lk.device.thermal.changed'); + expect(attributesOf(record)['lk.device.thermal.state']).toBe('serious'); + expect(pipeline.statsWindow).toBe(30); + pipeline.stop(); + vi.unstubAllGlobals(); + }); +}); + +describe('the write-ahead cache', () => { + let fetchMock: ReturnType; + + beforeEach(() => { + fetchMock = vi.fn(async () => new Response(null, { status: 200 })); + vi.stubGlobal('fetch', fetchMock); + }); + + afterEach(() => vi.unstubAllGlobals()); + + test('a batch the network refused is still there next time', async () => { + const storage = new MemoryStorage(4 * 1024 * 1024, 512); + const pipeline = new Pipeline(); + pipeline.configure({ + endpoint: 'http://collector.test/v1/logs', + encoding: 'json', + flushInterval: 3600, + storage, + }); + const scope = new TelemetryScope(pipeline); + scope.emit('lk.test.offline'); + + fetchMock.mockRejectedValueOnce(new TypeError('offline')); + await pipeline.flush(true); + // The record left the queue, but it is in the cache, not gone. + expect(storage.pending()).toHaveLength(1); + expect(pipeline.diagnostics()).toContain('lost 0'); + + await pipeline.flush(true); + expect(storage.pending()).toHaveLength(0); + expect(recordsOf(fetchMock.mock.calls[1])[0].eventName).toBe('lk.test.offline'); + pipeline.stop(); + }); + + test('a backlog from a previous launch replays four batches a tick', async () => { + // What the store looks like after a crash: batches nobody has sent yet. + const storage = new MemoryStorage(4 * 1024 * 1024, 512); + for (let i = 0; i < 6; i += 1) { + storage.put(batchId('logs', i, 10, 'protobuf'), new Uint8Array([1, 2, 3])); + } + const pipeline = new Pipeline(); + pipeline.configure({ endpoint: 'http://collector.test/v1/logs', flushInterval: 3600, storage }); + + await pipeline.flush(); + expect(fetchMock).toHaveBeenCalledTimes(4); + expect(storage.pending()).toHaveLength(2); + + // A shutdown drains without the budget. + await pipeline.flush(true); + expect(storage.pending()).toHaveLength(0); + pipeline.stop(); + }); + + test('an eviction costs a known number of records, not a batch', async () => { + const storage = new MemoryStorage(64, 512); + const pipeline = new Pipeline(); + // No destination: batches pile up in the cache, which is where eviction happens. + pipeline.configure({ encoding: 'json', flushInterval: 3600, storage }); + const scope = new TelemetryScope(pipeline); + scope.emit('lk.test.one'); + await pipeline.flush(true); + scope.emit('lk.test.two'); + await pipeline.flush(true); + + expect(storage.pending()).toHaveLength(1); + // The evicted batch held one record, and the report says so — not "one batch". + expect(pipeline.diagnostics()).toContain('lost 1'); + pipeline.stop(); + }); + + test('a cached batch keeps the encoding it was written with', async () => { + const storage = new MemoryStorage(4 * 1024 * 1024, 512); + const pipeline = new Pipeline(); + pipeline.configure({ encoding: 'json', flushInterval: 3600, storage }); + new TelemetryScope(pipeline).emit('lk.test.written_as_json'); + await pipeline.flush(true); // no destination yet: the batch is cached as JSON + + // The app upgrades, or simply flips the switch, before the batch ever left. + pipeline.configure({ endpoint: 'http://collector.test/v1/logs', encoding: 'protobuf' }); + await pipeline.flush(true); + + const headers = (fetchMock.mock.calls[0][1] as RequestInit).headers as Record; + expect(headers['Content-Type']).toBe('application/json'); + pipeline.stop(); + }); +}); diff --git a/src/telemetry/telemetrySetup.ts b/src/telemetry/telemetrySetup.ts new file mode 100644 index 0000000000..7910c4fcfa --- /dev/null +++ b/src/telemetry/telemetrySetup.ts @@ -0,0 +1,42 @@ +import type { TestProject } from 'vitest/node'; +import { createToken } from '../test/signalToken'; + +/** + * vitest globalSetup for the telemetry session test. It runs in Node (the browser cannot sign a + * token) and hands the browser a room and two identities on a live `livekit-server --dev`, plus + * the collector the session reports to — the same pair the mobile harness uses: + * + * livekit-server --dev + * otelcol-contrib --config src/telemetry/otelcol-web.yaml + * + * `LK_TELEMETRY_ENDPOINT` and `LK_TELEMETRY_TOKEN` point the same session at LiveKit Cloud + * instead (`https:///observability/client/logs/otlp/v0`, a token with an + * `observability:write` grant) — how the pipeline gets exercised against the real ingest. + */ +export default async function setup({ provide }: TestProject) { + const url = process.env.LK_URL ?? 'ws://127.0.0.1:7880'; + const endpoint = process.env.LK_TELEMETRY_ENDPOINT ?? 'http://127.0.0.1:4320/v1/logs'; + const room = `telemetry-${Date.now()}`; + const token = process.env.LK_TELEMETRY_TOKEN; + provide('telemetry', { + url, + endpoint, + headers: token ? { Authorization: `Bearer ${token}` } : {}, + room, + publisherToken: await createToken({ room, identity: 'publisher' }), + subscriberToken: await createToken({ room, identity: 'subscriber' }), + }); +} + +declare module 'vitest' { + interface ProvidedContext { + telemetry: { + url: string; + endpoint: string; + headers: Record; + room: string; + publisherToken: string; + subscriberToken: string; + }; + } +} diff --git a/src/telemetry/webrtc.ts b/src/telemetry/webrtc.ts new file mode 100644 index 0000000000..62f2eee849 --- /dev/null +++ b/src/telemetry/webrtc.ts @@ -0,0 +1,67 @@ +/** + * The SDK's per-track monitors already parse `getStats()` every two seconds; this turns what they + * hold into one SPEC reading, in SPEC units. Nothing here calls `getStats()` — measuring a call + * must not cost a second poll. + */ +import type { + AudioReceiverStats, + AudioSenderStats, + VideoReceiverStats, + VideoSenderStats, +} from '../room/stats'; +import type { StatsSample } from './scope'; + +const seconds = (value: number | undefined): number | undefined => + value === undefined ? undefined : value * 1000; + +function add(total: number | undefined, value: number | undefined): number | undefined { + if (value === undefined) return total; + return (total ?? 0) + value; +} + +/** Simulcast layers are one track: their counters sum, their gauges take the liveliest layer. */ +export function senderSample(stats: Array): StatsSample { + const sample: StatsSample = {}; + for (const layer of stats) { + sample.bytes = add(sample.bytes, layer.bytesSent); + sample.packets = add(sample.packets, layer.packetsSent); + sample.packetsLost = add(sample.packetsLost, layer.packetsLost); + sample.jitterMs = Math.max(sample.jitterMs ?? 0, seconds(layer.jitter) ?? 0) || undefined; + sample.rttMs = Math.max(sample.rttMs ?? 0, seconds(layer.roundTripTime) ?? 0) || undefined; + if (layer.type === 'video') { + sample.fps = Math.max(sample.fps ?? 0, layer.framesPerSecond ?? 0) || undefined; + const durations = layer.qualityLimitationDurations; + if (durations) { + sample.qualityLimitationBandwidthMs = add( + sample.qualityLimitationBandwidthMs, + seconds(durations.bandwidth), + ); + sample.qualityLimitationCpuMs = add(sample.qualityLimitationCpuMs, seconds(durations.cpu)); + sample.qualityLimitationOtherMs = add( + sample.qualityLimitationOtherMs, + seconds(durations.other), + ); + } + } + } + return sample; +} + +export function receiverSample(stats: AudioReceiverStats | VideoReceiverStats): StatsSample { + const sample: StatsSample = { + codec: 'mimeType' in stats ? stats.mimeType : undefined, + bytes: stats.bytesReceived, + packets: stats.packetsReceived, + packetsLost: stats.packetsLost, + jitterMs: seconds(stats.jitter), + jitterBufferDelayMs: seconds(stats.jitterBufferDelay), + }; + if (stats.type === 'video') { + sample.framesDropped = stats.framesDropped; + } else { + sample.concealedSamples = stats.concealedSamples; + sample.concealmentEvents = stats.concealmentEvents; + sample.silentConcealedSamples = stats.silentConcealedSamples; + } + return sample; +} diff --git a/src/test/signalServerSetup.ts b/src/test/signalServerSetup.ts index af8a375ff3..2c14f49a42 100644 --- a/src/test/signalServerSetup.ts +++ b/src/test/signalServerSetup.ts @@ -59,7 +59,6 @@ export default async function setup(project: { provide: (name: string, value: un project.provide('e2eUnavailable', ''); }; const skip = (reason: string) => { - // eslint-disable-next-line no-console console.warn(`\n[e2e] signal-connection e2e suite SKIPPED: ${reason}\n`); project.provide('serverUrl', ''); project.provide('e2eUnavailable', reason); diff --git a/vite.config.mjs b/vite.config.mjs index af40737514..774b4ce458 100644 --- a/vite.config.mjs +++ b/vite.config.mjs @@ -44,6 +44,6 @@ export default defineConfig({ environment: 'happy-dom', // e2e tests need a real server + node WebSocket; run them via the // dedicated `pnpm test:e2e` config (vitest.e2e.config.mts), not the unit run. - exclude: [...configDefaults.exclude, '**/*.e2e.test.ts'], + exclude: [...configDefaults.exclude, '**/*.e2e.test.ts', '**/*.browser.test.ts'], }, }); diff --git a/vitest.telemetry.config.mts b/vitest.telemetry.config.mts new file mode 100644 index 0000000000..d84c184ec6 --- /dev/null +++ b/vitest.telemetry.config.mts @@ -0,0 +1,29 @@ +import { playwright } from '@vitest/browser-playwright'; +import { defineConfig } from 'vitest/config'; + +// The telemetry session test: a real browser, a real livekit-server --dev, a real collector. +// Prerequisites are checked by the globalSetup, which also mints the tokens (the browser cannot). +export default defineConfig({ + test: { + include: ['src/telemetry/*.browser.test.ts'], + globalSetup: ['./src/telemetry/telemetrySetup.ts'], + testTimeout: 90_000, + hookTimeout: 60_000, + fileParallelism: false, + browser: { + enabled: true, + provider: playwright({ + launchOptions: { + args: [ + '--use-fake-device-for-media-stream', + '--use-fake-ui-for-media-stream', + '--autoplay-policy=no-user-gesture-required', + ], + }, + }), + headless: true, + instances: [{ browser: 'chromium' }], + screenshotFailures: false, + }, + }, +});