feat(telemetry): add client telemetry

.changeset/client-telemetry-pipeline.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': minor
+---
+
+Add client telemetry: every Room on LiveKit Cloud reports its connect, reconnect, publish and subscribe spans, RTC statistics windows, SDK warnings and device state to its own project when the token carries the observability grant; `disableTelemetry()` opts the process out, `Room.emitTelemetryEvent()` and `Room.setTelemetryAttribute()` add the app's own events and correlation ids.

.github/workflows/e2e.yaml

@@ -24,6 +24,28 @@ jobs:
         image: livekit/test-server:latest
         ports:
           - 9999:9999
+      # The telemetry story (src/telemetry/telemetry.browser.test.ts) runs a whole call on a real
+      # server and reports it to a real OTLP collector; without them it skips.
+      livekit:
+        image: livekit/livekit-server:latest
+        env:
+          LIVEKIT_KEYS: 'devkey: secret'
+          # media over the published ports only: one UDP port, candidates on 127.0.0.1
+          LIVEKIT_CONFIG: |
+            port: 7880
+            rtc:
+              tcp_port: 7881
+              udp_port: 7882
+              use_external_ip: false
+              node_ip: 127.0.0.1
+        ports:
+          - 7880:7880
+          - 7881:7881
+          - 7882:7882/udp
+      otelcol:
+        image: otel/opentelemetry-collector-contrib:latest
+        ports:
+          - 4318:4318
 
     steps:
       - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6.1.0
@@ -53,6 +75,9 @@ jobs:
         env:
           # Connect to the test-server service above (validate + WS on 9999).
           LK_TEST_SERVER_URL: ws://127.0.0.1:9999
+          # The telemetry story's server and collector (services above).
+          LK_URL: ws://127.0.0.1:7880
+          LK_TELEMETRY_ENDPOINT: http://127.0.0.1:4318
 
       - name: Collect test-server logs
         if: failure()

.size-limit.cjs

@@ -7,6 +7,9 @@ module.exports = [
   {
     path: 'dist/livekit-client.umd.js',
     import: '{ Room }',
-    limit: '130 kB',
+    // Telemetry cannot be shaken out of the UMD bundle: 130 kB no longer fits, and after the
+    // review rounds the bundle sits at the 135 kB raised first. Raising it is one of two answers —
+    // the other is a separate UMD entry point, as the e2ee and frame-metadata workers already have.
+    limit: '140 kB',
   },
 ];

TELEMETRY.md

@@ -0,0 +1,20 @@
+# Client telemetry (maintainers)
+
+The contract — events, attributes, cadences, destination and upload policy — is
+[`livekit-telemetry/SPEC.md`](https://github.com/livekit/rust-sdks/blob/main/livekit-telemetry/SPEC.md)
+in rust-sdks. The Swift, Kotlin and Dart SDKs get it from the Rust core; a browser cannot link
+that core, so `src/telemetry/` is the same pipeline in TypeScript, and `@livekit/react-native`
+reuses it (it only wires `configureTelemetryHost()`: a file cache, device state, resource names).
+
+| File | Role |
+|---|---|
+| `index.ts` | installs the pipeline with the first Room; `disableTelemetry()`; the host hook; SDK warn/error capture |
+| `scope.ts` | one per Room: trace id, Room attributes, `lk.connect` / `lk.reconnect` / `lk.publish` / `lk.subscribe`, RTC windows, the two app calls |
+| `pipeline.ts` | queue → write-ahead cache → one request in flight; projects, tokens, holds, budget, the answer table, the self-report, opt-out |
+| `transport.ts` | Cloud URL rules, token claims, answer classification, Retry-After, backoff, the request |
+| `otlp.ts` | record shapes and a protobuf writer for the two OTLP export requests |
+| `webrtc.ts`, `poller.ts` | one `getStats()` per peer connection → SPEC readings → windows |
+| `storage.ts`, `device.ts` | the cache interface and memory store; device state → events, cadence, holds |
+
+Tests: `telemetry.test.ts` (unit, mock fetch) and `telemetry.browser.test.ts` (the whole story in
+Chromium against `livekit-server --dev` and a local collector; see the PR for the commands).

src/api/SignalClient.ts

@@ -224,6 +224,9 @@ export class SignalClient {
 
   onTokenRefresh?: (token: string) => void;
 
+  /** @internal The WebSocket upgrade completed (telemetry's `ws_open` checkpoint). */
+  onWebSocketOpen?: () => void;
+
   onLeave?: (leave: LeaveRequest) => void;
 
   onRequestResponse?: (response: RequestResponse) => void;
@@ -578,6 +581,7 @@ export class SignalClient {
           if (!connection) {
             return;
           }
+          this.onWebSocketOpen?.();
           const signalReader = connection.readable.getReader();
           this.streamWriter = connection.writable.getWriter();
 

src/e2ee/E2eeManager.ts

@@ -3,7 +3,14 @@ import { EventEmitter } from 'events';
 import type TypedEventEmitter from 'typed-emitter';
 import type { FrameMetadata } from '../frameMetadata/types';
 import { hasFrameMetadataPublishOptions } from '../frameMetadata/utils';
-import { LogLevel, LoggerNames, getLogger, onWorkerLogLevelChanged, workerLogger } from '../logger';
+import {
+  LOG_OWNER,
+  LogLevel,
+  LoggerNames,
+  getLogger,
+  onWorkerLogLevelChanged,
+  workerLogger,
+} from '../logger';
 import type RTCEngine from '../room/RTCEngine';
 import type Room from '../room/Room';
 import { ConnectionState } from '../room/Room';
@@ -22,6 +29,7 @@ import {
   isScriptTransformSupportedForWorker,
   isVideoTrack,
 } from '../room/utils';
+import { telemetry } from '../telemetry';
 import type { NonSharedUint8Array } from '../type-polyfills/non-shared-typed-arrays';
 import type { BaseKeyProvider } from './KeyProvider';
 import { E2EE_FLAG, E2EE_TRACK_ID } from './constants';
@@ -97,6 +105,7 @@ export class E2EEManager
     return {
       room: this.room?.name,
       participant: this.room?.localParticipant.identity,
+      [LOG_OWNER]: this.room ? telemetry.carrier(telemetry.scopeOf(this.room)) : undefined,
     };
   }
 

src/index.ts

@@ -201,3 +201,12 @@ export {
   type SerializerOutput,
   serializers,
 } from './utils/serializer';
+
+export { disableTelemetry } from './telemetry';
+// @internal: the one seam a non-browser host (React Native) wires; not public API.
+export {
+  configureTelemetryHost,
+  type TelemetryHost,
+  type TelemetryStorage,
+  type DeviceState as TelemetryDeviceState,
+} from './telemetry';

src/logger.ts

@@ -49,6 +49,80 @@ livekitLogger.setDefaultLevel(LogLevel.info);
 
 export default livekitLogger as StructuredLogger;
 
+/**
+ * @internal Receives every warn/error line of the SDK's loggers whatever their console level
+ * (telemetry). The console gets what it got before; the devtools call site becomes the forwarder.
+ */
+export type LogCapture = (
+  level: LogLevel,
+  msg: unknown,
+  context?: object,
+  loggerName?: string,
+) => void;
+
+let logCapture: LogCapture | undefined;
+
+/**
+ * @internal The object a line is about (a Room's telemetry scope), carried through the structured
+ * context without being enumerable: the console and log extensions see the context as before.
+ */
+export const LOG_OWNER: unique symbol = Symbol('livekit.logOwner');
+
+/** @internal */
+export function setLogCapture(capture?: LogCapture) {
+  logCapture = capture;
+}
+
+const captured = new WeakSet<log.Logger>();
+
+type CapturedMethod = ((...args: unknown[]) => void) & { lkForwards?: boolean };
+
+function forwarder(logger: log.Logger, method: 'warn' | 'error'): CapturedMethod {
+  const real = logger[method] as CapturedMethod;
+  if (real.lkForwards) return real;
+  const name = (logger as { name?: string }).name;
+  const forward: CapturedMethod = (...args) => {
+    try {
+      const context = typeof args[1] === 'object' && args[1] !== null ? args[1] : undefined;
+      // Whatever path a context took here, the console never lists the owner.
+      if (context && Object.getOwnPropertyDescriptor(context, LOG_OWNER)?.enumerable) {
+        Object.defineProperty(context, LOG_OWNER, { enumerable: false });
+      }
+      logCapture?.(LogLevel[method], args[0], context, name);
+    } catch {
+      // telemetry never breaks logging
+    }
+    real(...args); // the original call, argument for argument
+  };
+  forward.lkForwards = true;
+  return forward;
+}
+
+// loglevel assigns `warn`/`error` anew on every `setLevel` (a noop below the level), so the
+// forwarder is re-applied after each `setLevel` and runs before whatever loglevel assigned.
+function captureWarnings(logger: log.Logger) {
+  if (captured.has(logger)) return;
+  captured.add(logger);
+  const wrap = () => {
+    logger.warn = forwarder(logger, 'warn');
+    logger.error = forwarder(logger, 'error');
+  };
+  // `setLevel`, `resetLevel` and `rebuild` all reassign the methods.
+  const target = logger as unknown as Record<string, (...args: unknown[]) => unknown>;
+  for (const name of ['setLevel', 'resetLevel', 'rebuild']) {
+    const original = target[name];
+    if (typeof original !== 'function') continue;
+    target[name] = (...args: unknown[]) => {
+      const result = original.apply(logger, args);
+      wrap();
+      return result;
+    };
+  }
+  wrap();
+}
+
+livekitLoggers.forEach(captureWarnings);
+
 /**
  * @internal
  *
@@ -64,6 +138,7 @@ export default livekitLogger as StructuredLogger;
 export function getLogger(name: string, ctxFn?: ContextProvider) {
   const logger = log.getLogger(name);
   logger.setDefaultLevel(livekitLogger.getLevel());
+  captureWarnings(logger);
   if (!ctxFn) {
     return logger as StructuredLogger;
   }
@@ -78,6 +153,9 @@ function wrapWithContext(base: StructuredLogger, ctxFn: ContextProvider): Struct
   const wrap = (method: LogMethod) => (msg: string, extra?: object) => {
     const ctx = ctxFn();
     const merged = ctx || extra ? { ...ctx, ...extra } : undefined;
+    const owner = (ctx as Record<symbol, unknown> | undefined)?.[LOG_OWNER];
+    if (merged && owner)
+      Object.defineProperty(merged, LOG_OWNER, { value: owner, enumerable: false });
     base[method](msg, merged);
   };
 

src/room/RTCEngine.ts

@@ -60,7 +60,7 @@ import {
   isFrameMetadataSupported,
   shouldUseFrameMetadataScriptTransform,
 } from '../frameMetadata/utils';
-import log, { LoggerNames, getLogger } from '../logger';
+import log, { LOG_OWNER, LoggerNames, getLogger } from '../logger';
 import type { InternalRoomOptions } from '../options';
 import type { NonSharedUint8Array } from '../type-polyfills/non-shared-typed-arrays';
 import TypedPromise from '../utils/TypedPromise';
@@ -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;
@@ -328,12 +331,16 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
   }
 
   /** @internal */
+  /** @internal The Room this engine serves (looked up lazily: the engine exists before the Room's scope), for telemetry to file its warnings by identity. */
+  logOwner?: () => object | undefined;
+
   get logContext() {
     return {
       room: this.latestJoinResponse?.room?.name,
       roomID: this.latestJoinResponse?.room?.sid,
       participant: this.latestJoinResponse?.participant?.identity,
       participantID: this.participantSid,
+      [LOG_OWNER]: this.logOwner?.(),
     };
   }
 
@@ -737,6 +744,10 @@ export default class RTCEngine extends (EventEmitter as new () => TypedEventEmit
       this.emit(EngineEvent.TokenRefreshed, token);
     };
 
+    this.client.onWebSocketOpen = () => {
+      this.emit(EngineEvent.SignalOpened);
+    };
+
     this.client.onRemoteMuteChanged = (trackSid: string, muted: boolean) => {
       this.emit(EngineEvent.RemoteMute, trackSid, muted);
     };
@@ -1309,6 +1320,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');
@@ -1406,7 +1418,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();
@@ -1479,7 +1491,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();
@@ -2070,12 +2082,13 @@ 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;
+  signalOpened: () => void;
   closing: () => void;
   mediaTrackAdded: (
     track: MediaStreamTrack,