From d28248283e7cbdd2d1eb677d932378f30f8d7419 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?= <86720177+pblazej@users.noreply.github.com> Date: Thu, 1 Oct 2026 12:41:00 +0200 Subject: [PATCH 1/4] feat(uniffi): expose the telemetry core `telemetry_configure` installs the process pipeline with a host-implemented `TelemetryTransport` or the `livekit-net` HTTP client, and `telemetry_configure_pulled` hands bindings with thread-bound callbacks (Dart) a bounded pull queue instead. `telemetry_scope` gives each Room its `TelemetryScope`: `set_server` at connect and on every token refresh, spans, the subscribe lifecycle, `record_peer_stats` paced by `stats_poll_interval_ms`, `track_ended`, `disconnected`, and the app API (`emit_custom`, `set_attribute`). `telemetry_disable` is the synchronous opt-out and `telemetry_is_disabled` reads it process-wide; device state, device events and log records come from the host. Dart tests cover the pull queue, and macOS links reserve header padding for Flutter's install name rewrite. The module is a core module of `livekit-ffi` (feature `core-modules`), the crate `livekit-uniffi` merged into. --- .cargo/config.toml | 6 +- .changeset/uniffi-telemetry.md | 5 + .gitignore | 1 + Cargo.lock | 1 + livekit-ffi/Cargo.toml | 4 + livekit-ffi/src/core_modules/mod.rs | 3 + livekit-ffi/src/core_modules/telemetry.rs | 801 ++++++++++++++++++ .../support/dart/test/telemetry_test.dart | 82 ++ livekit-telemetry/SPEC.md | 2 +- livekit-telemetry/uniffi.toml | 2 +- 10 files changed, 903 insertions(+), 4 deletions(-) create mode 100644 .changeset/uniffi-telemetry.md create mode 100644 livekit-ffi/src/core_modules/telemetry.rs create mode 100644 livekit-ffi/support/dart/test/telemetry_test.dart diff --git a/.cargo/config.toml b/.cargo/config.toml index 2adead06d..122b31359 100644 --- a/.cargo/config.toml +++ b/.cargo/config.toml @@ -8,16 +8,18 @@ rustflags = ["-C", "target-feature=+crt-static"] [target.aarch64-pc-windows-msvc] rustflags = ["-C", "target-feature=+crt-static"] +# `-headerpad_max_install_names` on macOS: Flutter's native-assets step rewrites the cdylib's +# install name with install_name_tool, which fails without the padding. # `-ObjC` must also be passed to rustdoc: doctest binaries are linked by rustdoc # and do not inherit `rustflags`. Without it the ObjC categories in libwebrtc's # static lib (e.g. `NSString (StdString)`) are not loaded, and static # initializers such as RTCH264ProfileLevelId.mm abort at process startup. [target.x86_64-apple-darwin] -rustflags = ["-C", "link-args=-ObjC"] +rustflags = ["-C", "link-args=-ObjC -Wl,-headerpad_max_install_names"] rustdocflags = ["-C", "link-args=-ObjC"] [target.aarch64-apple-darwin] -rustflags = ["-C", "link-args=-ObjC"] +rustflags = ["-C", "link-args=-ObjC -Wl,-headerpad_max_install_names"] rustdocflags = ["-C", "link-args=-ObjC"] [target.aarch64-apple-ios] diff --git a/.changeset/uniffi-telemetry.md b/.changeset/uniffi-telemetry.md new file mode 100644 index 000000000..f1abfab0c --- /dev/null +++ b/.changeset/uniffi-telemetry.md @@ -0,0 +1,5 @@ +--- +livekit-ffi: minor +--- + +Expose the client telemetry core over UniFFI: `telemetry_configure` / `telemetry_configure_pulled` (a bounded pull queue for bindings with thread-bound callbacks, served with `next` or polled with `try_next`), `telemetry_scope` with `TelemetryScope.set_server` (connect and every token refresh), spans, the subscribe lifecycle, `record_peer_stats` + `stats_poll_interval_ms`, `track_ended`, `emit_custom` / `set_attribute` (the Room-scoped app API), `telemetry_disable` (a synchronous opt-out, in effect when it returns), device state and log forwarding, with a host-implemented `TelemetryTransport` or the `livekit-net` HTTP client. diff --git a/.gitignore b/.gitignore index 328df4d65..fbae334f9 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ soxr-sys/test-output.wav .env .cursor __pycache__ +/target-*/ diff --git a/Cargo.lock b/Cargo.lock index 59651eecf..11ff39475 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3919,6 +3919,7 @@ dependencies = [ "livekit-datatrack", "livekit-net", "livekit-protocol", + "livekit-telemetry", "livekit-token", "log", "once_cell", diff --git a/livekit-ffi/Cargo.toml b/livekit-ffi/Cargo.toml index deb3d4a68..bb6cc321a 100644 --- a/livekit-ffi/Cargo.toml +++ b/livekit-ffi/Cargo.toml @@ -45,6 +45,7 @@ core-modules = [ "dep:livekit-data-stream", "dep:livekit-datatrack", "dep:livekit-net", + "dep:livekit-telemetry", "dep:livekit-token", "dep:prost", "dep:async-trait", @@ -53,6 +54,8 @@ core-modules = [ "livekit-common/uniffi", "livekit-datatrack/uniffi", "livekit-net/uniffi", + "livekit-telemetry/net", + "livekit-telemetry/uniffi", "uniffi/tokio", ] @@ -97,6 +100,7 @@ livekit-common = { workspace = true, optional = true } livekit-token = { workspace = true, optional = true } livekit-datatrack = { workspace = true, optional = true } livekit-net = { workspace = true, optional = true } +livekit-telemetry = { workspace = true, optional = true } livekit-data-stream = { workspace = true, optional = true } tokio-util = { version = "0.7.18", optional = true } once_cell = { version = "1.21.3", optional = true } diff --git a/livekit-ffi/src/core_modules/mod.rs b/livekit-ffi/src/core_modules/mod.rs index a0b7cf857..9cb8b327f 100644 --- a/livekit-ffi/src/core_modules/mod.rs +++ b/livekit-ffi/src/core_modules/mod.rs @@ -32,6 +32,9 @@ pub mod data_stream; /// Access token generation and verification from [`livekit-api::access_token`]. pub mod access_token; +/// Client telemetry core from [`livekit-telemetry`]. +pub mod telemetry; + /// Forward log messages from Rust. pub mod log_forward; diff --git a/livekit-ffi/src/core_modules/telemetry.rs b/livekit-ffi/src/core_modules/telemetry.rs new file mode 100644 index 000000000..30cdab7be --- /dev/null +++ b/livekit-ffi/src/core_modules/telemetry.rs @@ -0,0 +1,801 @@ +//! Client telemetry core from the [`livekit-telemetry`] crate. +//! +//! What a platform calls, and when: +//! +//! - SDK init: [`telemetry_configure`] (or [`telemetry_configure_pulled`] where foreign callbacks +//! are thread-bound) with storage and instruments. No endpoint, no headers: destinations come +//! from the rooms. +//! - Room created: [`telemetry_scope`]; connect: [`TelemetryScope::set_server`] with the server +//! URL and the participant token, and again with every refreshed token. +//! - Room life: spans ([`TelemetryScope::start`]), subscribe lifecycle, one `getStats()` report +//! per peer connection ([`TelemetryScope::record_peer_stats`]) every +//! [`TelemetryScope::stats_poll_interval_ms`], [`TelemetryScope::disconnected`]. +//! - App API: [`TelemetryScope::emit_custom`], [`TelemetryScope::set_attribute`] (room-scoped), +//! [`telemetry_disable`] (process-wide opt-out). +//! - OS signals: [`telemetry_set_device_state`], [`telemetry_device_event`], [`telemetry_log`]. +//! +//! Transport: pass a host-implemented `TelemetryTransport`, or `None` to ride the HTTP client +//! the host registered with `livekit-net` (`set_http_client`) for signaling. + +use std::{ + collections::HashMap, + sync::{ + atomic::{AtomicBool, AtomicU64, Ordering}, + Arc, Mutex, + }, +}; + +use livekit_telemetry::{ + global::{self, TelemetryInstrument}, + Attribute, AttributeValue, DeviceEvent, DeviceState, DisconnectReason, ExportError, + ExportRequest, ExportResponse, LogRecord, NetTransport, ReconnectReason, RoomIdentity, RtcStat, + SpanName, SpanOutcome, SpanStep, SpanTrack, StreamDirection, TelemetryConfig, TelemetryEvent, + TelemetryStats, TelemetryTransport, TraceContext, TrackKind, +}; +use tokio::sync::{mpsc, oneshot, watch}; + +/// Why a [`Telemetry`] pipeline could not be created. +#[derive(uniffi::Error, thiserror::Error, Debug)] +#[uniffi(flat_error)] +pub enum TelemetryError { + /// No transport was passed and no HTTP client is registered with `livekit-net`. + #[error("no telemetry transport: pass one, or register an HTTP client with livekit-net first")] + NoTransport, +} + +/// Start the process pipeline; a previous one is drained and replaced. `transport = None` uses +/// the HTTP client registered with `livekit-net`, if any. After [`telemetry_disable`] nothing +/// starts, and whatever `config.storage_dir` still holds is deleted. +#[uniffi::export] +pub fn telemetry_configure( + config: TelemetryConfig, + transport: Option>, + instruments: Vec>, +) -> Result<(), TelemetryError> { + let transport: Arc = match transport { + Some(transport) => transport, + None => Arc::new(NetTransport::from_registry().ok_or(TelemetryError::NoTransport)?), + }; + install(livekit_telemetry::Telemetry::new(config, transport), instruments); + Ok(()) +} + +/// Like [`telemetry_configure`], exporting through a queue the host drains from its own thread. +/// For bindings whose callbacks cannot be invoked from Rust threads (uniffi-dart today): such a +/// host passes no `instruments` — it starts and stops its own around configure and disable, so +/// Rust never calls back into it — and polls the queue with +/// [`try_next`](TelemetryExportQueue::try_next). +#[uniffi::export] +pub fn telemetry_configure_pulled( + config: TelemetryConfig, + instruments: Vec>, +) -> Arc { + let queue = TelemetryExportQueue::new(); + queue.process.store(true, Ordering::SeqCst); + install(livekit_telemetry::Telemetry::new(config, queue.clone()), instruments); + queue +} + +fn install( + (telemetry, exporter): (livekit_telemetry::Telemetry, livekit_telemetry::Exporter), + instruments: Vec>, +) { + let previous = global::install(telemetry, instruments); + // Refused after the opt-out (the new pipeline was purged instead, its `storage_dir` with it), + // or opted out meanwhile (every generation revoked): nothing to run, nothing to drain. + if global::is_disabled() { + return; + } + crate::runtime::runtime().spawn(exporter.run()); + if let Some(previous) = previous { + crate::runtime::runtime().spawn(async move { previous.shutdown().await }); + } +} + +/// Drain and stop the process pipeline; every call is a no-op afterwards. After +/// [`telemetry_disable`], returns once its purge is complete. +#[uniffi::export(async_runtime = "tokio")] +pub async fn telemetry_shutdown() { + purged().await; + global::shutdown().await; +} + +/// Opt-out for the rest of the process, in effect when this returns: no scope is handed out, +/// every existing scope and instrument captures nothing, and later configures are refused. +/// Everything not yet sent — queued, open, cached on disk — is then deleted in the background +/// without another upload (awaited by [`telemetry_shutdown`] and [`telemetry_flush`]). Data +/// already sent cannot be recalled. +/// +/// The installed instruments' `stop()` runs synchronously on the calling thread, under the +/// lifecycle lock, before this returns: it must not block on the caller's queue and must not +/// call back into configure, disable, flush or shutdown. +#[uniffi::export] +pub fn telemetry_disable() { + start_purge(&PURGE, global::disable()); +} + +/// Whether the process opted out with [`telemetry_disable`]: `true` for every thread and isolate +/// from the moment that call revokes collection, before it returns. A cheap read that never +/// panics; platforms with per-isolate state may check it right before each `getStats` or submit. +#[uniffi::export] +pub fn telemetry_is_disabled() -> bool { + global::is_disabled() +} + +/// The opt-out's purge in progress: done once it holds `true`. +type PurgeSlot = Mutex>>; + +/// The process-wide opt-out's purge. +static PURGE: PurgeSlot = Mutex::new(None); + +/// Run `purge` in the background and record it in `slot`, done only once every purge recorded +/// there before it is done too: a repeated opt-out never lets a waiter skip the first purge. +fn start_purge(slot: &PurgeSlot, purge: impl std::future::Future + Send + 'static) { + let (done, receiver) = watch::channel(false); + let previous = slot.lock().unwrap_or_else(|e| e.into_inner()).replace(receiver); + crate::runtime::runtime().spawn(async move { + purge.await; + if let Some(mut previous) = previous { + let _ = previous.wait_for(|done| *done).await; + } + let _ = done.send(true); + }); +} + +/// Wait for the opt-out's purge, if one was started. +async fn purged() { + wait_purge(&PURGE).await; +} + +/// Wait for the purge recorded in `slot`, if any. +async fn wait_purge(slot: &PurgeSlot) { + let receiver = slot.lock().unwrap_or_else(|e| e.into_inner()).clone(); + if let Some(mut receiver) = receiver { + let _ = receiver.wait_for(|done| *done).await; + } +} + +/// Cache everything queued and upload what the network allows. After [`telemetry_disable`], +/// returns once its purge is complete. +#[uniffi::export(async_runtime = "tokio")] +pub async fn telemetry_flush() { + purged().await; + global::flush().await; +} + +/// A scope — one room, one call — on the process pipeline; `None` while telemetry is off. +#[uniffi::export] +pub fn telemetry_scope() -> Option> { + global::scope().map(|scope| Arc::new(TelemetryScope(scope))) +} + +/// A process-level event (outside any room). +#[uniffi::export] +pub fn telemetry_emit(event: TelemetryEvent) { + global::emit(event); +} + +/// A captured log line; the core applies the per-source floor and builds the record. For the +/// platform's own and WebRTC's lines only — never a Rust entry received through the log +/// forwarder, which the core has already copied (`log_forward_bootstrap`). +#[uniffi::export] +pub fn telemetry_log(record: LogRecord) { + global::log(record); +} + +/// Audio route, interruption, denied permission: a process-level record built by the core. +#[uniffi::export] +pub fn telemetry_device_event(event: DeviceEvent) { + global::device_event(event); +} + +/// The device's current state; drives the upload cadence and yields change events. +#[uniffi::export] +pub fn telemetry_set_device_state(state: DeviceState) { + global::set_device_state(state); +} + +/// SDK metadata added at export to every record of every scope (not an app API: app +/// attributes are room-scoped, [`TelemetryScope::set_attribute`]); `None` removes it. +#[uniffi::export] +pub fn telemetry_set_attribute(key: String, value: Option) { + global::set_attribute(&key, value); +} + +/// Pipeline health (drops by reason, uploads, backlog); `None` while telemetry is off. +#[uniffi::export] +pub fn telemetry_stats() -> Option { + global::stats() +} + +/// The stats as one line for a debug console, or `off`. +#[uniffi::export] +pub fn telemetry_diagnostics() -> String { + global::diagnostics() +} + +/// One Room's telemetry session: its trace, attributes, spans, stats and destination. +#[derive(uniffi::Object)] +pub struct TelemetryScope(livekit_telemetry::Scope); + +#[uniffi::export] +impl TelemetryScope { + /// The session's trace id as 32 hex characters. + pub fn trace_id(&self) -> String { + self.0.trace_id() + } + + /// An SDK event filed under this session. + pub fn emit(&self, event: TelemetryEvent) { + self.0.emit(event); + } + + /// The room's server URL and participant token: at connect and with every refreshed token. + /// Cheap and idempotent. The core derives the ingest URL (LiveKit Cloud only), reads the + /// grant and expiry, and uploads this room's records with this room's token. + pub fn set_server(&self, url: String, token: String) { + self.0.set_server(&url, &token); + } + + /// An app event, exported as `custom.` with the room's correlation attributes. Names + /// and keys up to 128 bytes, values up to 1024, at most 64 attributes, none of the SDK's + /// keys (`lk.*`, `otel.*`, `code.*`, `session.id`, `error.type`); anything else is rejected + /// and counted. + pub fn emit_custom(&self, name: String, attributes: HashMap) { + let attributes = attributes.into_iter().map(|(k, v)| Attribute::new(k, v)).collect(); + self.0.emit_custom(&name, attributes); + } + + /// An app correlation attribute on every record this room captures from now on; `None` + /// removes it. Same limits as `emit_custom`, at most 64 per room. + pub fn set_attribute(&self, key: String, value: Option) { + self.0.set_attribute(&key, value.map(AttributeValue::Str)); + } + + /// One peer connection's whole `getStats()` report and its tracks (MediaStreamTrack id → + /// track sid); the core maps every RTP stream to its track. + pub fn record_peer_stats( + &self, + report: Vec, + tracks: HashMap, + timestamp_ns: Option, + ) { + self.0.record_peer_stats(report, tracks, timestamp_ns); + } + + /// When to poll `getStats()` next, in milliseconds; ask after every poll. + pub fn stats_poll_interval_ms(&self) -> u64 { + self.0.stats_poll_interval_ms() + } + + /// Start a typed span in this session's trace, stamped now; `parent` nests it. + pub fn start(&self, name: SpanName, parent: Option>) -> Arc { + Arc::new(TelemetrySpan(self.0.start(name, parent.map(|p| p.0.clone())))) + } + + /// The room and local participant, on every record this session captures from now on. + pub fn set_room(&self, room: RoomIdentity) { + self.0.set_room(room); + } + + /// The session ended for good (not a reconnect). + pub fn disconnected(&self, reason: DisconnectReason) { + self.0.disconnected(reason); + } + + /// A log record filed under this session even without an ambient span. + pub fn log(&self, record: LogRecord) { + self.0.log(record); + } + + /// One track's whole `getStats()` report; the core maps it (see `record_stats`). + pub fn record_stats_report( + &self, + track_sid: String, + kind: TrackKind, + direction: StreamDirection, + report: Vec, + timestamp_ns: Option, + ) { + self.0.record_stats_report(&track_sid, kind, direction, report, timestamp_ns); + } + + /// Intent to subscribe (autoSubscribe: the remote publish; manual: the call; tracks already + /// in the room at join: at connect): opens `lk.subscribe`. + pub fn subscribe_started(&self, track: SpanTrack) { + self.0.subscribe_started(track); + } + + /// The server confirmed the subscription. Without an earlier `subscribe_started` this is the + /// intent (a fallback, measured from here): `lk.subscribe` opens and `stats_poll_interval_ms` + /// turns fast at once. + pub fn subscribed(&self, track: SpanTrack) { + self.0.subscribed(track); + } + + /// A track left the room (unpublished, unsubscribed, publisher gone): a pending subscribe + /// ends, the track's last RTC window ships, its state is forgotten. + pub fn track_ended(&self, sid: String) { + self.0.track_ended(&sid); + } + + /// The subscription failed; `error_type` is the platform's error name. + pub fn subscribe_failed(&self, sid: String, error_type: String) { + self.0.subscribe_failed(&sid, &error_type); + } +} + +/// The protocol's `DisconnectReason` number as the shared enum. +#[uniffi::export] +pub fn telemetry_disconnect_reason(proto: i32) -> DisconnectReason { + DisconnectReason::from_proto(proto) +} + +/// The protocol's `ReconnectReason` number (`RR_*`) as the shared enum. +#[uniffi::export] +pub fn telemetry_reconnect_reason(proto: i32) -> ReconnectReason { + ReconnectReason::from_proto(proto) +} + +/// One export the host has to perform on behalf of a pulled pipeline. +#[derive(uniffi::Record)] +pub struct PendingExport { + pub id: u64, + pub request: ExportRequest, +} + +struct Pending { + export: PendingExport, + done: oneshot::Sender>, +} + +/// Pull-side transport: Rust never calls into the host. The exporter queues each request; the +/// host awaits [`next`](Self::next) (a Rust future — those cross every binding) or polls +/// [`try_next`](Self::try_next), performs the HTTP call on its own thread, and reports the +/// outcome with [`complete`](Self::complete), +/// which unblocks the exporter's retry/drop/go-silent logic exactly as a direct transport would. +/// +/// Exists because uniffi-dart's foreign-trait callbacks are isolate-bound (`Pointer.fromFunction`) +/// and abort the VM when invoked from a tokio thread; Swift and Kotlin callbacks are thread-agnostic +/// and use [`TelemetryTransport`] directly. +/// One attempt at an SDK operation, owned by the core. Every call is synchronous and stamps the +/// clock inside, so the only skew is the FFI call; `describe()` is the console line on every +/// platform. Detached (no session) it still times and describes itself. +#[derive(uniffi::Object)] +pub struct TelemetrySpan(Arc); + +#[uniffi::export] +impl TelemetrySpan { + /// A checkpoint, stamped now. + pub fn step(&self, step: SpanStep) { + self.0.step(step); + } + + /// The open bag; replaces an existing key. + pub fn set_attribute(&self, key: String, value: AttributeValue) { + self.0.set_attribute(key, value); + } + + /// The track the operation is about. + pub fn set_track(&self, track: SpanTrack) { + self.0.set_track(track); + } + + /// End once; `error` becomes `error.type` when the outcome is `Error`. + pub fn end(&self, outcome: SpanOutcome, error: Option) { + self.0.end(outcome, error); + } + + /// End with an error; `error` becomes `error.type`. + pub fn fail(&self, error: String) { + self.0.fail(error); + } + + /// End as cancelled. + pub fn cancel(&self) { + self.0.cancel(); + } + + /// Whether the span has ended. + pub fn is_ended(&self) -> bool { + self.0.is_ended() + } + + /// `None` for a detached span. + pub fn context(&self) -> Option { + self.0.context() + } + + /// Seconds to the end, or to the last step while running. + pub fn total_secs(&self) -> f64 { + self.0.total_secs() + } + + /// `lk.connect: ws_open +1.49s, …, total 1.83s, ok` + pub fn describe(&self) -> String { + self.0.describe() + } +} + +/// The pull queue's serving side (see its type docs above [`TelemetrySpan`]). +#[derive(uniffi::Object)] +pub struct TelemetryExportQueue { + /// Bounded to one request: the exporter has one request out at a time, so nothing piles up + /// behind a host that stopped serving. `None` once finished. + tx: Mutex>>, + rx: tokio::sync::Mutex>, + inflight: Mutex>>>, + seq: AtomicU64, + finished: AtomicBool, + /// Serves the process pipeline (`telemetry_configure_pulled`): nothing is handed out once + /// the app opted out, even before the purge has cancelled the request. + process: AtomicBool, + /// Requests handed to the queue so far (tests only). + #[cfg(test)] + queued: AtomicU64, +} + +#[uniffi::export(async_runtime = "tokio")] +impl TelemetryExportQueue { + /// An empty queue; `telemetry_configure_pulled` makes the one it serves. + #[uniffi::constructor] + pub fn new() -> Arc { + let (tx, rx) = mpsc::channel(1); + Arc::new(Self { + tx: Mutex::new(Some(tx)), + rx: tokio::sync::Mutex::new(rx), + inflight: Mutex::new(HashMap::new()), + seq: AtomicU64::new(0), + finished: AtomicBool::new(false), + process: AtomicBool::new(false), + #[cfg(test)] + queued: AtomicU64::new(0), + }) + } + + /// End the serving loop: `next` resolves `None` from now on, and whatever is still queued is + /// discarded, never served. Call after `telemetry_shutdown` or `telemetry_disable`, or when a + /// later `telemetry_configure_pulled` replaced this queue. + /// (Not `close`: UniFFI's Kotlin objects already have `AutoCloseable.close`.) + pub fn finish(&self) { + self.finished.store(true, Ordering::SeqCst); + self.tx.lock().unwrap_or_else(|e| e.into_inner()).take(); + self.inflight.lock().unwrap_or_else(|e| e.into_inner()).clear(); + // Dropping what is queued tells its exporter the host dropped it. + if let Ok(mut rx) = self.rx.try_lock() { + while rx.try_recv().is_ok() {} + } + } + + /// The next request to perform; `None` once finished. A request the exporter already gave up + /// on (timed out, cancelled by shutdown or the opt-out) is never handed out. + pub async fn next(&self) -> Option { + loop { + if self.finished.load(Ordering::SeqCst) { + return None; + } + let pending = self.rx.lock().await.recv().await?; + if let Some(export) = self.serve(pending) { + return Some(export); + } + if self.finished.load(Ordering::SeqCst) { + return None; + } + } + } + + /// The request waiting right now, if any — without waiting: for hosts that must never leave + /// a pending Rust future holding one of their continuations (Dart: poll from a `Timer`). A + /// request nobody polls for still times out and is withdrawn exactly as with + /// [`next`](Self::next); `None` also once finished, or while a `next` is waiting. + pub fn try_next(&self) -> Option { + let mut rx = self.rx.try_lock().ok()?; + loop { + if self.finished.load(Ordering::SeqCst) { + return None; + } + if let Some(export) = self.serve(rx.try_recv().ok()?) { + return Some(export); + } + } + } + + /// The collector's answer to the request with `id`, whatever its status; the core classifies it. + pub fn complete(&self, id: u64, response: ExportResponse) { + self.settle(id, Ok(response)); + } + + /// The request with `id` got no answer (network error, timeout, invalid URL). + pub fn fail(&self, id: u64, error: ExportError) { + self.settle(id, Err(error)); + } +} + +impl TelemetryExportQueue { + /// Hand `pending` out, unless the queue is finished or its exporter gave up on it. + fn serve(&self, pending: Pending) -> Option { + let opted_out = self.process.load(Ordering::SeqCst) && global::is_disabled(); + if opted_out || self.finished.load(Ordering::SeqCst) || pending.done.is_closed() { + return None; + } + let mut inflight = self.inflight.lock().unwrap_or_else(|e| e.into_inner()); + inflight.retain(|_, done| !done.is_closed()); + inflight.insert(pending.export.id, pending.done); + Some(pending.export) + } + + fn settle(&self, id: u64, outcome: Result) { + let done = self.inflight.lock().unwrap_or_else(|e| e.into_inner()).remove(&id); + if let Some(done) = done { + let _ = done.send(outcome); + } + } + + /// Drop queued requests nobody waits for any more (their exporter timed out or stopped): + /// with one request out at a time, anything still queued when a new one comes is stale. + fn discard_stale(&self) { + let Ok(mut rx) = self.rx.try_lock() else { return }; + let mut live = Vec::new(); + while let Ok(pending) = rx.try_recv() { + if !pending.done.is_closed() { + live.push(pending); + } + } + drop(rx); + let tx = self.tx.lock().unwrap_or_else(|e| e.into_inner()).clone(); + for pending in live { + if let Some(tx) = &tx { + let _ = tx.try_send(pending); + } + } + } +} + +/// Forgets an in-flight entry when the exporter stops waiting for its answer (timeout, +/// shutdown, opt-out), so nothing is kept for an answer nobody will read. +struct Forget<'a> { + queue: &'a TelemetryExportQueue, + id: u64, +} + +impl Drop for Forget<'_> { + fn drop(&mut self) { + self.queue.inflight.lock().unwrap_or_else(|e| e.into_inner()).remove(&self.id); + } +} + +#[async_trait::async_trait] +impl TelemetryTransport for TelemetryExportQueue { + async fn send(&self, request: ExportRequest) -> Result { + let closed = || ExportError::Retryable { + reason: "export queue closed".into(), + retry_after_ms: None, + }; + self.discard_stale(); + let id = self.seq.fetch_add(1, Ordering::Relaxed); + let (done, wait) = oneshot::channel(); + let pending = Pending { export: PendingExport { id, request }, done }; + let _forget = Forget { queue: self, id }; + let Some(tx) = self.tx.lock().unwrap_or_else(|e| e.into_inner()).clone() else { + return Err(closed()); + }; + // Waits for room while the host is busy; cancelled with the exporter's timeout, and then + // the request was never queued. The sender is not kept while the answer is awaited. + tx.send(pending).await.map_err(|_| closed())?; + drop(tx); + #[cfg(test)] + self.queued.fetch_add(1, Ordering::SeqCst); + wait.await.unwrap_or(Err(ExportError::Retryable { + reason: "host dropped the export".into(), + retry_after_ms: None, + })) + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::*; + + fn request() -> ExportRequest { + ExportRequest { + url: "https://p.livekit.cloud/observability/client/logs/otlp/v0".into(), + headers: HashMap::from([("Authorization".into(), "Bearer secret".into())]), + body: vec![1, 2, 3], + } + } + + fn runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread().enable_time().build().expect("runtime") + } + + /// A host that stopped serving never gets a pile of stale requests — and their tokens — to send + /// later: requests the exporter gave up on are discarded, not served. + #[test] + fn a_suspended_host_is_never_served_stale_requests() { + runtime().block_on(async { + let queue = TelemetryExportQueue::new(); + for _ in 0..5 { + let sent = tokio::time::timeout(Duration::from_millis(10), queue.send(request())); + assert!(sent.await.is_err(), "the exporter's timeout gives up"); + } + assert!(queue.inflight.lock().expect("lock").is_empty(), "nothing kept for answers"); + let served = tokio::time::timeout(Duration::from_millis(50), queue.next()).await; + assert!(served.is_err(), "no stale request is handed out"); + }); + } + + /// A request cancelled before the host asked for it (shutdown, opt-out) is never served; a live + /// one still is. + #[test] + fn a_cancelled_request_is_never_served() { + runtime().block_on(async { + let queue = TelemetryExportQueue::new(); + let cancelled = { + let queue = queue.clone(); + tokio::spawn(async move { queue.send(request()).await }) + }; + tokio::task::yield_now().await; + cancelled.abort(); + let _ = cancelled.await; + let live = { + let queue = queue.clone(); + tokio::spawn(async move { queue.send(request()).await }) + }; + tokio::task::yield_now().await; + let served = queue.next().await.expect("the live request"); + queue.complete(served.id, ExportResponse::accepted()); + assert_eq!(live.await.expect("task"), Ok(ExportResponse::accepted())); + }); + } + + /// Finishing the queue discards what is queued instead of serving it. + #[test] + fn finishing_discards_what_is_queued() { + runtime().block_on(async { + let queue = TelemetryExportQueue::new(); + let waiting = { + let queue = queue.clone(); + tokio::spawn(async move { queue.send(request()).await }) + }; + tokio::task::yield_now().await; + queue.finish(); + assert!(queue.next().await.is_none(), "not served after finish"); + assert!(waiting.await.expect("task").is_err(), "the exporter hears it was dropped"); + }); + } + + /// `try_next` never waits. It serves a live request, never one its exporter gave up on while + /// nobody polled, and nothing once finished. + #[test] + fn try_next_serves_without_waiting_and_never_a_stale_request() { + runtime().block_on(async { + let queue = TelemetryExportQueue::new(); + assert!(queue.try_next().is_none(), "empty: returns at once"); + let sent = tokio::time::timeout(Duration::from_millis(10), queue.send(request())); + assert!(sent.await.is_err(), "nobody polled: the exporter's timeout gives up"); + assert!(queue.try_next().is_none(), "the withdrawn request is never served"); + let live = { + let queue = queue.clone(); + tokio::spawn(async move { queue.send(request()).await }) + }; + tokio::task::yield_now().await; + let served = queue.try_next().expect("the live request"); + queue.complete(served.id, ExportResponse::accepted()); + assert_eq!(live.await.expect("task"), Ok(ExportResponse::accepted())); + queue.complete(served.id, ExportResponse::accepted()); // stale id: ignored + queue.fail( + u64::MAX, + ExportError::Retryable { reason: "x".into(), retry_after_ms: None }, + ); + queue.finish(); + assert!(queue.try_next().is_none()); + }); + } + + /// An unsigned participant token with the observability grant (the core reads claims, never + /// verifies them). + fn granted_token() -> String { + fn b64url(bytes: &[u8]) -> String { + const ABC: &[u8; 64] = + b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"; + let mut out = String::new(); + for chunk in bytes.chunks(3) { + let n = chunk + .iter() + .enumerate() + .fold(0u32, |n, (i, b)| n | (*b as u32) << (16 - 8 * i)); + for i in 0..=chunk.len() { + out.push(ABC[(n >> (18 - 6 * i) & 63) as usize] as char); + } + } + out + } + let exp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("clock") + .as_secs() + + 3600; + let claims = format!(r#"{{"exp":{exp},"observability":{{"write":true}}}}"#); + format!("{}.{}.sig", b64url(br#"{"alg":"HS256"}"#), b64url(claims.as_bytes())) + } + + /// Wait (bounded) until `condition` holds. + fn eventually(condition: impl Fn() -> bool) -> bool { + (0..500).any(|_| { + condition() || { + std::thread::sleep(Duration::from_millis(10)); + false + } + }) + } + + /// A second opt-out while the first one's purge is blocked does not let a waiter + /// (`telemetry_flush`/`telemetry_shutdown`) return before that first purge finishes. + #[test] + fn a_repeated_opt_out_waits_for_the_first_purge() { + static SLOT: PurgeSlot = Mutex::new(None); + let (release, blocked) = oneshot::channel::<()>(); + start_purge(&SLOT, blocked); + start_purge(&SLOT, async {}); + runtime().block_on(async { + let early = tokio::time::timeout(Duration::from_millis(200), wait_purge(&SLOT)).await; + assert!(early.is_err(), "the first purge is still running"); + release.send(()).expect("first purge waiting"); + tokio::time::timeout(Duration::from_secs(5), wait_purge(&SLOT)) + .await + .expect("done once the first purge is"); + }); + } + + /// The opt-out is in effect when `telemetry_disable` returns — no scope, nothing captured — and + /// once its purge is awaited (`telemetry_shutdown`) the pulled request of a pipeline replaced + /// while its upload was pending is never handed to the host; `telemetry_is_disabled` turns + /// `true` for every thread. (The only test here that touches the process-wide pipeline: the + /// opt-out is one-way.) + #[test] + fn the_opt_out_withdraws_a_replaced_generations_pulled_request() { + let config = || TelemetryConfig { resource: Vec::new(), ..Default::default() }; + let first = telemetry_configure_pulled(config(), Vec::new()); + let room = telemetry_scope().expect("scope"); + room.set_server("wss://p.livekit.cloud".into(), granted_token()); + room.emit_custom("ping".into(), HashMap::new()); + crate::runtime::runtime().spawn(telemetry_flush()); + assert!( + eventually(|| first.queued.load(Ordering::SeqCst) == 1), + "precondition: a pulled request is pending in the first queue" + ); + let _second = telemetry_configure_pulled(config(), Vec::new()); // `first` now drains + assert!(!telemetry_is_disabled()); + telemetry_disable(); + assert!(telemetry_is_disabled(), "visible process-wide once the call returns"); + assert!(std::thread::spawn(telemetry_is_disabled).join().expect("thread")); + assert!(first.try_next().is_none(), "nothing is served once the call returns"); + assert!(telemetry_scope().is_none(), "no scope once the call returns"); + assert!(telemetry_stats().is_none()); + room.emit_custom("after".into(), HashMap::new()); // refused: its generation is revoked + // A configure after the opt-out deletes what its storage dir still holds and starts no + // exporter: the queue it hands back is held by nothing else. + let dir = std::env::temp_dir().join(format!("lk-refused-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir(&dir).expect("dir"); + let stamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("clock") + .as_nanos(); + std::fs::write(dir.join(format!("{stamp:020}-000001-1-l.otlp")), b"cached").expect("batch"); + let refused = telemetry_configure_pulled( + TelemetryConfig { storage_dir: Some(dir.to_string_lossy().into_owned()), ..config() }, + Vec::new(), + ); + assert_eq!(Arc::strong_count(&refused), 1, "no exporter holds its transport"); + assert_eq!(std::fs::read_dir(&dir).expect("dir").count(), 0, "the cache is purged"); + let _ = std::fs::remove_dir_all(&dir); + runtime().block_on(async { + telemetry_shutdown().await; // awaits the purge + let served = tokio::time::timeout(Duration::from_millis(300), first.next()).await; + assert!(!matches!(served, Ok(Some(_))), "withdrawn, never served"); + }); + } +} diff --git a/livekit-ffi/support/dart/test/telemetry_test.dart b/livekit-ffi/support/dart/test/telemetry_test.dart new file mode 100644 index 000000000..3bd160705 --- /dev/null +++ b/livekit-ffi/support/dart/test/telemetry_test.dart @@ -0,0 +1,82 @@ +import 'dart:convert'; +import 'dart:typed_data'; + +import 'package:livekit_ffi/livekit_ffi.dart'; +import 'package:livekit_ffi/livekit_telemetry.dart'; +import 'package:test/test.dart'; + +/// Rust must never call back into Dart from its own threads: uniffi-dart's foreign-trait +/// callbacks are isolate-bound (`Pointer.fromFunction`) and the VM aborts with "Cannot invoke +/// native callback outside an isolate" when the exporter invokes `TelemetryTransport.send` from +/// a tokio worker. The pull queue inverts the direction: Dart polls `tryNext()` from its own +/// timer (never leaving a Rust future holding a continuation of an isolate that may die), +/// performs the request, and reports the collector's answer back with `complete()` (or `fail()` +/// when there was none); the core classifies the status and body. +Future serve(TelemetryExportQueue queue, List sink, int count) async { + for (var i = 0; i < count; i++) { + var pending = queue.tryNext(); + while (pending == null) { + await Future.delayed(const Duration(milliseconds: 10)); + pending = queue.tryNext(); + } + sink.add(pending.request); + queue.complete(id: pending.id, response: ExportResponse(status: 200, headers: {}, body: Uint8List(0))); + } +} + +/// An unsigned participant token with the observability grant: the core reads claims, never +/// verifies them (the collector does). +String grantedToken() { + String part(Map json) => + base64Url.encode(utf8.encode(jsonEncode(json))).replaceAll('=', ''); + final exp = DateTime.now().millisecondsSinceEpoch ~/ 1000 + 3600; + return '${part({'alg': 'HS256'})}.${part({'exp': exp, 'observability': {'write': true}})}.sig'; +} + +void main() { + group('telemetry', () { + test('exports a room\'s records to its project through the pull queue', () async { + final requests = []; + final queue = telemetryConfigurePulled( + config: TelemetryConfig(resource: [], logSeverity: Severity.warn), + instruments: [], + ); + final serving = serve(queue, requests, 2); + + final room = telemetryScope()!; + final token = grantedToken(); + room.setServer(url: 'wss://my-project.livekit.cloud', token: token); + room.setAttribute(key: 'app.call_id', value: 'c1'); + room.emitCustom(name: 'checkout', attributes: {'plan': 'pro'}); + await telemetryFlush(); + expect(requests, hasLength(1)); + expect(requests.single.url, 'https://my-project.livekit.cloud/observability/client/logs/otlp/v0'); + expect(requests.single.headers['Authorization'], 'Bearer $token'); + expect(requests.single.headers['Content-Type'], 'application/x-protobuf'); + expect(requests.single.body, isNotEmpty); + expect(telemetryStats()!.uploadsSent, 1); + expect(telemetryStats()!.dropped, 0); + expect(room.statsPollIntervalMs(), 30000); + + // Shutdown leaves the session summary, a process-level batch of its own. + await telemetryShutdown(); + await serving; + expect(requests, hasLength(2)); + expect(telemetryStats(), isNull); + // The queue outlives the pipeline (Dart holds it): finishing it ends the serving loop. + queue.finish(); + expect(await queue.next(), isNull); + }); + + test('refuses to start without any transport', () { + expect( + () => telemetryConfigure( + config: TelemetryConfig(resource: [], logSeverity: Severity.warn), + transport: null, + instruments: [], + ), + throwsA(anything), + ); + }); + }); +} diff --git a/livekit-telemetry/SPEC.md b/livekit-telemetry/SPEC.md index a3b199703..1ccbcabc8 100644 --- a/livekit-telemetry/SPEC.md +++ b/livekit-telemetry/SPEC.md @@ -415,7 +415,7 @@ never. Only plain records at or above `log_severity` (default `warn`) leave the are dropped in `emit`. The core's own warnings and errors (Rust `log` records from `livekit*` targets, never -`livekit_telemetry*`) reach the pipeline by themselves through `livekit-uniffi`'s log forwarder: +`livekit_telemetry*`) reach the pipeline by themselves through `livekit-ffi`'s log forwarder: it copies them as `ffi` records — every JWT-shaped substring (`eyJ…` `.` … `.` …) masked as ``, so a token quoted in an error never leaves the device — and forwards the console entry unchanged. The copy is active wherever the platform calls `log_forward_bootstrap` (Swift's diff --git a/livekit-telemetry/uniffi.toml b/livekit-telemetry/uniffi.toml index bb41bd4cc..67242c4ba 100644 --- a/livekit-telemetry/uniffi.toml +++ b/livekit-telemetry/uniffi.toml @@ -3,5 +3,5 @@ ffi_module_name = "RustLiveKitTelemetry" [bindings.kotlin] # The Kotlin checksum test is broken on ARM in every UniFFI release this workspace can use; see -# livekit-uniffi/uniffi.toml for the full explanation. +# livekit-ffi/uniffi.toml for the full explanation. omit_checksums = true From 22ff47b9aa7d0552f6e90b5b09a5d375e8e522b4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?= <86720177+pblazej@users.noreply.github.com> Date: Thu, 1 Oct 2026 12:41:24 +0200 Subject: [PATCH 2/4] feat(uniffi): copy the core's own warnings and errors into telemetry Once a platform installs the log forwarder, warnings and errors from `livekit*` targets also reach the telemetry pipeline as log records, with any compact JWT and bearer or token credential values masked and the target as the logger (`log` has no function name to fill in); the telemetry crate's own records are never copied. The platform's level still filters only what is forwarded to it, so the console output is unchanged. --- livekit-ffi/src/core_modules/log_forward.rs | 326 +++++++++++++++++++- 1 file changed, 324 insertions(+), 2 deletions(-) diff --git a/livekit-ffi/src/core_modules/log_forward.rs b/livekit-ffi/src/core_modules/log_forward.rs index 1cb669c55..fd3ed751c 100644 --- a/livekit-ffi/src/core_modules/log_forward.rs +++ b/livekit-ffi/src/core_modules/log_forward.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::sync::atomic::{AtomicUsize, Ordering}; + use log::{Level, LevelFilter, Log, Record}; use once_cell::sync::OnceCell; use tokio::sync::{mpsc, Mutex}; @@ -24,11 +26,18 @@ static LOGGER: OnceCell = OnceCell::new(); /// Generally, you will invoke this once early in program execution. However, /// subsequent invocations are allowed to change the log level. /// +/// Also turns on the telemetry copy of the core's own warnings and errors (see +/// `telemetry_copy`), whatever `level` is: `level` filters only what is forwarded here. Don't +/// pass the forwarded entries on to `telemetry_log`. +/// #[uniffi::export] fn log_forward_bootstrap(level: LevelFilter) { let logger = LOGGER.get_or_init(Logger::new); _ = log::set_logger(logger); // Returns an error if already set (ignore) - log::set_max_level(level); + logger.forward.store(level as usize, Ordering::Relaxed); + // Never stricter than `Warn`: the telemetry copy needs the core's warnings even when the + // platform forwards only errors. + log::set_max_level(level.max(LevelFilter::Warn)); } /// Asynchronously receives a forwarded log entry. @@ -77,12 +86,15 @@ pub struct LogForwardEntry { struct Logger { tx: mpsc::UnboundedSender, rx: Mutex>, + /// The platform's level (a `LevelFilter` as `usize`): what is forwarded to it. The global + /// `log` max level may be looser, for the telemetry copy. + forward: AtomicUsize, } impl Logger { fn new() -> Self { let (tx, rx) = mpsc::unbounded_channel(); - Self { tx, rx: rx.into() } + Self { tx, rx: rx.into(), forward: AtomicUsize::new(LevelFilter::Trace as usize) } } } @@ -91,6 +103,15 @@ impl Log for Logger { true } fn log(&self, record: &Record) { + // Telemetry gets its copy first, built from the borrowed record; the console entry below + // is exactly what it always was. + if let Some(copy) = telemetry_copy(record) { + livekit_telemetry::global::log(copy); + } + // `Level` and `LevelFilter` share their numbering (`Error` = 1 … `Trace` = 5). + if record.level() as usize > self.forward.load(Ordering::Relaxed) { + return; + } let record = LogForwardEntry { level: record.metadata().level(), target: record.target().to_string(), @@ -103,3 +124,304 @@ impl Log for Logger { } fn flush(&self) {} } + +/// The core's own warnings and errors — `livekit*` targets at `Warn` or `Error` — as telemetry +/// log records, so no platform has to route them there itself. Works once the platform has +/// installed this forwarder (`log_forward_bootstrap`): Rust allows one logger per process, and +/// without it the core's `log` records go nowhere. Platforms must therefore not also feed these +/// forwarded entries into `telemetry_log`, or they would be counted twice. +/// +/// Never the telemetry crate's own records (`livekit_telemetry*`): a record about a failed +/// upload must not produce another upload. The pipeline applies the same guard again. +fn telemetry_copy(record: &Record) -> Option { + let severity = match record.level() { + Level::Error => livekit_telemetry::Severity::Error, + Level::Warn => livekit_telemetry::Severity::Warn, + _ => return None, + }; + let target = record.target(); + if !target.starts_with("livekit") || target.starts_with("livekit_telemetry") { + return None; + } + Some(livekit_telemetry::LogRecord { + severity, + source: livekit_telemetry::LogSource::Ffi, + body: mask_jwts(&record.args().to_string()), + logger: Some(target.to_owned()), + // `log` knows no function name; the module path is the target, already in `logger`. + function: None, + file: record.file().map(str::to_owned), + line: record.line(), + timestamp_ns: None, + span_id: None, + }) +} + +/// `body` with credentials replaced: the value after `Bearer`, or after a `token` key (`token=`, +/// `access_token=`, `"token": "…"`, quoted or not, any spacing), becomes ``, and every +/// compact JWT — three base64url segments whose header and payload decode to JSON objects, +/// whatever their formatting or the punctuation around them — becomes ``. A participant +/// token quoted in an error must not leave the device. +fn mask_jwts(body: &str) -> String { + let body = mask_credential_values(body); + let jwt_char = |c: char| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.'); + let mut out = String::with_capacity(body.len()); + let mut rest = body.as_str(); + while let Some(start) = rest.find(jwt_char) { + out.push_str(&rest[..start]); + let run = &rest[start..]; + let run = &run[..run.find(|c| !jwt_char(c)).unwrap_or(run.len())]; + mask_run(run, &mut out); + rest = &rest[start + run.len()..]; + } + out.push_str(rest); + out +} + +/// Push `run` (base64url characters and dots) with every `header.payload.signature` inside it — +/// the signature may be empty, dots or a word glued on with `-`/`_` may surround it — as ``. +fn mask_run(run: &str, out: &mut String) { + let segments: Vec<&str> = run.split('.').collect(); + let mut i = 0; + while i < segments.len() { + if i > 0 { + out.push('.'); + } + let token = segments.get(i + 2).and_then(|_| { + json_object(segments[i + 1]).then_some(())?; + header_start(segments[i]) + }); + let Some(at) = token else { + out.push_str(segments[i]); + i += 1; + continue; + }; + out.push_str(&segments[i][..at]); + out.push_str(""); + i += 3; + } +} + +/// Where a JWT header starts in `segment`: at its start, or after a `-`/`_` joining a word to it. +fn header_start(segment: &str) -> Option { + std::iter::once(0) + .chain(segment.match_indices(['-', '_']).map(|(i, _)| i + 1)) + .find(|&at| json_object(&segment[at..])) +} + +/// A base64url segment whose decoded bytes start, after whitespace, with `{`. +fn json_object(segment: &str) -> bool { + let sextet = |c: u8| match c { + b'A'..=b'Z' => c - b'A', + b'a'..=b'z' => c - b'a' + 26, + b'0'..=b'9' => c - b'0' + 52, + b'-' => 62, + _ => 63, // `_`: the caller only passes base64url characters + }; + let (mut acc, mut bits) = (0u32, 0); + for c in segment.bytes() { + acc = (acc << 6 | u32::from(sextet(c))) & 0xFFF; + bits += 6; + if bits >= 8 { + bits -= 8; + let byte = (acc >> bits) as u8; + if !byte.is_ascii_whitespace() { + return byte == b'{'; + } + } + } + false +} + +/// The value after a credential marker (matched case-insensitively) as ``: after +/// `bearer` and whitespace, or after `bearer`/`token`, an optional closing quote, `=` or `:`, with any +/// whitespace and an optional opening quote in between. A quoted value ends at its closing quote, +/// any other at a delimiter. +fn mask_credential_values(body: &str) -> String { + // Same byte offsets as `body`: ASCII lowercasing never changes a length. + let lower = body.to_ascii_lowercase(); + let mut out = String::with_capacity(body.len()); + let (mut at, mut from) = (0, 0); + while let Some((key, len)) = ["bearer", "token"] + .iter() + .filter_map(|k| Some((lower[from..].find(k)? + from, k.len()))) + .min() + { + from = key + len; + let Some(value) = credential_value(&body[from..], len == "bearer".len()) else { continue }; + let (start, end) = (from + value.start, from + value.end); + out.push_str(&body[at..start]); + out.push_str(""); + at = end; + from = end; + } + out.push_str(&body[at..]); + out +} + +/// The non-empty value in `rest`, which follows a `bearer` (`bearer`) or `token` key, as a byte +/// range. Only a bearer value may follow its key after whitespace alone: `token expired` is prose. +fn credential_value(rest: &str, bearer: bool) -> Option> { + let trimmed = |s: &str| s.len() - s.trim_start().len(); + let skip = if bearer && rest.starts_with(char::is_whitespace) { + trimmed(rest) + } else { + let quote = usize::from(rest.starts_with(['"', '\''])); + let after = &rest[quote..]; + let gap = trimmed(after); + if !after[gap..].starts_with(['=', ':']) { + return None; + } + quote + gap + 1 + trimmed(&after[gap + 1..]) + }; + let value = &rest[skip..]; + let (start, len) = match value.chars().next() { + Some(q @ ('"' | '\'')) => (skip + 1, value[1..].find(q).unwrap_or(value.len() - 1)), + _ => ( + skip, + value + .find(|c: char| { + c.is_whitespace() || matches!(c, '&' | '"' | '\'' | ',' | ';' | ')' | '<' | '>') + }) + .unwrap_or(value.len()), + ), + }; + (len > 0).then_some(start..start + len) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn record<'a>(level: Level, target: &'a str, args: std::fmt::Arguments<'a>) -> Record<'a> { + Record::builder() + .level(level) + .target(target) + .module_path(Some(target)) + .args(args) + .file(Some("f.rs")) + .line(Some(7)) + .build() + } + + /// The core's own warnings and errors reach telemetry; nothing else does, and never the + /// telemetry crate's own (no feedback loop). + #[test] + fn only_the_cores_own_warnings_and_errors_are_copied() { + let copied = |level, target| telemetry_copy(&record(level, target, format_args!("boom"))); + let copy = copied(Level::Warn, "livekit::rtc_engine").expect("a core warning"); + assert_eq!(copy.severity, livekit_telemetry::Severity::Warn); + assert_eq!(copy.source, livekit_telemetry::LogSource::Ffi); + assert_eq!( + (copy.body.as_str(), copy.logger.as_deref()), + ("boom", Some("livekit::rtc_engine")) + ); + assert_eq!((copy.file.as_deref(), copy.line), (Some("f.rs"), Some(7))); + assert_eq!(copy.function, None, "the module path is already the logger"); + assert!(copied(Level::Error, "livekit_datatrack").is_some()); + assert!(copied(Level::Info, "livekit::room").is_none(), "info stays on the console"); + assert!(copied(Level::Error, "livekit_telemetry::exporter").is_none(), "no feedback loop"); + assert!(copied(Level::Error, "hyper::client").is_none(), "not the core's"); + assert!(copied(Level::Error, "my_app").is_none()); + } + + /// A token quoted in a core error never reaches telemetry; the console still gets the line. + #[test] + fn tokens_in_copied_records_are_masked() { + let jwt = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJib2IifQ.c2lnbmF0dXJl-_x"; + let copy = telemetry_copy(&record( + Level::Error, + "livekit::signal_client", + format_args!("connect failed: wss://x/rtc?access_token={jwt}&auto=1 ({jwt})"), + )) + .expect("copied"); + assert_eq!(copy.body, "connect failed: wss://x/rtc?access_token=&auto=1 ()"); + // A valid token need not start `eyJ` — a header `{ "alg":…` and a payload with leading + // whitespace encode differently — and a transport error may quote the bearer value itself. + let spaced = "eyAiYWxnIjoiSFMyNTYiLCJ0eXAiOiJKV1QifQ\ + .CiAgeyJzdWIiOiJib2IiLCJ2aWRlbyI6eyJyb29tIjoiciJ9fQ.c2lnbmF0dXJlLWJ5dGVz"; + let copy = telemetry_copy(&record( + Level::Warn, + "livekit_signaling", + format_args!( + "signal transport error: HTTP 401 for request with Authorization: Bearer opaque-\ + credential; token {spaced}. retrying" + ), + )) + .expect("copied"); + assert_eq!( + copy.body, + "signal transport error: HTTP 401 for request with Authorization: Bearer ; \ + token . retrying" + ); + assert_eq!(mask_jwts(r#"{"token":"abc.def"}"#), r#"{"token":""}"#); + // A JWT inside a punctuation-delimited run, and credential values that are quoted or + // spaced. + for (logged, masked) in [ + (format!("token (...{jwt})"), "token (...)".to_owned()), + (format!("token (...{spaced})"), "token (...)".to_owned()), + (format!("id=lk_{jwt}.x"), "id=lk_.x".to_owned()), + (r#"token="opaque-secret""#.into(), r#"token="""#.into()), + (r#"{"token": "opaque-secret"}"#.into(), r#"{"token": ""}"#.into()), + ( + r#"{"token" : 'opaque secret', "k": 1}"#.into(), + r#"{"token" : '', "k": 1}"#.into(), + ), + ( + "Authorization: Bearer opaque-secret".into(), + "Authorization: Bearer ".into(), + ), + ("ACCESS_TOKEN = opaque-secret&x".into(), "ACCESS_TOKEN = &x".into()), + ] { + assert_eq!(mask_jwts(&logged), masked, "{logged}"); + } + let unsigned = "eyAiYWxnIjoiSFMyNTYiLCJ0eXAiOiJKV1QifQ.CiAgeyJzdWIiOiJib2IifQ."; + assert_eq!(mask_jwts(unsigned), "", "an unsigned token is masked too"); + for kept in [ + "eyJ", + "eyJabc.", + "eyJabc..sig", + "see eyJ-not-a-token.", + "plain", + "wss://p.livekit.cloud/rtc", + "v1.2.3 of livekit_api.client.rs", + "token expired, tokens: 3, token_type=x", + "(...eyJ)", + ] { + assert_eq!(mask_jwts(kept), kept); + } + } + + /// A platform forwarding only errors (`OSLogger(minLevel: .error)`) still gets the core's + /// warnings into telemetry, and its console still sees only errors. + #[test] + fn the_platforms_level_filters_the_console_not_the_telemetry_copy() { + log_forward_bootstrap(LevelFilter::Error); + assert_eq!(log::max_level(), LevelFilter::Warn, "warnings still reach the logger"); + let logger = Logger::new(); + logger.forward.store(LevelFilter::Error as usize, Ordering::Relaxed); + let warning = record(Level::Warn, "livekit::room", format_args!("careful")); + assert!(telemetry_copy(&warning).is_some(), "telemetry copies the warning"); + logger.log(&warning); + logger.log(&record(Level::Error, "livekit::room", format_args!("boom"))); + let mut rx = logger.rx.try_lock().expect("rx"); + assert_eq!(rx.try_recv().expect("forwarded").level, Level::Error, "the error is forwarded"); + assert!(rx.try_recv().is_err(), "the warning is not: the console is unchanged"); + drop(rx); + log_forward_bootstrap(LevelFilter::Debug); + assert_eq!(log::max_level(), LevelFilter::Debug, "a looser level is kept as is"); + } + + /// The console entry is unchanged by the copy. + #[test] + fn the_console_entry_is_what_it_always_was() { + let logger = Logger::new(); + logger.log(&record(Level::Warn, "livekit::room", format_args!("hello {}", 1))); + let entry = logger.rx.try_lock().expect("rx").try_recv().expect("forwarded"); + assert_eq!((entry.level, entry.target.as_str()), (Level::Warn, "livekit::room")); + assert_eq!( + (entry.message.as_str(), entry.file.as_deref(), entry.line), + ("hello 1", Some("f.rs"), Some(7)) + ); + } +} From 9c011ac0470d6fdf8f402d638fec6933b711035b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?= <86720177+pblazej@users.noreply.github.com> Date: Thu, 1 Oct 2026 12:42:26 +0200 Subject: [PATCH 3/4] docs(telemetry): add the telemetry_ping example and local testing guide `telemetry_ping` records a small session (an `lk.connect` span with its checkpoints, an `lk.publish`, an `lk.subscribe` ended by first media, a custom event) and exports it to the collector `LK_TELEMETRY_ENDPOINT` names; the README shows how to run it against a local OpenTelemetry backend, and a Dart test does the same through the pull queue. --- Cargo.lock | 10 ++ Cargo.toml | 1 + examples/telemetry_ping/Cargo.toml | 11 +++ examples/telemetry_ping/src/main.rs | 91 +++++++++++++++++++ .../dart/test/live_collector_test.dart | 88 ++++++++++++++++++ livekit-telemetry/README.md | 18 ++++ 6 files changed, 219 insertions(+) create mode 100644 examples/telemetry_ping/Cargo.toml create mode 100644 examples/telemetry_ping/src/main.rs create mode 100644 livekit-ffi/support/dart/test/live_collector_test.dart diff --git a/Cargo.lock b/Cargo.lock index 11ff39475..0df7e69c4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7256,6 +7256,16 @@ version = "0.13.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "adb6935a6f5c20170eeceb1a3835a49e12e19d792f6dd344ccc76a985ca5a6ca" +[[package]] +name = "telemetry_ping" +version = "0.1.0" +dependencies = [ + "env_logger 0.11.11", + "livekit-net", + "livekit-telemetry", + "tokio", +] + [[package]] name = "tempfile" version = "3.27.0" diff --git a/Cargo.toml b/Cargo.toml index a969ac0c6..49417b2c5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -41,6 +41,7 @@ members = [ "examples/rpc", "examples/save_to_disk", "examples/screensharing", + "examples/telemetry_ping", "examples/token_source", "examples/webhooks", ] diff --git a/examples/telemetry_ping/Cargo.toml b/examples/telemetry_ping/Cargo.toml new file mode 100644 index 000000000..dee6f7709 --- /dev/null +++ b/examples/telemetry_ping/Cargo.toml @@ -0,0 +1,11 @@ +[package] +name = "telemetry_ping" +version = "0.1.0" +edition = "2021" +publish = false + +[dependencies] +livekit-telemetry = { workspace = true, features = ["net"] } +livekit-net = { workspace = true, features = ["native", "rustls-tls-native-roots"] } +tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } +env_logger = { workspace = true } diff --git a/examples/telemetry_ping/src/main.rs b/examples/telemetry_ping/src/main.rs new file mode 100644 index 000000000..79cc00a13 --- /dev/null +++ b/examples/telemetry_ping/src/main.rs @@ -0,0 +1,91 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Records a small session through the telemetry core — an `lk.connect` span with its +//! checkpoints, an `lk.publish`, an `lk.subscribe` ended by first media, and a custom event — +//! and exports it. +//! +//! - A local OpenTelemetry collector: `LK_TELEMETRY_ENDPOINT=http://localhost:4318` (the core's +//! test-only override; any server URL and token are accepted). +//! - A LiveKit Cloud project: `LK_URL=wss://.livekit.cloud` and `LK_TOKEN=`. + +use std::{env, sync::Arc}; + +use livekit_telemetry::{ + Attribute, NetTransport, RoomIdentity, RtcStatsSample, SpanName, SpanOutcome, SpanStep, + SpanTrack, StreamDirection, Telemetry, TelemetryConfig, TrackKind, TrackSource, + ENDPOINT_OVERRIDE_ENV, +}; + +#[tokio::main] +async fn main() { + env_logger::init(); + let url = env::var("LK_URL").unwrap_or_else(|_| "ws://localhost:7880".to_owned()); + let token = env::var("LK_TOKEN").unwrap_or_default(); + if env::var(ENDPOINT_OVERRIDE_ENV).is_err() && env::var("LK_URL").is_err() { + eprintln!("set {ENDPOINT_OVERRIDE_ENV}=http://localhost:4318, or LK_URL and LK_TOKEN"); + return; + } + + let config = TelemetryConfig { + resource: vec![ + Attribute::new("service.name", "telemetry_ping"), + Attribute::new("os.name", env::consts::OS), + ], + // Optional on-disk cache: run once with the collector down, once with it up. + storage_dir: env::var("LK_TELEMETRY_DIR").ok(), + ..Default::default() + }; + let transport = NetTransport::from_registry().expect("livekit-net has no HTTP client"); + let (telemetry, exporter) = Telemetry::new(config, Arc::new(transport)); + tokio::spawn(exporter.run()); + + let room = telemetry.begin_scope(); + room.set_server(&url, &token); + room.set_room(RoomIdentity { name: Some("telemetry-ping".into()), ..Default::default() }); + + let connect = room.start(SpanName::Connect, None); + for step in [SpanStep::WsOpen, SpanStep::Signal, SpanStep::JoinRecv, SpanStep::PcCreated] { + connect.step(step); + } + connect.end(SpanOutcome::Ok, None); + + let microphone = SpanTrack { + sid: Some("TR_ping_mic".into()), + kind: TrackKind::Audio, + source: TrackSource::Microphone, + remote_identity: None, + }; + let publish = room.start(SpanName::Publish, None); + publish.set_track(microphone); + publish.end(SpanOutcome::Ok, None); + + let remote = SpanTrack { + sid: Some("TR_ping_remote".into()), + kind: TrackKind::Video, + source: TrackSource::Camera, + remote_identity: Some("bob".into()), + }; + room.subscribe_started(remote.clone()); + room.subscribed(remote); + let mut media = + RtcStatsSample::new("TR_ping_remote", TrackKind::Video, StreamDirection::Inbound); + media.bytes = Some(1_500); + room.record_stats(media); // first media: the lk.subscribe span ends ok + + room.emit_custom("ping", vec![Attribute::new("seq", 1i64)]); + telemetry.shutdown().await; + println!("trace {} — {}", room.trace_id(), telemetry.stats()); +} diff --git a/livekit-ffi/support/dart/test/live_collector_test.dart b/livekit-ffi/support/dart/test/live_collector_test.dart new file mode 100644 index 000000000..fc32a9050 --- /dev/null +++ b/livekit-ffi/support/dart/test/live_collector_test.dart @@ -0,0 +1,88 @@ +import 'dart:io'; +import 'dart:typed_data'; + +import 'package:http/http.dart' as http; +import 'package:livekit_ffi/livekit_ffi.dart'; +import 'package:livekit_ffi/livekit_telemetry.dart'; +import 'package:test/test.dart'; + +/// Posts one custom event to a real collector through the pull queue. +/// +/// Skipped unless a collector is named: `LK_TELEMETRY_ENDPOINT` (a local OpenTelemetry collector, +/// e.g. `http://localhost:4318` — read by the Rust core itself, the test-only override), or +/// `LK_TELEMETRY_URL` + `LK_TELEMETRY_TOKEN` (a LiveKit Cloud project and a token carrying the +/// observability grant). Unlike `telemetry_test.dart`, which fakes the collector's answer, this +/// one performs the request so the Dart binding is exercised against a real ingest. +/// +/// Dart drives the transport rather than implementing `TelemetryTransport`: uniffi-dart's +/// foreign-trait callbacks are isolate-bound, so Rust calling into Dart from a tokio worker aborts +/// the VM. The queue inverts that — Dart awaits `next()` and reports the answer with `complete()`. +Future serve(TelemetryExportQueue queue, List statuses, int count) async { + final client = http.Client(); + try { + for (var i = 0; i < count; i++) { + final pending = await queue.next(); + if (pending == null) return; + final request = pending.request; + try { + final response = await client.post( + Uri.parse(request.url), + headers: request.headers, + body: request.body, + ); + statuses.add(response.statusCode); + queue.complete( + id: pending.id, + response: ExportResponse( + status: response.statusCode, + headers: {}, + body: Uint8List.fromList(response.bodyBytes), + ), + ); + } catch (error) { + statuses.add(-1); + queue.fail( + id: pending.id, + error: RetryableExportException(reason: '$error', retryAfterMs: null), + ); + } + } + } finally { + client.close(); + } +} + +void main() { + final endpoint = Platform.environment['LK_TELEMETRY_ENDPOINT']; + final url = Platform.environment['LK_TELEMETRY_URL']; + final token = Platform.environment['LK_TELEMETRY_TOKEN']; + final target = endpoint ?? url; + + test('posts a custom event to a live collector through the pull queue', () async { + final statuses = []; + final queue = telemetryConfigurePulled( + config: TelemetryConfig( + resource: [ + Attribute(key: 'service.name', value: StrAttributeValue('livekit-client-dart')), + Attribute(key: 'service.version', value: StrAttributeValue('0.0.0-local')), + Attribute(key: 'os.name', value: StrAttributeValue(Platform.operatingSystem)), + ], + logSeverity: Severity.warn, + ), + instruments: [], + ); + final serving = serve(queue, statuses, 2); + + final room = telemetryScope()!; + room.setServer(url: url ?? 'ws://localhost:7880', token: token ?? 'local'); + room.emitCustom(name: 'ping', attributes: {'seq': '1'}); + await telemetryFlush(); + await telemetryShutdown(); + await serving; + queue.finish(); + + print('dart → $target: statuses $statuses'); + expect(statuses, isNotEmpty); + expect(statuses.first, inInclusiveRange(200, 299)); + }, skip: target == null ? 'set LK_TELEMETRY_ENDPOINT, or LK_TELEMETRY_URL and LK_TELEMETRY_TOKEN' : null); +} diff --git a/livekit-telemetry/README.md b/livekit-telemetry/README.md index 3e8691b74..e205eab85 100644 --- a/livekit-telemetry/README.md +++ b/livekit-telemetry/README.md @@ -59,6 +59,24 @@ println!("{}", telemetry.stats()); // drops by reason, uploads, cached batches Event names, attributes, cadences and the upload policy are defined in [`SPEC.md`](SPEC.md). +## Local testing + +Set `LK_TELEMETRY_ENDPOINT` in the process environment to send every upload to an OpenTelemetry +backend of your own instead of LiveKit Cloud — the test-only override, not part of any platform +API. A base URL gets the standard OTLP paths (`/v1/logs`, `/v1/traces`). + +```sh +docker run -d --name lk-lgtm -p 3000:3000 -p 4318:4318 grafana/otel-lgtm # OTLP/HTTP :4318, UI :3000 +LK_TELEMETRY_ENDPOINT=http://localhost:4318 cargo run -p telemetry_ping # prints the trace id +``` + +`telemetry_ping` records a small session (an `lk.connect` span with its checkpoints, an +`lk.publish`, an `lk.subscribe` ended by first media, a custom event). In the UI +(http://localhost:3000 → Explore): Loki `{service_name="telemetry_ping"}` for the log records, +Tempo with the printed trace id for the spans. Add `LK_TELEMETRY_DIR=/tmp/lk-telemetry` to use +the file cache: stop the container, run once, start it again and run once more to watch the +cached batches replay. Platform end-to-end tests set the same variable before the SDK starts. + ## Design notes - **The room decides the destination.** A platform passes only the server URL and the token From a2558a40ff1a2eb9a88067da1fe655e3ad3de753 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?B=C5=82az=CC=87ej=20Pankowski?= <86720177+pblazej@users.noreply.github.com> Date: Thu, 22 Oct 2026 14:17:03 +0200 Subject: [PATCH 4/4] fix(telemetry): hand exporters over in order and close pull-queue races One lifecycle task now owns the whole process pipeline lifecycle. Configure, shutdown and the opt-out's purge are commands it carries out one at a time, in the order they were sent: it installs, uninstalls and purges itself, so a configure and a shutdown can no longer interleave. `telemetry_configure` waits until its pipeline is installed, which never includes a drain, and the instruments start on the lifecycle's thread. The lifecycle hands each installed pipeline's exporter to its exporter runner, which runs them one at a time, in install order. A replacing exporter starts only once the one before it stopped: drained within its pipeline's own `shutdown_bound`, or else cancelled and joined at that bound, never left running behind a dropped task handle. Two exporters therefore never replay a shared cache at the same time, and a batch in flight is not sent twice, however many replaces come within one drain. `telemetry_shutdown` returns once every exporter queued before it has stopped; when the slot still holds the lifecycle its command went to, and that command was the last one, it then ends the lifecycle and joins it with its runner and exporter, and the next configure starts a fresh one. Each lifecycle has a generation of its own, so a shutdown that resumes late never ends a newer lifecycle whose last command has the same number. `telemetry_flush` and `telemetry_shutdown` wait for the opt-out's purge through the same command order. `global::uninstall` is public, the synchronous half of `global::shutdown`, for the lifecycle to uninstall in turn. The pull queue's `serve` and `finish` read the finished flag under the in-flight lock, so nothing is handed out or kept once `finish` returned. A pulled configure that is refused after the opt-out finishes its queue, so `next` no longer waits forever. Platform log lines passed to `telemetry_log` or `TelemetryScope.log` get the same credential masking as the core's own copied lines. Its test reads the body uploaded through `TelemetryScope.log`, so it fails if that call drops the mask (`flate2` is a new dev-dependency for it). The pull queue's docs sit on the queue, it has no foreign constructor, and its fields and the span methods document their edge cases. `Pending` is `QueuedExport` and `mask_jwts` is `mask_credentials`. The forwarder maps levels with a new `From` for `Severity`. The module has its license header, the SPEC states the real masking rules, the example uses the workspace edition, the stray `.gitignore` entry is gone, and the `livekit-ffi` changeset names the `log_forward_bootstrap` change. A `livekit-telemetry` changeset records the new conversion and `global::uninstall`. --- .../telemetry-severity-from-log-level.md | 5 + .changeset/uniffi-telemetry.md | 2 +- .gitignore | 1 - Cargo.lock | 1 + examples/telemetry_ping/Cargo.toml | 2 +- livekit-ffi/Cargo.toml | 1 + livekit-ffi/src/core_modules/log_forward.rs | 23 +- livekit-ffi/src/core_modules/telemetry.rs | 1093 +++++++++++++++-- livekit-telemetry/SPEC.md | 13 +- livekit-telemetry/src/event.rs | 13 + livekit-telemetry/src/global.rs | 6 +- 11 files changed, 1017 insertions(+), 143 deletions(-) create mode 100644 .changeset/telemetry-severity-from-log-level.md diff --git a/.changeset/telemetry-severity-from-log-level.md b/.changeset/telemetry-severity-from-log-level.md new file mode 100644 index 000000000..39c295252 --- /dev/null +++ b/.changeset/telemetry-severity-from-log-level.md @@ -0,0 +1,5 @@ +--- +livekit-telemetry: minor +--- + +Convert a `log::Level` into a telemetry `Severity` with `From`, and remove the process pipeline without draining it with `global::uninstall`. diff --git a/.changeset/uniffi-telemetry.md b/.changeset/uniffi-telemetry.md index f1abfab0c..d26a0df49 100644 --- a/.changeset/uniffi-telemetry.md +++ b/.changeset/uniffi-telemetry.md @@ -2,4 +2,4 @@ livekit-ffi: minor --- -Expose the client telemetry core over UniFFI: `telemetry_configure` / `telemetry_configure_pulled` (a bounded pull queue for bindings with thread-bound callbacks, served with `next` or polled with `try_next`), `telemetry_scope` with `TelemetryScope.set_server` (connect and every token refresh), spans, the subscribe lifecycle, `record_peer_stats` + `stats_poll_interval_ms`, `track_ended`, `emit_custom` / `set_attribute` (the Room-scoped app API), `telemetry_disable` (a synchronous opt-out, in effect when it returns), device state and log forwarding, with a host-implemented `TelemetryTransport` or the `livekit-net` HTTP client. +Expose the client telemetry core through the `core-modules` UniFFI surface (`telemetry_configure`, `telemetry_configure_pulled`, `telemetry_scope`, spans, stats and `telemetry_disable`); `log_forward_bootstrap` now copies the core's own warnings and errors into telemetry and keeps the Rust log level at `warn` or looser. diff --git a/.gitignore b/.gitignore index fbae334f9..328df4d65 100644 --- a/.gitignore +++ b/.gitignore @@ -6,4 +6,3 @@ soxr-sys/test-output.wav .env .cursor __pycache__ -/target-*/ diff --git a/Cargo.lock b/Cargo.lock index 0df7e69c4..91650ba92 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3906,6 +3906,7 @@ dependencies = [ "device-info", "downcast-rs", "env_logger 0.11.11", + "flate2", "from_variants", "futures-util", "imgproc", diff --git a/examples/telemetry_ping/Cargo.toml b/examples/telemetry_ping/Cargo.toml index dee6f7709..199f067b7 100644 --- a/examples/telemetry_ping/Cargo.toml +++ b/examples/telemetry_ping/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "telemetry_ping" version = "0.1.0" -edition = "2021" +edition.workspace = true publish = false [dependencies] diff --git a/livekit-ffi/Cargo.toml b/livekit-ffi/Cargo.toml index bb6cc321a..7cfc86136 100644 --- a/livekit-ffi/Cargo.toml +++ b/livekit-ffi/Cargo.toml @@ -130,6 +130,7 @@ prost-build = { workspace = true, optional = true } webrtc-sys-build = { workspace = true, optional = true } [dev-dependencies] +flate2 = "1" livekit-token = { workspace = true } serial_test = "3.0" diff --git a/livekit-ffi/src/core_modules/log_forward.rs b/livekit-ffi/src/core_modules/log_forward.rs index fd3ed751c..32136a3b1 100644 --- a/livekit-ffi/src/core_modules/log_forward.rs +++ b/livekit-ffi/src/core_modules/log_forward.rs @@ -134,19 +134,17 @@ impl Log for Logger { /// Never the telemetry crate's own records (`livekit_telemetry*`): a record about a failed /// upload must not produce another upload. The pipeline applies the same guard again. fn telemetry_copy(record: &Record) -> Option { - let severity = match record.level() { - Level::Error => livekit_telemetry::Severity::Error, - Level::Warn => livekit_telemetry::Severity::Warn, - _ => return None, - }; + if record.level() > Level::Warn { + return None; + } let target = record.target(); if !target.starts_with("livekit") || target.starts_with("livekit_telemetry") { return None; } Some(livekit_telemetry::LogRecord { - severity, + severity: record.level().into(), source: livekit_telemetry::LogSource::Ffi, - body: mask_jwts(&record.args().to_string()), + body: mask_credentials(&record.args().to_string()), logger: Some(target.to_owned()), // `log` knows no function name; the module path is the target, already in `logger`. function: None, @@ -162,7 +160,7 @@ fn telemetry_copy(record: &Record) -> Option { /// compact JWT — three base64url segments whose header and payload decode to JSON objects, /// whatever their formatting or the punctuation around them — becomes ``. A participant /// token quoted in an error must not leave the device. -fn mask_jwts(body: &str) -> String { +pub(crate) fn mask_credentials(body: &str) -> String { let body = mask_credential_values(body); let jwt_char = |c: char| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.'); let mut out = String::with_capacity(body.len()); @@ -220,6 +218,7 @@ fn json_object(segment: &str) -> bool { }; let (mut acc, mut bits) = (0u32, 0); for c in segment.bytes() { + // Keep the 12 bits that can still hold an undecoded byte. acc = (acc << 6 | u32::from(sextet(c))) & 0xFFF; bits += 6; if bits >= 8 { @@ -354,7 +353,7 @@ mod tests { "signal transport error: HTTP 401 for request with Authorization: Bearer ; \ token . retrying" ); - assert_eq!(mask_jwts(r#"{"token":"abc.def"}"#), r#"{"token":""}"#); + assert_eq!(mask_credentials(r#"{"token":"abc.def"}"#), r#"{"token":""}"#); // A JWT inside a punctuation-delimited run, and credential values that are quoted or // spaced. for (logged, masked) in [ @@ -373,10 +372,10 @@ mod tests { ), ("ACCESS_TOKEN = opaque-secret&x".into(), "ACCESS_TOKEN = &x".into()), ] { - assert_eq!(mask_jwts(&logged), masked, "{logged}"); + assert_eq!(mask_credentials(&logged), masked, "{logged}"); } let unsigned = "eyAiYWxnIjoiSFMyNTYiLCJ0eXAiOiJKV1QifQ.CiAgeyJzdWIiOiJib2IifQ."; - assert_eq!(mask_jwts(unsigned), "", "an unsigned token is masked too"); + assert_eq!(mask_credentials(unsigned), "", "an unsigned token is masked too"); for kept in [ "eyJ", "eyJabc.", @@ -388,7 +387,7 @@ mod tests { "token expired, tokens: 3, token_type=x", "(...eyJ)", ] { - assert_eq!(mask_jwts(kept), kept); + assert_eq!(mask_credentials(kept), kept); } } diff --git a/livekit-ffi/src/core_modules/telemetry.rs b/livekit-ffi/src/core_modules/telemetry.rs index 30cdab7be..ca32e4571 100644 --- a/livekit-ffi/src/core_modules/telemetry.rs +++ b/livekit-ffi/src/core_modules/telemetry.rs @@ -1,3 +1,17 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + //! Client telemetry core from the [`livekit-telemetry`] crate. //! //! What a platform calls, and when: @@ -32,9 +46,9 @@ use livekit_telemetry::{ SpanName, SpanOutcome, SpanStep, SpanTrack, StreamDirection, TelemetryConfig, TelemetryEvent, TelemetryStats, TelemetryTransport, TraceContext, TrackKind, }; -use tokio::sync::{mpsc, oneshot, watch}; +use tokio::sync::{mpsc, oneshot}; -/// Why a [`Telemetry`] pipeline could not be created. +/// Why a [`livekit_telemetry::Telemetry`] pipeline could not be created. #[derive(uniffi::Error, thiserror::Error, Debug)] #[uniffi(flat_error)] pub enum TelemetryError { @@ -45,7 +59,8 @@ pub enum TelemetryError { /// Start the process pipeline; a previous one is drained and replaced. `transport = None` uses /// the HTTP client registered with `livekit-net`, if any. After [`telemetry_disable`] nothing -/// starts, and whatever `config.storage_dir` still holds is deleted. +/// starts, and a configure that gets as far as installing deletes whatever `config.storage_dir` +/// still holds. #[uniffi::export] pub fn telemetry_configure( config: TelemetryConfig, @@ -56,7 +71,7 @@ pub fn telemetry_configure( Some(transport) => transport, None => Arc::new(NetTransport::from_registry().ok_or(TelemetryError::NoTransport)?), }; - install(livekit_telemetry::Telemetry::new(config, transport), instruments); + install(config, transport, instruments); Ok(()) } @@ -72,32 +87,315 @@ pub fn telemetry_configure_pulled( ) -> Arc { let queue = TelemetryExportQueue::new(); queue.process.store(true, Ordering::SeqCst); - install(livekit_telemetry::Telemetry::new(config, queue.clone()), instruments); + if !install(config, queue.clone(), instruments) { + // No exporter will ever queue a request: `next` must not wait for one. + queue.finish(); + } queue } +/// Install the pipeline through the process lifecycle; `false` when the install was refused +/// (after the opt-out) and no exporter runs. fn install( - (telemetry, exporter): (livekit_telemetry::Telemetry, livekit_telemetry::Exporter), + config: TelemetryConfig, + transport: Arc, instruments: Vec>, -) { - let previous = global::install(telemetry, instruments); - // Refused after the opt-out (the new pipeline was purged instead, its `storage_dir` with it), - // or opted out meanwhile (every generation revoked): nothing to run, nothing to drain. - if global::is_disabled() { - return; +) -> bool { + let (telemetry, exporter) = livekit_telemetry::Telemetry::new(config, transport); + configure_in(&LIFECYCLE, telemetry, exporter, move |telemetry| { + // The pipeline it replaces is the one the lifecycle runs, and is stopped there in turn. + global::install(telemetry, instruments); + // Refused after the opt-out (the new pipeline was purged instead, its `storage_dir` with + // it), or opted out meanwhile (every generation revoked): nothing to run. + !global::is_disabled() + }) +} + +/// Have the lifecycle in `slot` install `telemetry` with `install` and run its exporter in turn, +/// and wait until it installed it: `false`, and no exporter, when `install` refused it. The +/// lifecycle installs on its own thread, so that is where the instruments start; it answers +/// before any drain (its exporter runner drains), so this waits only for the commands before it. +fn configure_in( + slot: &LifecycleSlot, + telemetry: livekit_telemetry::Telemetry, + exporter: livekit_telemetry::Exporter, + install: impl FnOnce(livekit_telemetry::Telemetry) -> bool + Send + 'static, +) -> bool { + // A blocking wait: `telemetry_configure` is synchronous. + let (done, installed) = std::sync::mpsc::sync_channel(1); + let exporter = Box::new(exporter); + slot.send(Command::Configure { install: Box::new(install), telemetry, exporter, done }); + installed.recv().unwrap_or(false) +} + +/// One lifecycle operation. A lifecycle carries them out one at a time, in the order sent. +enum Command { + /// Install a pipeline with `install` and run its exporter once the one before it stopped; + /// answers whether it was installed. + Configure { + install: Box bool + Send>, + telemetry: livekit_telemetry::Telemetry, + /// Boxed: an exporter is large. + exporter: Box, + done: std::sync::mpsc::SyncSender, + }, + /// Uninstall with `uninstall`, then answer once every exporter queued before has stopped. + Shutdown { uninstall: Box, done: oneshot::Sender<()> }, + /// The opt-out's purge, revoked already: the commands after it wait for it. + Purge(std::pin::Pin + Send>>), + /// Answers once every command before it is done. + Barrier(oneshot::Sender<()>), +} + +/// Where a lifecycle lives: the running one, held by its owner, if any. +struct LifecycleSlot { + lifecycle: Mutex>, + /// Lifecycles started in this slot so far: each new one's generation is the next number. + generations: std::sync::atomic::AtomicU64, + /// The runtime it runs on instead of the bridge's (tests only: so they can count its tasks). + #[cfg(test)] + runtime: Option, +} + +/// The process pipeline's lifecycle. +static LIFECYCLE: LifecycleSlot = LifecycleSlot::new(); + +impl LifecycleSlot { + const fn new() -> Self { + Self { + lifecycle: Mutex::new(None), + generations: std::sync::atomic::AtomicU64::new(0), + #[cfg(test)] + runtime: None, + } + } + + fn lock(&self) -> std::sync::MutexGuard<'_, Option> { + self.lifecycle.lock().unwrap_or_else(|e| e.into_inner()) + } + + /// Send `command`, starting a lifecycle when none runs; returns its ticket. + fn send(&self, command: Command) -> Ticket { + let mut lifecycle = self.lock(); + lifecycle.get_or_insert_with(|| self.spawn()).send(command) + } + + /// Send `command` to the running lifecycle; `None`, and nothing sent, when none runs. + fn send_if_running(&self, command: Command) -> Option { + self.lock().as_mut().map(|lifecycle| lifecycle.send(command)) + } + + /// A new lifecycle, of the next generation, on the bridge runtime. + fn spawn(&self) -> LifecycleHandle { + let generation = self.generations.fetch_add(1, Ordering::SeqCst) + 1; + #[cfg(test)] + if let Some(runtime) = &self.runtime { + return Lifecycle::spawn(runtime, generation); + } + Lifecycle::spawn(crate::runtime::runtime().handle(), generation) + } + + /// Take the lifecycle out for its owner to end, only when the slot still holds the lifecycle + /// `ticket` went to and that command is still its last one. A ticket of an earlier lifecycle + /// never matches a later one, whatever its command number. + fn take_after(&self, ticket: Ticket) -> Option { + self.lock().take_if(|lifecycle| lifecycle.ticket() == ticket) + } +} + +/// A running lifecycle, as its owner holds it. +struct LifecycleHandle { + commands: mpsc::UnboundedSender, + task: Owned, + /// Which lifecycle of its slot this is: no other lifecycle of the slot has the same one. + generation: u64, + /// Commands sent so far. A shutdown ends this lifecycle only when the slot still holds it + /// and the shutdown's command is still the last one sent (see [`LifecycleSlot::take_after`]). + sent: u64, +} + +/// A sent command: the generation of the lifecycle it went to, and its number there. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct Ticket { + generation: u64, + command: u64, +} + +/// How long a lifecycle may take to end once its commands are gone: it has stopped every exporter +/// by then, so it ends at once, and this bound only guards against a stuck one. +const END_BOUND: std::time::Duration = std::time::Duration::from_secs(1); + +impl LifecycleHandle { + /// Send `command`; returns its ticket. + fn send(&mut self, command: Command) -> Ticket { + self.sent += 1; + // A lifecycle ends only once its owner dropped this sender: the command arrives. + let _ = self.commands.send(command); + self.ticket() + } + + /// The ticket of the last command sent. + fn ticket(&self) -> Ticket { + Ticket { generation: self.generation, command: self.sent } + } + + /// End the lifecycle and join it: without commands it ends its exporter runner and returns. + /// Past [`END_BOUND`] it is cancelled and joined all the same. + async fn end(self) { + drop(self.commands); + self.task.join(END_BOUND, async {}).await; + } +} + +/// A task its owner joins. Dropped unjoined (its owner was cancelled), it is cancelled too, never +/// left running on its own. +struct Owned(tokio::task::JoinHandle<()>); + +impl Owned { + /// Run `first`, then wait for the task, both within `bound`; past it, cancel the task and + /// wait until it is gone. + async fn join(mut self, bound: std::time::Duration, first: impl std::future::Future) { + let joined = tokio::time::timeout(bound, async { + first.await; + let _ = (&mut self.0).await; + }) + .await; + if joined.is_err() { + self.0.abort(); + let _ = (&mut self.0).await; + } + } +} + +impl Drop for Owned { + fn drop(&mut self) { + self.0.abort(); + } +} + +/// The bridge's lifecycle owner: carries out every lifecycle operation (install, uninstall, the +/// opt-out's purge) in the order its commands were sent, so a configure and a shutdown never +/// interleave. Each installed pipeline's exporter goes to its exporter runner, which starts one +/// only once the one before it stopped; the lifecycle never waits for a drain itself, so a +/// configure waits only for the commands before it. +struct Lifecycle { + commands: mpsc::UnboundedReceiver, + exporters: mpsc::UnboundedSender, +} + +impl Lifecycle { + /// Start a lifecycle and its exporter runner on `runtime`. The channels are unbounded: + /// `telemetry_configure` and `telemetry_disable` are synchronous and must neither block nor + /// fail to queue, and a command stands for one such call. + fn spawn(runtime: &tokio::runtime::Handle, generation: u64) -> LifecycleHandle { + let (sender, commands) = mpsc::unbounded_channel(); + let task = runtime.spawn(async move { + let (exporters, steps) = mpsc::unbounded_channel(); + let runner = Owned(tokio::spawn(Exporters { steps, running: None }.run())); + Self { commands, exporters }.run(runner).await; + }); + LifecycleHandle { commands: sender, task: Owned(task), generation, sent: 0 } + } + + async fn run(mut self, runner: Owned) { + while let Some(command) = self.commands.recv().await { + match command { + Command::Configure { install, telemetry, exporter, done } => { + let installed = install(telemetry.clone()); + if installed { + let _ = self.exporters.send(Step::Run(telemetry, exporter)); + } else { + // Gone before the answer: a refused pulled queue is held by its caller only. + drop((telemetry, exporter)); + } + let _ = done.send(installed); + } + Command::Shutdown { uninstall, done } => { + uninstall(); + let _ = self.exporters.send(Step::Stop(done)); + } + Command::Purge(purge) => purge.await, + Command::Barrier(done) => { + let _ = done.send(()); + } + } + } + // No command can come any more: the runner stops what it runs and ends. + drop(self.exporters); + runner.join(END_BOUND, async {}).await; } - crate::runtime::runtime().spawn(exporter.run()); - if let Some(previous) = previous { - crate::runtime::runtime().spawn(async move { previous.shutdown().await }); +} + +/// One request to the [`Exporters`] runner. +enum Step { + /// Stop the running exporter, then run this pipeline's. + Run(livekit_telemetry::Telemetry, Box), + /// Stop the running exporter, then answer. + Stop(oneshot::Sender<()>), +} + +/// A lifecycle's exporter runner: one exporter at a time, in the order the steps come. A +/// pipeline's exporter starts only once the one before it is gone, drained or else cancelled and +/// joined at its bound, so two exporters never replay a shared cache at the same time and none +/// outlives its stop. New captures wait in the new pipeline's queue meanwhile. +struct Exporters { + steps: mpsc::UnboundedReceiver, + /// The running exporter's pipeline and task. + running: Option<(livekit_telemetry::Telemetry, Owned)>, +} + +impl Exporters { + async fn run(mut self) { + while let Some(step) = self.steps.recv().await { + self.stop().await; + match step { + Step::Run(telemetry, exporter) => { + self.running = Some((telemetry, Owned(tokio::spawn(exporter.run())))); + } + Step::Stop(done) => { + let _ = done.send(()); + } + } + } + self.stop().await; + } + + /// Drain the running pipeline and join its exporter, within the pipeline's shutdown bound. + /// Past it the exporter is cancelled and joined all the same: dropping its handle would + /// leave it running. + async fn stop(&mut self) { + let Some((telemetry, task)) = self.running.take() else { return }; + task.join(telemetry.shutdown_bound(), telemetry.shutdown()).await; } } /// Drain and stop the process pipeline; every call is a no-op afterwards. After -/// [`telemetry_disable`], returns once its purge is complete. +/// [`telemetry_disable`], returns once its purge is complete. Returns once every exporter +/// started before the call has stopped as well: each drains within its pipeline's shutdown bound +/// (its export timeout plus 1 s) or is cancelled there, so right after a replace this waits one +/// bound for each pipeline still in line. A shutdown that is the last call ends the lifecycle +/// and joins it; the next configure starts a fresh one. #[uniffi::export(async_runtime = "tokio")] pub async fn telemetry_shutdown() { - purged().await; - global::shutdown().await; + shutdown_in(&LIFECYCLE, || { + // Its exporter is stopped by the lifecycle, in turn. + global::uninstall(); + }) + .await; +} + +/// Have the lifecycle in `slot` uninstall with `uninstall` and stop every exporter queued before, +/// and wait for that. Then, when the slot still holds that same lifecycle and this was its last +/// command, end the lifecycle and join it. +async fn shutdown_in(slot: &LifecycleSlot, uninstall: impl FnOnce() + Send + 'static) { + let (done, stopped) = oneshot::channel(); + let command = Command::Shutdown { uninstall: Box::new(uninstall), done }; + // No lifecycle: nothing was installed through one, nothing runs. + let Some(ticket) = slot.send_if_running(command) else { return }; + let _ = stopped.await; + if let Some(lifecycle) = slot.take_after(ticket) { + lifecycle.end().await; + } } /// Opt-out for the rest of the process, in effect when this returns: no scope is handed out, @@ -111,7 +409,11 @@ pub async fn telemetry_shutdown() { /// call back into configure, disable, flush or shutdown. #[uniffi::export] pub fn telemetry_disable() { - start_purge(&PURGE, global::disable()); + // The revocation holds on return; the purge then runs on the lifecycle, in turn. + let purge = global::disable(); + LIFECYCLE.send(Command::Purge(Box::pin(async move { + purge.await; + }))); } /// Whether the process opted out with [`telemetry_disable`]: `true` for every thread and isolate @@ -122,36 +424,12 @@ pub fn telemetry_is_disabled() -> bool { global::is_disabled() } -/// The opt-out's purge in progress: done once it holds `true`. -type PurgeSlot = Mutex>>; - -/// The process-wide opt-out's purge. -static PURGE: PurgeSlot = Mutex::new(None); - -/// Run `purge` in the background and record it in `slot`, done only once every purge recorded -/// there before it is done too: a repeated opt-out never lets a waiter skip the first purge. -fn start_purge(slot: &PurgeSlot, purge: impl std::future::Future + Send + 'static) { - let (done, receiver) = watch::channel(false); - let previous = slot.lock().unwrap_or_else(|e| e.into_inner()).replace(receiver); - crate::runtime::runtime().spawn(async move { - purge.await; - if let Some(mut previous) = previous { - let _ = previous.wait_for(|done| *done).await; - } - let _ = done.send(true); - }); -} - -/// Wait for the opt-out's purge, if one was started. -async fn purged() { - wait_purge(&PURGE).await; -} - -/// Wait for the purge recorded in `slot`, if any. -async fn wait_purge(slot: &PurgeSlot) { - let receiver = slot.lock().unwrap_or_else(|e| e.into_inner()).clone(); - if let Some(mut receiver) = receiver { - let _ = receiver.wait_for(|done| *done).await; +/// Wait until every command sent to the lifecycle in `slot` so far is done; at once when none +/// runs. +async fn barrier(slot: &LifecycleSlot) { + let (done, passed) = oneshot::channel(); + if slot.send_if_running(Command::Barrier(done)).is_some() { + let _ = passed.await; } } @@ -159,7 +437,7 @@ async fn wait_purge(slot: &PurgeSlot) { /// returns once its purge is complete. #[uniffi::export(async_runtime = "tokio")] pub async fn telemetry_flush() { - purged().await; + barrier(&LIFECYCLE).await; global::flush().await; } @@ -177,10 +455,19 @@ pub fn telemetry_emit(event: TelemetryEvent) { /// A captured log line; the core applies the per-source floor and builds the record. For the /// platform's own and WebRTC's lines only — never a Rust entry received through the log -/// forwarder, which the core has already copied (`log_forward_bootstrap`). +/// forwarder, which the core has already copied (`log_forward_bootstrap`). Credentials in the +/// line (a compact JWT, the value after `Bearer` or a `token` key) are masked, as in the core's +/// own copied lines. #[uniffi::export] pub fn telemetry_log(record: LogRecord) { - global::log(record); + global::log(masked(record)); +} + +/// `record` with its credentials masked: a token quoted in a platform error never leaves the +/// device. +fn masked(mut record: LogRecord) -> LogRecord { + record.body = crate::log_forward::mask_credentials(&record.body); + record } /// Audio route, interruption, denied permission: a process-level record built by the core. @@ -196,7 +483,9 @@ pub fn telemetry_set_device_state(state: DeviceState) { } /// SDK metadata added at export to every record of every scope (not an app API: app -/// attributes are room-scoped, [`TelemetryScope::set_attribute`]); `None` removes it. +/// attributes are room-scoped, [`TelemetryScope::set_attribute`]); `None` removes it. Keys up to +/// 128 bytes, values up to 1024, at most 64 keys, never `session.id`, `error.type` or +/// `lk.outcome`; anything else is rejected and counted. #[uniffi::export] pub fn telemetry_set_attribute(key: String, value: Option) { global::set_attribute(&key, value); @@ -239,8 +528,8 @@ impl TelemetryScope { /// An app event, exported as `custom.` with the room's correlation attributes. Names /// and keys up to 128 bytes, values up to 1024, at most 64 attributes, none of the SDK's - /// keys (`lk.*`, `otel.*`, `code.*`, `session.id`, `error.type`); anything else is rejected - /// and counted. + /// keys (`lk.*`, `otel.*`, `code.*`, `session.id`, `error.type`, `network.connection.type`); + /// anything else is rejected and counted. pub fn emit_custom(&self, name: String, attributes: HashMap) { let attributes = attributes.into_iter().map(|(k, v)| Attribute::new(k, v)).collect(); self.0.emit_custom(&name, attributes); @@ -278,14 +567,16 @@ impl TelemetryScope { self.0.set_room(room); } - /// The session ended for good (not a reconnect). + /// The session ended for good (not a reconnect): `lk.room.disconnected`, once; a repeated call + /// is a no-op until the session starts a new `lk.connect`. pub fn disconnected(&self, reason: DisconnectReason) { self.0.disconnected(reason); } - /// A log record filed under this session even without an ambient span. + /// A log record filed under this session even without an ambient span. Credentials in it are + /// masked, as with [`telemetry_log`]. pub fn log(&self, record: LogRecord) { - self.0.log(record); + self.0.log(masked(record)); } /// One track's whole `getStats()` report; the core maps it (see `record_stats`). @@ -328,36 +619,33 @@ impl TelemetryScope { /// The protocol's `DisconnectReason` number as the shared enum. #[uniffi::export] pub fn telemetry_disconnect_reason(proto: i32) -> DisconnectReason { - DisconnectReason::from_proto(proto) + DisconnectReason::from(proto) } /// The protocol's `ReconnectReason` number (`RR_*`) as the shared enum. #[uniffi::export] pub fn telemetry_reconnect_reason(proto: i32) -> ReconnectReason { - ReconnectReason::from_proto(proto) + ReconnectReason::from(proto) } /// One export the host has to perform on behalf of a pulled pipeline. #[derive(uniffi::Record)] pub struct PendingExport { + /// What [`TelemetryExportQueue::complete`] and [`TelemetryExportQueue::fail`] take to answer + /// this request. pub id: u64, + /// The request to POST, exactly as given. pub request: ExportRequest, } -struct Pending { +/// A queued export and the channel its exporter awaits the answer on. +struct QueuedExport { + /// What the host is handed. export: PendingExport, + /// The exporter's end: closed once it gave up on the request. done: oneshot::Sender>, } -/// Pull-side transport: Rust never calls into the host. The exporter queues each request; the -/// host awaits [`next`](Self::next) (a Rust future — those cross every binding) or polls -/// [`try_next`](Self::try_next), performs the HTTP call on its own thread, and reports the -/// outcome with [`complete`](Self::complete), -/// which unblocks the exporter's retry/drop/go-silent logic exactly as a direct transport would. -/// -/// Exists because uniffi-dart's foreign-trait callbacks are isolate-bound (`Pointer.fromFunction`) -/// and abort the VM when invoked from a tokio thread; Swift and Kotlin callbacks are thread-agnostic -/// and use [`TelemetryTransport`] directly. /// One attempt at an SDK operation, owned by the core. Every call is synchronous and stamps the /// clock inside, so the only skew is the FFI call; `describe()` is the console line on every /// platform. Detached (no session) it still times and describes itself. @@ -366,12 +654,12 @@ pub struct TelemetrySpan(Arc); #[uniffi::export] impl TelemetrySpan { - /// A checkpoint, stamped now. + /// A checkpoint, stamped now. Ignored once the span has ended. pub fn step(&self, step: SpanStep) { self.0.step(step); } - /// The open bag; replaces an existing key. + /// The open bag; replaces an existing key. An app-defined span refuses the SDK's keys. pub fn set_attribute(&self, key: String, value: AttributeValue) { self.0.set_attribute(key, value); } @@ -381,7 +669,8 @@ impl TelemetrySpan { self.0.set_track(track); } - /// End once; `error` becomes `error.type` when the outcome is `Error`. + /// End once; every later call is ignored. `error` becomes `error.type` when the outcome is + /// `Error`. pub fn end(&self, outcome: SpanOutcome, error: Option) { self.0.end(outcome, error); } @@ -391,7 +680,7 @@ impl TelemetrySpan { self.0.fail(error); } - /// End as cancelled. + /// End as cancelled: the status stays unset and `lk.outcome` is `cancelled`. pub fn cancel(&self) { self.0.cancel(); } @@ -417,50 +706,55 @@ impl TelemetrySpan { } } -/// The pull queue's serving side (see its type docs above [`TelemetrySpan`]). +/// Pull-side transport: Rust never calls into the host. The exporter queues each request; the +/// host awaits [`next`](Self::next) (a Rust future — those cross every binding) or polls +/// [`try_next`](Self::try_next), performs the HTTP call on its own thread, and reports the +/// outcome with [`complete`](Self::complete), which unblocks the exporter's retry/drop/go-silent +/// logic exactly as a direct transport would. +/// +/// Exists because uniffi-dart's foreign-trait callbacks are isolate-bound (`Pointer.fromFunction`) +/// and abort the VM when invoked from a tokio thread; Swift and Kotlin callbacks are thread-agnostic +/// and use [`TelemetryTransport`] directly. Only [`telemetry_configure_pulled`] makes one. #[derive(uniffi::Object)] pub struct TelemetryExportQueue { /// Bounded to one request: the exporter has one request out at a time, so nothing piles up /// behind a host that stopped serving. `None` once finished. - tx: Mutex>>, - rx: tokio::sync::Mutex>, + tx: Mutex>>, + /// The requests waiting for the host. + rx: tokio::sync::Mutex>, + /// The requests handed out and not answered yet, by id. `serve` and `finish` read `finished` + /// under this lock, so nothing is handed out once `finish` returned. inflight: Mutex>>>, + /// The next request id. seq: AtomicU64, + /// `finish` was called: nothing is served from now on. finished: AtomicBool, /// Serves the process pipeline (`telemetry_configure_pulled`): nothing is handed out once - /// the app opted out, even before the purge has cancelled the request. + /// the app opted out, even before the purge has cancelled the request. `false` only for the + /// unit tests' own queues. process: AtomicBool, /// Requests handed to the queue so far (tests only). #[cfg(test)] queued: AtomicU64, + /// Runs inside `serve` between its checks and the hand-out (tests only: holds a serve there + /// while `finish` races it). + #[cfg(test)] + before_serve: Mutex>>, } #[uniffi::export(async_runtime = "tokio")] impl TelemetryExportQueue { - /// An empty queue; `telemetry_configure_pulled` makes the one it serves. - #[uniffi::constructor] - pub fn new() -> Arc { - let (tx, rx) = mpsc::channel(1); - Arc::new(Self { - tx: Mutex::new(Some(tx)), - rx: tokio::sync::Mutex::new(rx), - inflight: Mutex::new(HashMap::new()), - seq: AtomicU64::new(0), - finished: AtomicBool::new(false), - process: AtomicBool::new(false), - #[cfg(test)] - queued: AtomicU64::new(0), - }) - } - /// End the serving loop: `next` resolves `None` from now on, and whatever is still queued is /// discarded, never served. Call after `telemetry_shutdown` or `telemetry_disable`, or when a /// later `telemetry_configure_pulled` replaced this queue. /// (Not `close`: UniFFI's Kotlin objects already have `AutoCloseable.close`.) pub fn finish(&self) { - self.finished.store(true, Ordering::SeqCst); + { + let mut inflight = self.inflight.lock().unwrap_or_else(|e| e.into_inner()); + self.finished.store(true, Ordering::SeqCst); + inflight.clear(); + } self.tx.lock().unwrap_or_else(|e| e.into_inner()).take(); - self.inflight.lock().unwrap_or_else(|e| e.into_inner()).clear(); // Dropping what is queued tells its exporter the host dropped it. if let Ok(mut rx) = self.rx.try_lock() { while rx.try_recv().is_ok() {} @@ -474,8 +768,8 @@ impl TelemetryExportQueue { if self.finished.load(Ordering::SeqCst) { return None; } - let pending = self.rx.lock().await.recv().await?; - if let Some(export) = self.serve(pending) { + let queued = self.rx.lock().await.recv().await?; + if let Some(export) = self.serve(queued) { return Some(export); } if self.finished.load(Ordering::SeqCst) { @@ -500,28 +794,51 @@ impl TelemetryExportQueue { } } - /// The collector's answer to the request with `id`, whatever its status; the core classifies it. + /// The collector's answer to the request with `id`, whatever its status; the core classifies + /// it. An unknown or stale `id` (already answered, or given up on) is ignored. pub fn complete(&self, id: u64, response: ExportResponse) { self.settle(id, Ok(response)); } - /// The request with `id` got no answer (network error, timeout, invalid URL). + /// The request with `id` got no answer (network error, timeout, invalid URL). An unknown or + /// stale `id` is ignored. pub fn fail(&self, id: u64, error: ExportError) { self.settle(id, Err(error)); } } impl TelemetryExportQueue { - /// Hand `pending` out, unless the queue is finished or its exporter gave up on it. - fn serve(&self, pending: Pending) -> Option { + /// An empty queue. + fn new() -> Arc { + let (tx, rx) = mpsc::channel(1); + Arc::new(Self { + tx: Mutex::new(Some(tx)), + rx: tokio::sync::Mutex::new(rx), + inflight: Mutex::new(HashMap::new()), + seq: AtomicU64::new(0), + finished: AtomicBool::new(false), + process: AtomicBool::new(false), + #[cfg(test)] + queued: AtomicU64::new(0), + #[cfg(test)] + before_serve: Mutex::new(None), + }) + } + + /// Hand `queued` out, unless the queue is finished or its exporter gave up on it. + fn serve(&self, queued: QueuedExport) -> Option { + let mut inflight = self.inflight.lock().unwrap_or_else(|e| e.into_inner()); let opted_out = self.process.load(Ordering::SeqCst) && global::is_disabled(); - if opted_out || self.finished.load(Ordering::SeqCst) || pending.done.is_closed() { + if opted_out || self.finished.load(Ordering::SeqCst) || queued.done.is_closed() { return None; } - let mut inflight = self.inflight.lock().unwrap_or_else(|e| e.into_inner()); + #[cfg(test)] + if let Some(hook) = self.before_serve.lock().unwrap_or_else(|e| e.into_inner()).as_ref() { + hook(); + } inflight.retain(|_, done| !done.is_closed()); - inflight.insert(pending.export.id, pending.done); - Some(pending.export) + inflight.insert(queued.export.id, queued.done); + Some(queued.export) } fn settle(&self, id: u64, outcome: Result) { @@ -536,16 +853,16 @@ impl TelemetryExportQueue { fn discard_stale(&self) { let Ok(mut rx) = self.rx.try_lock() else { return }; let mut live = Vec::new(); - while let Ok(pending) = rx.try_recv() { - if !pending.done.is_closed() { - live.push(pending); + while let Ok(queued) = rx.try_recv() { + if !queued.done.is_closed() { + live.push(queued); } } drop(rx); let tx = self.tx.lock().unwrap_or_else(|e| e.into_inner()).clone(); - for pending in live { + for queued in live { if let Some(tx) = &tx { - let _ = tx.try_send(pending); + let _ = tx.try_send(queued); } } } @@ -574,14 +891,14 @@ impl TelemetryTransport for TelemetryExportQueue { self.discard_stale(); let id = self.seq.fetch_add(1, Ordering::Relaxed); let (done, wait) = oneshot::channel(); - let pending = Pending { export: PendingExport { id, request }, done }; + let queued = QueuedExport { export: PendingExport { id, request }, done }; let _forget = Forget { queue: self, id }; let Some(tx) = self.tx.lock().unwrap_or_else(|e| e.into_inner()).clone() else { return Err(closed()); }; // Waits for room while the host is busy; cancelled with the exporter's timeout, and then // the request was never queued. The sender is not kept while the answer is awaited. - tx.send(pending).await.map_err(|_| closed())?; + tx.send(queued).await.map_err(|_| closed())?; drop(tx); #[cfg(test)] self.queued.fetch_add(1, Ordering::SeqCst); @@ -694,6 +1011,533 @@ mod tests { }); } + /// `finish` and `serve` take the in-flight lock in turn: once `finish` returned, nothing is + /// handed out or kept, even by a `serve` that passed its checks before. + #[test] + fn nothing_is_served_after_finish_returns() { + let queue = TelemetryExportQueue::new(); + let (entered, inside) = std::sync::mpsc::channel(); + let (resume, resumed) = std::sync::mpsc::channel::<()>(); + let resumed = Mutex::new(resumed); + *queue.before_serve.lock().expect("lock") = Some(Box::new(move || { + let _ = entered.send(()); + let _ = resumed.lock().expect("lock").recv_timeout(Duration::from_millis(200)); + })); + let (done, _answer) = oneshot::channel(); + let queued = QueuedExport { export: PendingExport { id: 7, request: request() }, done }; + let serving = { + let queue = queue.clone(); + std::thread::spawn(move || queue.serve(queued)) + }; + inside.recv().expect("serve passed its checks"); + queue.finish(); + let _ = resume.send(()); + serving.join().expect("serve"); + assert!(queue.inflight.lock().expect("lock").is_empty(), "nothing kept after finish"); + } + + /// A platform's log line loses its credentials before it is captured: the upload carries the + /// mask, never the token. + #[test] + fn platform_log_lines_are_masked() { + let token = granted_token(); + let kept = Arc::new(Gated { + gate: tokio::sync::Semaphore::new(1_000), + bodies: Mutex::new(Vec::new()), + }); + runtime().block_on(async { + let (telemetry, exporter) = + livekit_telemetry::Telemetry::new(TelemetryConfig::default(), kept.clone()); + tokio::spawn(exporter.run()); + let scope = TelemetryScope(telemetry.begin_scope()); + scope.set_server("wss://p.livekit.cloud".into(), token.clone()); + scope.log(LogRecord { + severity: livekit_telemetry::Severity::Warn, + source: livekit_telemetry::LogSource::Sdk, + body: format!("connect failed: wss://p.livekit.cloud/rtc?access_token={token}"), + logger: None, + function: None, + file: None, + line: None, + timestamp_ns: None, + span_id: None, + }); + telemetry.flush().await; + }); + let mut sent = Vec::new(); + for body in kept.bodies.lock().expect("lock").iter() { + let mut gunzip = flate2::read::GzDecoder::new(body.as_slice()); + std::io::Read::read_to_end(&mut gunzip, &mut sent).expect("gzip"); + } + let has = |text: &str| sent.windows(text.len()).any(|w| w == text.as_bytes()); + assert!(has("access_token="), "the line went out, masked"); + assert!(!has(&token), "the token never leaves the device"); + } + + /// A transport that holds every request until the test opens its gate, and keeps the bodies. + struct Gated { + gate: tokio::sync::Semaphore, + bodies: Mutex>>, + } + + #[async_trait::async_trait] + impl TelemetryTransport for Gated { + async fn send(&self, request: ExportRequest) -> Result { + self.bodies.lock().expect("lock").push(request.body); + let _open = self.gate.acquire().await; + Ok(ExportResponse::accepted()) + } + } + + /// The process pipeline's slot, for `configure` and `shut_down`. + type Installed = Mutex>; + + /// What `install` does, with `installed` for the process pipeline: `pause` runs in the + /// lifecycle right after the install. + fn configure( + slot: &LifecycleSlot, + installed: &Arc, + telemetry: livekit_telemetry::Telemetry, + exporter: livekit_telemetry::Exporter, + pause: impl FnOnce() + Send + 'static, + ) { + let installed = installed.clone(); + configure_in(slot, telemetry, exporter, move |telemetry| { + installed.lock().expect("lock").replace(telemetry); + pause(); + true + }); + } + + /// What `telemetry_shutdown` does, with `installed` for the process pipeline. + async fn shut_down(slot: &LifecycleSlot, installed: &Arc) { + let installed = installed.clone(); + shutdown_in(slot, move || { + installed.lock().expect("lock").take(); + }) + .await; + } + + /// What the exporter runner does when it stops a running exporter. + async fn stop_running( + telemetry: livekit_telemetry::Telemetry, + task: tokio::task::JoinHandle<()>, + ) { + let steps = mpsc::unbounded_channel().1; + Exporters { steps, running: Some((telemetry, Owned(task))) }.stop().await; + } + + /// A slot whose lifecycles run on `runtime`, so a test can count their tasks. + fn slot_on(runtime: &tokio::runtime::Runtime) -> LifecycleSlot { + LifecycleSlot { runtime: Some(runtime.handle().clone()), ..LifecycleSlot::new() } + } + + /// The tasks alive on the runtime `slot`'s lifecycles run on. + fn alive_tasks(slot: &LifecycleSlot) -> usize { + let bridge = crate::runtime::runtime().handle(); + slot.runtime.as_ref().unwrap_or(bridge).metrics().num_alive_tasks() + } + + /// Wait until `sent` commands went to the lifecycle in `slot`. + fn wait_until_sent(slot: &LifecycleSlot, sent: u64) { + let reached = || slot.lock().as_ref().is_some_and(|lifecycle| lifecycle.sent == sent); + assert!(eventually(reached), "{sent} commands sent"); + } + + /// A replacing pipeline's exporter starts only once the previous one drained: the two never + /// replay a shared cache at the same time, so a batch in flight is not sent twice. + #[test] + fn a_replacing_exporter_waits_for_the_previous_drain() { + static SLOT: LifecycleSlot = LifecycleSlot::new(); + let installed = Arc::new(Installed::default()); + let dir = std::env::temp_dir().join(format!("lk-hand-over-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + let config = || TelemetryConfig { + storage_dir: Some(dir.to_string_lossy().into_owned()), + ..Default::default() + }; + let transport = Arc::new(Gated { + gate: tokio::sync::Semaphore::new(0), + bodies: Mutex::new(Vec::new()), + }); + let sent = || transport.bodies.lock().expect("lock").clone(); + runtime().block_on(async { + let (first, exporter) = livekit_telemetry::Telemetry::new(config(), transport.clone()); + configure(&SLOT, &installed, first.clone(), exporter, || ()); + let room = first.begin_scope(); + room.set_server("wss://p.livekit.cloud", &granted_token()); + room.emit_custom("ping", Vec::new()); + let flushing = tokio::spawn(async move { first.flush().await }); + while sent().is_empty() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + let in_flight = sent()[0].clone(); + let (second, exporter) = livekit_telemetry::Telemetry::new(config(), transport.clone()); + second.begin_scope().set_server("wss://p.livekit.cloud", &granted_token()); + configure(&SLOT, &installed, second.clone(), exporter, || ()); + // Long enough for a second exporter to replay the cache, were it running. + tokio::time::sleep(Duration::from_millis(100)).await; + transport.gate.add_permits(1_000); + flushing.await.expect("flush"); + second.flush().await; + shut_down(&SLOT, &installed).await; + let copies = sent().iter().filter(|body| **body == in_flight).count(); + assert_eq!(copies, 1, "the batch in flight is sent once"); + }); + let _ = std::fs::remove_dir_all(&dir); + } + + /// Two replaces while the first pipeline drains an upload held for 4.5 s (its export timeout + /// is 20 s): each later exporter starts only once the one before it stopped, however short its + /// own bound, so the batch the first has in flight is sent once. + #[test] + fn two_replaces_within_one_drain_never_run_two_exporters() { + static SLOT: LifecycleSlot = LifecycleSlot::new(); + let installed = Arc::new(Installed::default()); + let dir = std::env::temp_dir().join(format!("lk-hand-over-chain-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + let config = |ms| TelemetryConfig { + storage_dir: Some(dir.to_string_lossy().into_owned()), + export_timeout_ms: ms, + ..Default::default() + }; + let transport = Arc::new(Gated { + gate: tokio::sync::Semaphore::new(0), + bodies: Mutex::new(Vec::new()), + }); + let sent = || transport.bodies.lock().expect("lock").clone(); + runtime().block_on(async { + let (first, exporter) = + livekit_telemetry::Telemetry::new(config(20_000), transport.clone()); + configure(&SLOT, &installed, first.clone(), exporter, || ()); + let room = first.begin_scope(); + room.set_server("wss://p.livekit.cloud", &granted_token()); + room.emit_custom("ping", Vec::new()); + tokio::spawn(async move { first.flush().await }); + while sent().is_empty() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + let in_flight = sent()[0].clone(); + for _ in 0..2 { + let (next, exporter) = + livekit_telemetry::Telemetry::new(config(1_000), transport.clone()); + next.begin_scope().set_server("wss://p.livekit.cloud", &granted_token()); + configure(&SLOT, &installed, next, exporter, || ()); + } + // Past both later pipelines' own bounds (2 s each): long enough for a later exporter + // to replay the cache, were it running beside the first. + tokio::time::sleep(Duration::from_millis(4_500)).await; + transport.gate.add_permits(1_000); + shut_down(&SLOT, &installed).await; + let copies = sent().iter().filter(|body| **body == in_flight).count(); + assert_eq!(copies, 1, "the batch in flight is sent once"); + }); + let _ = std::fs::remove_dir_all(&dir); + } + + /// A shutdown right after a replace returns only once the replacing exporter, which starts + /// after the previous pipeline drained for its whole 10 s, has drained too, although the + /// shutdown's own wait for the replacing pipeline (2 s) ends long before that. + #[test] + fn a_shutdown_waits_for_an_exporter_that_started_late() { + static SLOT: LifecycleSlot = LifecycleSlot::new(); + let installed = Arc::new(Installed::default()); + let gated = Arc::new(Gated { + gate: tokio::sync::Semaphore::new(0), + bodies: Mutex::new(Vec::new()), + }); + let kept = Arc::new(Gated { + gate: tokio::sync::Semaphore::new(1_000), + bodies: Mutex::new(Vec::new()), + }); + let config = |ms| TelemetryConfig { export_timeout_ms: ms, ..Default::default() }; + runtime().block_on(async { + // The first pipeline's upload never answers: it drains for its whole 10 s timeout. + let (first, exporter) = + livekit_telemetry::Telemetry::new(config(10_000), gated.clone()); + configure(&SLOT, &installed, first.clone(), exporter, || ()); + let room = first.begin_scope(); + room.set_server("wss://p.livekit.cloud", &granted_token()); + room.emit_custom("ping", Vec::new()); + tokio::spawn(async move { first.flush().await }); + while gated.bodies.lock().expect("lock").is_empty() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + let (second, exporter) = livekit_telemetry::Telemetry::new(config(1_000), kept.clone()); + let room = second.begin_scope(); + room.set_server("wss://p.livekit.cloud", &granted_token()); + room.emit_custom("pong", Vec::new()); + configure(&SLOT, &installed, second, exporter, || ()); + shut_down(&SLOT, &installed).await; + let sent = kept.bodies.lock().expect("lock").len(); + assert!(sent > 0, "the late exporter drained before the shutdown returned"); + }); + } + + /// Two configures racing, the first paused in the lifecycle right after its install: the + /// exporters still run one at a time and in install order, so the pipeline installed last is + /// the one left exporting. + #[test] + fn racing_configures_run_exporters_in_install_order() { + static SLOT: LifecycleSlot = LifecycleSlot::new(); + /// Requests started, requests out right now, and the most out at once. + #[derive(Default)] + struct Load { + started: AtomicU64, + now: AtomicU64, + most: AtomicU64, + } + /// Counts a request in a load until it ends. + struct Out(Arc); + impl Out { + fn enter(load: &Arc) -> Self { + load.started.fetch_add(1, Ordering::SeqCst); + let now = load.now.fetch_add(1, Ordering::SeqCst) + 1; + load.most.fetch_max(now, Ordering::SeqCst); + Self(load.clone()) + } + } + impl Drop for Out { + fn drop(&mut self) { + self.0.now.fetch_sub(1, Ordering::SeqCst); + } + } + /// Holds every request until it is cancelled, counted in its pipeline's load and in the + /// load of both. + struct Held { + own: Arc, + all: Arc, + } + #[async_trait::async_trait] + impl TelemetryTransport for Held { + async fn send(&self, _request: ExportRequest) -> Result { + let _out = (Out::enter(&self.own), Out::enter(&self.all)); + std::future::pending().await + } + } + let all = Arc::new(Load::default()); + let installed = Arc::new(Installed::default()); + let pipeline = |name: &str| { + let own = Arc::new(Load::default()); + let transport = Arc::new(Held { own: own.clone(), all: all.clone() }); + let config = TelemetryConfig { export_timeout_ms: 1_000, ..Default::default() }; + let (telemetry, exporter) = livekit_telemetry::Telemetry::new(config, transport); + let room = telemetry.begin_scope(); + room.set_server("wss://p.livekit.cloud", &granted_token()); + room.emit_custom(name, Vec::new()); + (telemetry, exporter, own) + }; + let (a, a_exporter, a_load) = pipeline("a"); + let (b, b_exporter, b_load) = pipeline("b"); + let paused = Arc::new(std::sync::Barrier::new(2)); + let first = std::thread::spawn({ + let (installed, paused) = (installed.clone(), paused.clone()); + move || { + configure(&SLOT, &installed, a, a_exporter, move || { + paused.wait(); + std::thread::sleep(Duration::from_millis(300)); + }); + } + }); + paused.wait(); + // While the lifecycle is paused in the first configure, right after its install. + configure(&SLOT, &installed, b, b_exporter, || ()); + first.join().expect("first configure"); + let started = |load: &Load| load.started.load(Ordering::SeqCst); + assert!(eventually(|| started(&b_load) > 0), "the pipeline installed last exports"); + let replaced = started(&a_load); + // Long enough for both exporters to have started, in either order. + std::thread::sleep(Duration::from_secs(3)); + assert_eq!(all.most.load(Ordering::SeqCst), 1, "one exporter at a time"); + assert_eq!(started(&a_load), replaced, "the replaced pipeline sends nothing afterwards"); + runtime().block_on(shut_down(&SLOT, &installed)); + } + + /// Configure and shutdown take effect in the order they were sent. A configure sent while a + /// shutdown is under way comes after it: once the shutdown returns, the configure's pipeline + /// is installed and exporting, and the one before is stopped. + #[test] + fn a_configure_sent_during_a_shutdown_comes_after_it() { + static SLOT: LifecycleSlot = LifecycleSlot::new(); + /// Holds every request until it is cancelled, counting the ones out. + struct Held(Arc); + /// Takes a request out of the count when it ends. + struct Out(Arc); + impl Drop for Out { + fn drop(&mut self) { + self.0.fetch_sub(1, Ordering::SeqCst); + } + } + #[async_trait::async_trait] + impl TelemetryTransport for Held { + async fn send(&self, _request: ExportRequest) -> Result { + self.0.fetch_add(1, Ordering::SeqCst); + let _out = Out(self.0.clone()); + std::future::pending().await + } + } + let installed = Arc::new(Installed::default()); + let (a_out, b_out) = (Arc::new(AtomicU64::new(0)), Arc::new(AtomicU64::new(0))); + let pipeline = |ms, out: &Arc| { + let config = TelemetryConfig { export_timeout_ms: ms, ..Default::default() }; + let transport = Arc::new(Held(out.clone())); + let (telemetry, exporter) = livekit_telemetry::Telemetry::new(config, transport); + let room = telemetry.begin_scope(); + room.set_server("wss://p.livekit.cloud", &granted_token()); + room.emit_custom("pending", Vec::new()); + (telemetry, exporter) + }; + // A's upload never answers: its drain takes its whole 10 s. + let (a, a_exporter) = pipeline(10_000, &a_out); + configure(&SLOT, &installed, a, a_exporter, || ()); + assert!(eventually(|| a_out.load(Ordering::SeqCst) == 1), "A exports"); + // The shutdown is held where it uninstalls until the configure below was sent. + let (entered, uninstalling) = std::sync::mpsc::channel(); + let (release, resume) = std::sync::mpsc::channel::<()>(); + let shutting_down = std::thread::spawn({ + let installed = installed.clone(); + move || { + runtime().block_on(shutdown_in(&SLOT, move || { + entered.send(()).expect("entered"); + resume.recv().expect("resume"); + installed.lock().expect("lock").take(); + })) + } + }); + uninstalling.recv().expect("the shutdown uninstalls"); + let (b, b_exporter) = pipeline(1_000, &b_out); + let configuring = std::thread::spawn({ + let installed = installed.clone(); + move || configure(&SLOT, &installed, b, b_exporter, || ()) + }); + wait_until_sent(&SLOT, 3); + release.send(()).expect("release"); + configuring.join().expect("configure"); + shutting_down.join().expect("shutdown"); + let in_effect = installed.lock().expect("lock").is_some(); + let exporting = eventually(|| b_out.load(Ordering::SeqCst) == 1); + let stopped = a_out.load(Ordering::SeqCst) == 0; + runtime().block_on(shut_down(&SLOT, &installed)); + assert!(in_effect, "the configure sent after the shutdown is installed"); + assert!(exporting, "and its pipeline exports"); + assert!(stopped, "the shutdown stopped the pipeline before it"); + } + + /// A shutdown sent last ends the lifecycle and joins it, with its exporter runner and + /// exporter: no task is left behind, and the next configure starts a fresh lifecycle. + #[test] + fn a_last_shutdown_joins_every_task_of_the_lifecycle() { + let tasks = tokio::runtime::Builder::new_multi_thread() + .worker_threads(1) + .enable_all() + .build() + .expect("runtime"); + let slot = slot_on(&tasks); + let before = alive_tasks(&slot); + let installed = Arc::new(Installed::default()); + for round in 0..2 { + let transport = Arc::new(Gated { + gate: tokio::sync::Semaphore::new(1_000), + bodies: Mutex::new(Vec::new()), + }); + let config = TelemetryConfig::default(); + let (telemetry, exporter) = livekit_telemetry::Telemetry::new(config, transport); + configure(&slot, &installed, telemetry, exporter, || ()); + assert!(alive_tasks(&slot) > before, "round {round}: the lifecycle runs"); + runtime().block_on(shut_down(&slot, &installed)); + let joined = eventually(|| alive_tasks(&slot) == before); + assert!(joined, "round {round}: {} tasks left, {before} before", alive_tasks(&slot)); + } + } + + /// A shutdown that resumes late, after another shutdown ended its lifecycle and a configure + /// started a new one, never ends the new lifecycle, although the new one's last command has + /// the same number as the late shutdown's. + #[test] + fn a_late_shutdown_never_ends_a_newer_lifecycle() { + let tasks = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .expect("runtime"); + let slot = slot_on(&tasks); + let installed = Arc::new(Installed::default()); + let pipeline = || { + let config = TelemetryConfig { export_timeout_ms: 1_000, ..Default::default() }; + let transport = Arc::new(Gated { + gate: tokio::sync::Semaphore::new(1_000), + bodies: Mutex::new(Vec::new()), + }); + livekit_telemetry::Telemetry::new(config, transport) + }; + let (a, a_exporter) = pipeline(); + configure(&slot, &installed, a, a_exporter, || ()); // the first lifecycle's command 1 + runtime().block_on(async { + // The first half of a shutdown (command 2), which then pauses before it takes the slot. + let (done, stopped) = oneshot::channel(); + let uninstall = { + let installed = installed.clone(); + Box::new(move || { + installed.lock().expect("lock").take(); + }) + }; + let late = + slot.send_if_running(Command::Shutdown { uninstall, done }).expect("running"); + stopped.await.expect("its exporter stopped"); + // Meanwhile another shutdown (command 3) ends and joins that lifecycle, and a + // configure starts a new one: its command 1, then a barrier as its command 2. + shut_down(&slot, &installed).await; + assert!(eventually(|| alive_tasks(&slot) == 0), "the first lifecycle is joined"); + let (b, b_exporter) = pipeline(); + configure(&slot, &installed, b, b_exporter, || ()); + barrier(&slot).await; + // The late shutdown resumes where it paused. + let taken = slot.take_after(late); + let ended_the_new_one = taken.is_some(); + if let Some(lifecycle) = taken { + lifecycle.end().await; + } + let still_owned = slot.lock().is_some(); + shut_down(&slot, &installed).await; + assert!(eventually(|| alive_tasks(&slot) == 0), "every task joined"); + assert!(!ended_the_new_one, "the late shutdown ended the new lifecycle"); + assert!(still_owned, "the new lifecycle keeps its owner"); + }); + } + + /// An exporter that does not stop within its pipeline's shutdown bound is cancelled and + /// joined: once the stop returns, it is gone rather than left running. + #[test] + fn a_stop_past_its_bound_cancels_and_joins_the_exporter() { + /// Set once the task's future is dropped. + struct Gone(Arc); + impl Drop for Gone { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + let gone = Arc::new(AtomicBool::new(false)); + let config = TelemetryConfig { export_timeout_ms: 1_000, ..Default::default() }; + let transport = Arc::new(Gated { + gate: tokio::sync::Semaphore::new(0), + bodies: Mutex::new(Vec::new()), + }); + runtime().block_on(async { + // Its exporter never runs, so the shutdown is never answered: the stop waits out the + // whole bound (2 s), then has to cancel the task that stands in for the exporter. + let (telemetry, _exporter) = livekit_telemetry::Telemetry::new(config, transport); + let task = tokio::spawn({ + let gone = Gone(gone.clone()); + async move { + let _gone = gone; + std::future::pending::<()>().await; + } + }); + stop_running(telemetry, task).await; + assert!(gone.load(Ordering::SeqCst), "the exporter was cancelled and joined"); + }); + } + /// An unsigned participant token with the observability grant (the core reads claims, never /// verifies them). fn granted_token() -> String { @@ -735,17 +1579,20 @@ mod tests { /// (`telemetry_flush`/`telemetry_shutdown`) return before that first purge finishes. #[test] fn a_repeated_opt_out_waits_for_the_first_purge() { - static SLOT: PurgeSlot = Mutex::new(None); + static SLOT: LifecycleSlot = LifecycleSlot::new(); let (release, blocked) = oneshot::channel::<()>(); - start_purge(&SLOT, blocked); - start_purge(&SLOT, async {}); + SLOT.send(Command::Purge(Box::pin(async { + let _ = blocked.await; + }))); + SLOT.send(Command::Purge(Box::pin(async {}))); runtime().block_on(async { - let early = tokio::time::timeout(Duration::from_millis(200), wait_purge(&SLOT)).await; + let early = tokio::time::timeout(Duration::from_millis(200), barrier(&SLOT)).await; assert!(early.is_err(), "the first purge is still running"); release.send(()).expect("first purge waiting"); - tokio::time::timeout(Duration::from_secs(5), wait_purge(&SLOT)) + tokio::time::timeout(Duration::from_secs(5), barrier(&SLOT)) .await .expect("done once the first purge is"); + shutdown_in(&SLOT, || ()).await; }); } @@ -774,9 +1621,10 @@ mod tests { assert!(first.try_next().is_none(), "nothing is served once the call returns"); assert!(telemetry_scope().is_none(), "no scope once the call returns"); assert!(telemetry_stats().is_none()); - room.emit_custom("after".into(), HashMap::new()); // refused: its generation is revoked - // A configure after the opt-out deletes what its storage dir still holds and starts no - // exporter: the queue it hands back is held by nothing else. + // Refused: its generation is revoked. + room.emit_custom("after".into(), HashMap::new()); + // A configure after the opt-out deletes what its storage dir still holds and starts no + // exporter: the queue it hands back is held by nothing else. let dir = std::env::temp_dir().join(format!("lk-refused-{}", std::process::id())); let _ = std::fs::remove_dir_all(&dir); std::fs::create_dir(&dir).expect("dir"); @@ -790,6 +1638,9 @@ mod tests { Vec::new(), ); assert_eq!(Arc::strong_count(&refused), 1, "no exporter holds its transport"); + let next = runtime() + .block_on(async { tokio::time::timeout(Duration::from_secs(1), refused.next()).await }); + assert!(matches!(next, Ok(None)), "a refused queue is finished: `next` never waits"); assert_eq!(std::fs::read_dir(&dir).expect("dir").count(), 0, "the cache is purged"); let _ = std::fs::remove_dir_all(&dir); runtime().block_on(async { diff --git a/livekit-telemetry/SPEC.md b/livekit-telemetry/SPEC.md index 1ccbcabc8..9fea4a4c3 100644 --- a/livekit-telemetry/SPEC.md +++ b/livekit-telemetry/SPEC.md @@ -416,12 +416,15 @@ are dropped in `emit`. The core's own warnings and errors (Rust `log` records from `livekit*` targets, never `livekit_telemetry*`) reach the pipeline by themselves through `livekit-ffi`'s log forwarder: -it copies them as `ffi` records — every JWT-shaped substring (`eyJ…` `.` … `.` …) masked as -``, so a token quoted in an error never leaves the device — and forwards the console entry -unchanged. The copy is active wherever the platform calls `log_forward_bootstrap` (Swift's +it copies them as `ffi` records, with credentials masked so a token quoted in an error never +leaves the device: every compact JWT (three base64url segments whose header and payload decode +to JSON objects, wherever it sits) becomes ``, and the value after `Bearer` or after a +`token` key (`token=`, `access_token=`, `"token": …`) becomes ``. The console entry is +forwarded unchanged. Platform lines handed to `telemetry_log` or `TelemetryScope.log` are masked +the same way. The copy is active wherever the platform calls `log_forward_bootstrap` (Swift's default `OSLogger` with `ffi: true` does), whatever level it passes: that level filters only what -is forwarded to the platform's console (the global `log` level is kept at `warn` or looser) — Rust allows one logger per process, and without the -forwarder the core's records go nowhere. Platforms must not feed forwarded Rust log entries +is forwarded to the platform's console (the global `log` level is kept at `warn` or looser) — +Rust allows one logger per process, and without the forwarder the core's records go nowhere. Platforms must not feed forwarded Rust log entries (what `log_forward_receive` returns) to `telemetry_log` / `log(record)`: the core already copied them, so they would be counted twice. `telemetry_log` is for the platform's own and WebRTC's lines. diff --git a/livekit-telemetry/src/event.rs b/livekit-telemetry/src/event.rs index 8f5bd7c67..9d51ef29f 100644 --- a/livekit-telemetry/src/event.rs +++ b/livekit-telemetry/src/event.rs @@ -131,6 +131,19 @@ impl Severity { } } +impl From for Severity { + /// The severity of a `log` record at `level`. + fn from(level: log::Level) -> Self { + match level { + log::Level::Error => Self::Error, + log::Level::Warn => Self::Warn, + log::Level::Info => Self::Info, + log::Level::Debug => Self::Debug, + log::Level::Trace => Self::Trace, + } + } +} + /// Where a log line came from. WebRTC is chatty at warn, so only its errors become records; /// the SDK and the core use the configured floor. #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] diff --git a/livekit-telemetry/src/global.rs b/livekit-telemetry/src/global.rs index 85fb24761..b4d409235 100644 --- a/livekit-telemetry/src/global.rs +++ b/livekit-telemetry/src/global.rs @@ -166,8 +166,10 @@ pub fn install( previous.map(|p| p.telemetry) } -/// Stop the instruments and remove the process pipeline; the caller shuts it down. -fn uninstall() -> Option { +/// Stop the instruments and remove the process pipeline; the caller shuts it down. With +/// [`install`], the synchronous half of [`shutdown`]: a caller that orders installs and uninstalls +/// itself drains the pipeline it gets back when it is ready to. +pub fn uninstall() -> Option { let _lifecycle = lifecycle(); stop_installed_instruments(); INSTALLED.write().unwrap_or_else(|e| e.into_inner()).take().map(|i| i.telemetry)