feat(telemetry): add the pipeline, sessions, spans and exporter

.changeset/add-livekit-telemetry-crate.md

@@ -0,0 +1,5 @@
+---
+livekit-telemetry: minor
+---
+
+Add `livekit-telemetry`, the shared client telemetry core: records (events, warn/error logs, spans, RTC stats windows) are buffered on the device, committed to a write-ahead cache (memory, or fsynced files with `storage_dir`) and exported as OTLP/HTTP logs and traces to the LiveKit Cloud project each Room belongs to. A Room hands over only its server URL and token (`Scope::set_server`, again on every refresh); the core validates the URL, reads the token's grant and expiry, and uploads each record with the token of the Room and project that captured it. Every collector answer is classified (partial success, 4xx, 413 split, 401/403 disabled vs credential, 404, 429/503 `Retry-After`/`RetryInfo`, 5xx, no answer) with jittered backoff per destination, soft and hard device holds, loss counters by reason (`lk.telemetry.report`, `Telemetry::stats`), Room-scoped custom events and correlation attributes, and an opt-out that purges everything unsent. Defaults export and window RTC stats once a minute; `LK_TELEMETRY_ENDPOINT` points uploads at a local collector for tests.

Cargo.toml

@@ -11,6 +11,7 @@ members = [
     "livekit-ffi",
     "livekit-uniffi",
     "livekit-datatrack",
+    "livekit-telemetry",
     "livekit-token",
     "livekit-token-source",
     "livekit-ffi-node-bindings",
@@ -60,6 +61,7 @@ livekit-api = { version = "0.8.1", path = "livekit-api" }
 livekit-capture = { version = "0.1.3", path = "livekit-capture" }
 livekit-ffi = { version = "0.12.81", path = "livekit-ffi" }
 livekit-datatrack = { version = "0.2.1", path = "livekit-datatrack" }
+livekit-telemetry = { version = "0.1.0", path = "livekit-telemetry" }
 livekit-signaling = { version = "0.1.4", path = "livekit-signaling" }
 livekit-token = { version = "0.2.1", path = "livekit-token" }
 livekit-token-source = { version = "0.1.3", path = "livekit-token-source" }

knope.toml

@@ -178,6 +178,14 @@ versioned_files = [
 ]
 changelog = "livekit-datatrack/CHANGELOG.md"
 
+[packages.livekit-telemetry]
+versioned_files = [
+  "livekit-telemetry/Cargo.toml",
+  "Cargo.lock",
+  { path = "Cargo.toml", dependency = "livekit-telemetry" },
+]
+changelog = "livekit-telemetry/CHANGELOG.md"
+
 [packages.livekit-common]
 versioned_files = [
   "livekit-common/Cargo.toml",

livekit-telemetry/CHANGELOG.md

@@ -0,0 +1 @@
+# Changelog

livekit-telemetry/Cargo.toml

@@ -0,0 +1,44 @@
+[package]
+name = "livekit-telemetry"
+description = "Client telemetry core for LiveKit: buffers events on-device and exports them as OTLP"
+version = "0.1.0"
+readme = "README.md"
+license.workspace = true
+edition.workspace = true
+repository.workspace = true
+
+[dependencies]
+tokio = { workspace = true, default-features = false, features = ["macros", "sync", "time"] }
+log = { workspace = true }
+thiserror = { workspace = true }
+async-trait = "0.1"
+prost = { workspace = true }
+# `google.protobuf.Any`/`Duration` for the `google.rpc.Status` error body (RetryInfo).
+prost-types = { workspace = true }
+rand = { workspace = true }
+# Unverified JWT claims (observability grant, expiry); both already linked by livekit-api.
+base64 = "0.22"
+serde_json = { workspace = true }
+# Server URLs are parsed, never string-matched, before a token may follow them.
+url = "2.3"
+# gzip for request bodies; the deflate backend (miniz_oxide) is already linked by data streams.
+flate2 = "1"
+# OTLP message types only (`gen-tonic-messages` = prost structs, no tonic). Same prost as
+# livekit-protocol so a single prost is linked.
+opentelemetry-proto = { version = "0.32", default-features = false, features = ["logs", "trace", "gen-tonic-messages"] }
+livekit-net = { workspace = true, optional = true }
+uniffi = { workspace = true, features = ["scaffolding-ffi-buffer-fns"], optional = true }
+
+[features]
+# Default HTTP transport over the pluggable `livekit-net` client (native backend, or one
+# the host registered with `livekit_net::set_http_client`).
+net = ["dep:livekit-net"]
+uniffi = ["dep:uniffi"]
+
+[dev-dependencies]
+tokio = { workspace = true, default-features = false, features = ["rt", "rt-multi-thread", "macros", "sync", "time", "test-util", "net", "io-util"] }
+
+# How CI checks this crate's features, read by
+# `.github/workflows/feature-combinations-curated.yml` via `cargo metadata`.
+[package.metadata.feature-combinations]
+mode = "powerset"

livekit-telemetry/README.md

@@ -0,0 +1,81 @@
+# LiveKit Telemetry
+
+**Important**:
+This is an internal crate that powers client telemetry in LiveKit client SDKs (through
+`livekit-uniffi`: Swift, Kotlin, Dart) and is not meant to be used directly.
+
+The core buffers records on the device and ships them out-of-band as standard
+[OTLP/HTTP](https://opentelemetry.io/docs/specs/otlp/) logs and traces to the LiveKit Cloud
+project a room belongs to. Everything hard lives here once — destination and credentials,
+batching, encoding, retry, persistence, holds, loss accounting — while platforms provide only
+what they are uniquely placed to: OS signals (instruments) and the byte-moving transport.
+
+```text
+ SDK / instruments ──emit()──▶ Telemetry ──▶ Store ──drain──▶ Exporter ──push──▶ BatchCache ──upload oldest-first──▶ TelemetryTransport
+                              (source)      (records)         (tick · encode)   (MemoryCache | FileCache)            (NetTransport | host HTTP)
+```
+
+| Layer | OTel equivalent | Type |
+|---|---|---|
+| source | `Logger.emit`, `Tracer.start` | [`Telemetry`], [`Scope`] (one per room), [`Span`] |
+| store | `BatchLogRecordProcessor` queue | `Store` (in-memory, drop-oldest, bounded) |
+| exporter | batch processor timer + exporter | [`Exporter`] (actor, spawn `run()`) |
+| cache | disk buffering | [`BatchCache`]: [`MemoryCache`], [`FileCache`] (`storage_dir`) |
+| transport | exporter's HTTP client | [`TelemetryTransport`]; `NetTransport` over `livekit-net` (feature `net`) |
+
+## Usage
+
+```rust
+# use std::sync::Arc;
+# use livekit_telemetry::*;
+# struct Discard;
+# #[async_trait::async_trait]
+# impl TelemetryTransport for Discard {
+#     async fn send(&self, _: ExportRequest) -> Result<ExportResponse, ExportError> {
+#         Ok(ExportResponse::accepted())
+#     }
+# }
+# #[tokio::main(flavor = "current_thread")] async fn main() {
+let config = TelemetryConfig { storage_dir: None, ..Default::default() };
+let (telemetry, exporter) = Telemetry::new(config, Arc::new(Discard));
+tokio::spawn(exporter.run());
+
+// One scope per room. Its server URL and token decide where its records go; hand the token
+// over again on every refresh.
+let room = telemetry.begin_scope();
+room.set_server("wss://my-project.livekit.cloud", "<participant token>");
+room.set_attribute("app.call_id", Some("c-42".into()));
+room.emit_custom("checkout", vec![Attribute::new("plan", "pro")]);
+
+let connect = room.start(SpanName::Connect, None);
+connect.step(SpanStep::WsOpen);
+connect.end(SpanOutcome::Ok, None);
+
+telemetry.set_device_state(DeviceState { thermal: ThermalState::Serious, ..Default::default() });
+telemetry.shutdown().await; // cache, then upload what the network allows
+println!("{}", telemetry.stats()); // drops by reason, uploads, cached batches
+# }
+```
+
+Event names, attributes, cadences and the upload policy are defined in [`SPEC.md`](SPEC.md).
+
+## Design notes
+
+- **The room decides the destination.** A platform passes only the server URL and the token
+  ([`Scope::set_server`]). The core derives the ingest URL (LiveKit Cloud hosts only), reads
+  the token's unverified claims (observability grant, expiry) and routes each batch by its
+  owner — project and session — so two rooms on two projects never share a token.
+- **Write-ahead cache.** Every batch is encoded, gzipped and cached *before* the network is
+  tried, then uploaded oldest-first and removed on an answer. Tokens are attached at upload
+  time and never persisted.
+- **The core reads every answer.** Transports return status, headers and body; the core alone
+  classifies OTLP partial success, 4xx, 413 (split), 429/503 `Retry-After` and `RetryInfo`,
+  and Cloud's disabled and credential answers.
+- **Loss is counted, never silent.** Every way a record can be lost has a counter; the deltas
+  ride along with the next batch as `lk.telemetry.report` and are readable locally through
+  [`Telemetry::stats`].
+- **Device state comes from the host; the policy lives here.** Thermal, power, memory,
+  network and lifecycle are OS APIs the host watches and pushes through
+  [`Telemetry::set_device_state`]; the core stretches its cadence and holds uploads.
+- **Size.** OTLP types come from `opentelemetry-proto` (`gen-tonic-messages`, no tonic); JWT
+  claims are read with `base64` and `serde_json`, both already linked by the UniFFI library.