From 69d8442f4ec50c4f6cbf78e21e5f07214de99ce4 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:30:15 +0200 Subject: [PATCH 1/3] feat(telemetry): add the transport seam and LiveKit Cloud destinations `TelemetryTransport` moves encoded batches off the device (`NetTransport` over `livekit-net` behind the `net` feature) and returns status, headers and body; the core classifies every answer (partial success, 4xx, 413, 401/403, 404, 429/503 `Retry-After` and `RetryInfo`, 5xx, no answer). Destinations turn a Room's server URL and token into the ingest URL (LiveKit Cloud hosts only, parsed, never string-matched), read the token's unverified grant and expiry, and key credentials by project and session. `LK_TELEMETRY_ENDPOINT` points every upload at a local collector. --- Cargo.lock | 8 + livekit-telemetry/Cargo.toml | 17 + livekit-telemetry/src/destination.rs | 768 +++++++++++++++++++++++++++ livekit-telemetry/src/lib.rs | 19 + livekit-telemetry/src/transport.rs | 568 ++++++++++++++++++++ 5 files changed, 1380 insertions(+) create mode 100644 livekit-telemetry/src/destination.rs create mode 100644 livekit-telemetry/src/transport.rs diff --git a/Cargo.lock b/Cargo.lock index e43a2f1cd..f54f435ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4016,11 +4016,19 @@ dependencies = [ name = "livekit-telemetry" version = "0.1.0" dependencies = [ + "async-trait", + "base64 0.22.1", + "livekit-net", "log", "opentelemetry-proto", "prost 0.14.4", + "prost-types 0.14.4", "rand 0.9.5", + "serde_json", + "thiserror 2.0.19", + "tokio", "uniffi", + "url", ] [[package]] diff --git a/livekit-telemetry/Cargo.toml b/livekit-telemetry/Cargo.toml index 7364e25b3..ad1fe9df6 100644 --- a/livekit-telemetry/Cargo.toml +++ b/livekit-telemetry/Cargo.toml @@ -8,17 +8,34 @@ edition.workspace = true repository.workspace = true [dependencies] +tokio = { workspace = true, default-features = false, features = ["macros", "sync", "time"] } log = { workspace = true } +thiserror = { workspace = true } +async-trait = "0.1" prost = { workspace = true } +# `google.protobuf.Any`/`Duration` for the `google.rpc.Status` error body (RetryInfo). +prost-types = { workspace = true } rand = { workspace = true } +# Unverified JWT claims (observability grant, expiry); both already linked by livekit-api. +base64 = "0.22" +serde_json = { workspace = true } +# Server URLs are parsed, never string-matched, before a token may follow them. +url = "2.3" # OTLP message types only (`gen-tonic-messages` = prost structs, no tonic). Same prost as # livekit-protocol so a single prost is linked. opentelemetry-proto = { version = "0.32", default-features = false, features = ["logs", "trace", "gen-tonic-messages"] } +livekit-net = { workspace = true, optional = true } uniffi = { workspace = true, features = ["scaffolding-ffi-buffer-fns"], optional = true } [features] +# Default HTTP transport over the pluggable `livekit-net` client (native backend, or one +# the host registered with `livekit_net::set_http_client`). +net = ["dep:livekit-net"] uniffi = ["dep:uniffi"] +[dev-dependencies] +tokio = { workspace = true, default-features = false, features = ["rt", "rt-multi-thread", "macros", "sync", "time", "test-util", "net", "io-util"] } + # How CI checks this crate's features, read by # `.github/workflows/feature-combinations-curated.yml` via `cargo metadata`. [package.metadata.feature-combinations] diff --git a/livekit-telemetry/src/destination.rs b/livekit-telemetry/src/destination.rs new file mode 100644 index 000000000..55d86d198 --- /dev/null +++ b/livekit-telemetry/src/destination.rs @@ -0,0 +1,768 @@ +// 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. + +//! Where batches go and with which credential — decided here, once, for every platform. +//! +//! A platform hands over exactly two things: the LiveKit server URL a room connects to and the +//! participant token it connects with (again on every token refresh). The core derives the +//! ingest URL, checks that the host is LiveKit Cloud, reads the token's unverified claims (the +//! observability grant, the expiry), and routes every batch to the project its session belongs +//! to, so two rooms talking to two projects never share a token or a destination. + +use std::{ + collections::HashMap, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; + +use base64::{ + alphabet, + engine::{DecodePaddingMode, GeneralPurpose, GeneralPurposeConfig}, + Engine, +}; +use tokio::time::Instant; + +/// The environment variable that points every upload at a collector of your own — a local +/// OpenTelemetry collector for end-to-end tests. Not part of any platform API: set it in the +/// test process's environment before the pipeline starts. Either a base URL +/// (`http://localhost:4318`, OTLP paths `/v1/logs` and `/v1/traces` are appended) or a full logs +/// URL ending in `logs` (its traces URL is derived by replacing that segment). +pub const ENDPOINT_OVERRIDE_ENV: &str = "LK_TELEMETRY_ENDPOINT"; + +/// The longest a token is believed to stay valid, whatever its `exp` says. +const MAX_TOKEN_LIFETIME: Duration = Duration::from_secs(366 * 24 * 60 * 60); + +/// LiveKit Cloud project hosts: `.livekit.cloud` in production, +/// `.staging.livekit.cloud` in staging. Self-hosted servers have no ingest. +const CLOUD_SUFFIX: &str = ".livekit.cloud"; + +/// Which OTLP signal a batch carries; picks the ingest path. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Signal { + Logs, + Traces, +} + +/// A request target: URL per signal and the bearer token, when one is due. +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct Target { + /// The project the batch is sent to (`None` for the test override): what its answer is + /// attributed to — a 404, a disable, a pause. + pub project: Option, + pub logs: String, + pub traces: String, + pub token: Option, +} + +impl Target { + pub fn url(&self, signal: Signal) -> &str { + match signal { + Signal::Logs => &self.logs, + Signal::Traces => &self.traces, + } + } +} + +/// What the unverified claims of a participant token say — enough to not send with a token the +/// collector is known to refuse. The signature is the collector's business. +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct Token { + raw: String, + /// `observability.write` or `observability.clientWrite`. + grant: bool, + /// From `exp`, on the monotonic clock (so tests can move it). + expires: Option, + /// The credential's identity (a hash of the raw token): what a refusal is recorded against. + id: u64, +} + +impl Token { + pub fn parse(raw: &str) -> Self { + let claims = raw + .split('.') + .nth(1) + .and_then(|payload| { + const URL_SAFE: GeneralPurpose = GeneralPurpose::new( + &alphabet::URL_SAFE, + GeneralPurposeConfig::new() + .with_decode_padding_mode(DecodePaddingMode::Indifferent), + ); + URL_SAFE.decode(payload).ok() + }) + .and_then(|json| serde_json::from_slice::(&json).ok()); + let grant = claims.as_ref().is_some_and(|c| { + let observability = &c["observability"]; + observability["write"].as_bool() == Some(true) + || observability["clientWrite"].as_bool() == Some(true) + }); + // Untrusted input, read conservatively: a past or negative `exp` is expired, and so is + // one that is not a number at all; a fractional one counts in whole seconds; one beyond + // any real token lifetime is clamped (an `Instant` that far out would overflow). Only a + // token without `exp` has no known expiry. + let expires = claims.as_ref().and_then(|c| c.get("exp")).map(|exp| { + let now = + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs_f64(); + let left = match exp.as_f64() { + Some(exp) if exp.is_finite() && exp > now => { + Duration::from_secs_f64((exp - now).min(MAX_TOKEN_LIFETIME.as_secs_f64())) + } + _ => Duration::ZERO, + }; + Instant::now().checked_add(left).unwrap_or_else(Instant::now) + }); + let id = { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + raw.hash(&mut hasher); + hasher.finish() + }; + Self { raw: raw.to_owned(), grant, expires, id } + } + + fn expired(&self) -> bool { + self.expires.is_some_and(|at| Instant::now() >= at) + } + + /// May be sent: carries the grant, not expired, not refused. + fn usable(&self, refused: &Refusals) -> bool { + self.grant && !self.expired() && !refused.contains(self) + } +} + +/// Why a project receives nothing (anymore). Each is logged once. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Dead { + /// Not a LiveKit Cloud host (self-hosted, a typo): there is no ingest to talk to. + NotCloud, + /// The project's own token carries no observability grant: the customer did not opt in. + NoGrant, + /// The collector answered 404: no client ingest on this host. Revived by the next token + /// (a new connect or refresh), so a misrouted answer cannot silence a project for good. + NotFound, + /// The collector said the project's data recording is disabled by its owner. + Disabled, +} + +/// One LiveKit Cloud project, keyed by its host. +#[derive(Debug, Default)] +struct Project { + /// The latest token any room handed over for the project: what process-level batches and + /// batches from a previous launch are sent with. + latest: Option, + /// A token with the grant was seen: the customer opted in. Refreshed tokens may lack the + /// grant (today's server drops it on refresh) without withdrawing that. + consented: bool, + dead: Option, +} + +/// Offer `new` for a token slot: a refused token is never taken, into any slot; a token with the +/// grant wins; one without it does not replace a usable granted token (today's refreshes drop +/// the grant). +fn offer(slot: &mut Option, new: &Token, refused: &Refusals) { + if refused.contains(new) { + return; + } + let keep = slot.as_ref().is_some_and(|t| t.raw == new.raw || (!new.grant && t.usable(refused))); + if !keep { + *slot = Some(new.clone()); + } +} + +/// Credentials the collector refused, by identity, shared by every slot: a refused token is never +/// sent again, whichever Room or project hands it over. An entry lives until its token expires; +/// tokens without `exp` stay refused for the life of the process. +// ponytail: grows with refusals of tokens without `exp` (LiveKit tokens always carry one); a cap +// if a server ever mints tokens without expiry. +#[derive(Debug, Default)] +struct Refusals(HashMap>); + +impl Refusals { + fn contains(&self, token: &Token) -> bool { + self.0.contains_key(&token.id) + } + + /// Returns whether the refusal is new. + fn add(&mut self, token: &Token) -> bool { + let now = Instant::now(); + self.0.retain(|_, expires| expires.is_none_or(|at| at > now)); + self.0.insert(token.id, token.expires).is_none() + } +} + +/// The owner of process-level records (device state, pre-room errors, self-telemetry): no Room, +/// so they go to the latest project with its latest token. +pub(crate) const PROCESS_OWNER: &str = "process"; + +/// Where a batch is headed. +#[derive(Debug, Clone, PartialEq)] +pub(crate) enum Route { + /// Ready: send now. + Send(Target), + /// Keep it cached: no destination yet, or no usable token (missing, expired, grant-less, + /// refused) until the platform hands over a fresh one. A hard hold: no escape hatch sends + /// it anyway. + Wait, + /// Drop it: its project receives nothing. + Drop, +} + +/// Every project and session this process talks to, plus the test override. Tokens live here, +/// in memory only: never persisted, never in a batch. +#[derive(Debug, Default)] +pub(crate) struct Destinations { + projects: HashMap, + /// Each Room session's own latest token per project, by (host, session id): its batches for + /// that project are sent with it and never with another Room's, nor with the token it later + /// got for another project. + sessions: HashMap<(String, String), Token>, + /// Each Room session's first project: where the records it captured before it had a server + /// go. A Room's records never fall back to another Room's project. + rooms: HashMap, + refused: Refusals, + /// The host process-level batches (device state, pre-room logs, self-telemetry) go to: the + /// project most recently handed a token, while it is alive. + latest: Option, + /// [`ENDPOINT_OVERRIDE_ENV`]: every batch goes there, no Cloud rules, no token. + endpoint_override: Option, +} + +impl Destinations { + pub fn new(endpoint_override: Option<&str>) -> Self { + Self { + endpoint_override: endpoint_override.filter(|e| !e.is_empty()).map(override_target), + ..Self::default() + } + } + + /// The first project a Room session was routed to: where what it captured before it had a + /// server belongs. + pub fn first_project(&self, owner: &str) -> Option { + self.rooms.get(owner).cloned() + } + + pub fn has_override(&self) -> bool { + self.endpoint_override.is_some() + } + + /// Session `owner`'s server URL and token (at connect, and again on every refresh). Returns + /// the project host its batches carry, or `None` when the URL has no host. + pub fn set(&mut self, url: &str, token: &str, owner: &str) -> Option { + let Server { key: host, cloud } = Server::parse(url)?; + if self.endpoint_override.is_some() { + return Some(host); + } + let project = self.projects.entry(host.clone()).or_default(); + if project.dead.is_none() && !cloud { + project.dead = Some(Dead::NotCloud); + log::warn!("{host} is not LiveKit Cloud: client telemetry stays on this device"); + } + if matches!(project.dead, Some(Dead::NotCloud | Dead::Disabled)) { + return Some(host); + } + let token = Token::parse(token); + if project.dead == Some(Dead::NotFound) { + log::debug!("new token for {host}: trying its ingest again"); + project.dead = None; + } + if token.grant { + project.consented = true; + if project.dead == Some(Dead::NoGrant) { + project.dead = None; + } + } else if !project.consented && project.dead.is_none() { + project.dead = Some(Dead::NoGrant); + log::info!("the token for {host} has no observability grant: telemetry stays local"); + } + offer(&mut project.latest, &token, &self.refused); + let key = (host.clone(), owner.to_owned()); + let mut own = self.sessions.remove(&key); + offer(&mut own, &token, &self.refused); + self.sessions.extend(own.map(|own| (key, own))); + self.rooms.entry(owner.to_owned()).or_insert_with(|| host.clone()); + self.latest = Some(host.clone()); + Some(host) + } + + /// Where a batch goes: `host` is its project as captured with its records, `owner` the + /// session that produced them. A Room known to this process sends with its own latest token + /// for that project and waits when it has none; process-level batches and a previous + /// launch's use the project's latest. A Room's records captured before it had a server go + /// to its first project, never to another Room's. + pub fn route(&self, host: Option<&str>, owner: Option<&str>) -> Route { + if let Some(target) = &self.endpoint_override { + return Route::Send(target.clone()); + } + let room = owner.filter(|o| *o != PROCESS_OWNER); + let host = match (host, room) { + (Some(host), _) => host, + (None, Some(room)) => match self.rooms.get(room) { + Some(first) => first.as_str(), + None => return Route::Wait, + }, + (None, None) => match self.fallback() { + Some(host) => host, + // Nothing to talk to yet: wait. Only dead projects so far: nobody will ever + // take process-level data, so it is not kept either. + None if self.all_dead() => return Route::Drop, + None => return Route::Wait, + }, + }; + let Some(project) = self.projects.get(host) else { return Route::Wait }; + if project.dead.is_some() { + return Route::Drop; + } + let token = match room { + Some(room) if self.rooms.contains_key(room) => { + self.sessions.get(&(host.to_owned(), room.to_owned())) + } + _ => project.latest.as_ref(), + }; + match token { + Some(token) if token.usable(&self.refused) => Route::Send(Target { + project: Some(host.to_owned()), + logs: ingest_url(host, Signal::Logs), + traces: ingest_url(host, Signal::Traces), + token: Some(token.raw.clone()), + }), + _ => Route::Wait, + } + } + + /// The host of the project process-level batches go to: the latest, else any alive one. + fn fallback(&self) -> Option<&str> { + let alive = |host: &&String| self.projects.get(*host).is_some_and(|p| p.dead.is_none()); + self.latest + .as_ref() + .filter(alive) + .or_else(|| self.projects.keys().find(alive)) + .map(String::as_str) + } + + fn all_dead(&self) -> bool { + !self.projects.is_empty() && self.projects.values().all(|p| p.dead.is_some()) + } + + /// Whether a session routed to `host` should still collect: `false` once its project is + /// known to receive nothing, so its records are dropped at the door instead of cached. + pub fn alive(&self, host: Option<&str>) -> bool { + self.endpoint_override.is_some() || !matches!(self.route(host, None), Route::Drop) + } + + /// The collector refused `token` (401/403): it is not sent again, from any slot; batches that + /// would use it wait for the next one. + pub fn refuse(&mut self, token: &str) { + if self.refused.add(&Token::parse(token)) { + log::warn!("the collector refused a token; waiting for a new one"); + } + } + + /// Keep credentials only where something still needs them. `live` maps every session still + /// alive — held by its Room, or by records, windows or spans not yet cached — to its current + /// project; its credentials stay (for each project it was routed to: records it captured for + /// an earlier project may still be queued). `backlog` is (project or none, owner) of every + /// cached batch: their credentials stay too. Project-level copies stay while a live session + /// is routed there or the backlog has batches for the project (process-level batches: the + /// latest project). + pub fn retain_owners( + &mut self, + live: &HashMap>, + backlog: &std::collections::HashSet<(Option, String)>, + ) { + // A Room's batch captured before it had a server is cached without a project: it goes + // to that Room's first project, so it backs that project's credential. + let backlog: std::collections::HashSet<(Option, String)> = backlog + .iter() + .map(|(host, owner)| match (host, self.rooms.get(owner)) { + (None, Some(first)) if owner != PROCESS_OWNER => { + (Some(first.clone()), owner.clone()) + } + _ => (host.clone(), owner.clone()), + }) + .collect(); + let backed = + |host: &str, owner: &str| backlog.contains(&(Some(host.to_owned()), owner.to_owned())); + self.sessions.retain(|(host, owner), _| backed(host, owner) || live.contains_key(owner)); + self.rooms + .retain(|owner, _| live.contains_key(owner) || backlog.iter().any(|(_, o)| o == owner)); + let unrouted = backlog.iter().any(|(host, _)| host.is_none()); + let latest = self.latest.clone(); + for (host, project) in &mut self.projects { + let needed = live.values().any(|route| route.as_deref() == Some(host)) + || backlog.iter().any(|(h, _)| h.as_deref() == Some(host)) + || (unrouted && latest.as_deref() == Some(host)); + if !needed { + project.latest = None; + } + } + } + + /// How many credentials are held, Room and project copies together. + #[cfg(test)] + pub fn credentials(&self) -> usize { + self.sessions.len() + self.projects.values().filter(|p| p.latest.is_some()).count() + } + + /// Try `host`'s ingest again if it answered 404 (a reconnect). Returns whether it did. + pub fn revive(&mut self, host: &str) -> bool { + match self.projects.get_mut(host) { + Some(project) if project.dead == Some(Dead::NotFound) => { + project.dead = None; + true + } + _ => false, + } + } + + /// The project on `host` receives nothing (see [`Dead`] for how long). + pub fn kill(&mut self, host: &str, why: Dead) { + let project = self.projects.entry(host.to_owned()).or_default(); + if project.dead.is_none() { + match why { + Dead::NotFound => log::warn!("{host} has no client telemetry ingest; going silent"), + Dead::Disabled => log::warn!("{host} disabled data recording; going silent"), + Dead::NotCloud | Dead::NoGrant => {} + } + project.dead = Some(why); + } + } +} + +/// A server URL, parsed (WHATWG URL rules, the same a browser or `URLSession` applies): its +/// routing key — the canonical host, with the port when one is given — and whether it is a +/// LiveKit Cloud project a token may be sent to. +struct Server { + key: String, + cloud: bool, +} + +impl Server { + fn parse(url: &str) -> Option { + let url = url::Url::parse(url).ok()?; + let host = url.host_str().filter(|h| !h.is_empty())?.to_ascii_lowercase(); + let key = match url.port() { + Some(port) => format!("{host}:{port}"), + None => host.clone(), + }; + // Credentials only ever go to `https://.livekit.cloud/…`, built from the + // validated host alone: a TLS scheme, a domain under the Cloud suffix with a label of its + // own, the default port, no userinfo. Anything else — a look-alike that only a string + // check would accept (`evil.example\.livekit.cloud` is host `evil.example`), a + // plaintext scheme, an odd port — is not Cloud. + let cloud = matches!(url.scheme(), "wss" | "https") + && matches!(url.host(), Some(url::Host::Domain(_))) + && host.strip_suffix(CLOUD_SUFFIX).is_some_and(|label| !label.is_empty()) + && url.port().is_none() + && url.username().is_empty() + && url.password().is_none(); + Some(Self { key, cloud }) + } +} + +/// `https:///observability/client/{logs,traces}/otlp/v0`: the client table set. +fn ingest_url(host: &str, signal: Signal) -> String { + let signal = match signal { + Signal::Logs => "logs", + Signal::Traces => "traces", + }; + format!("https://{host}/observability/client/{signal}/otlp/v0") +} + +fn override_target(endpoint: &str) -> Target { + let endpoint = endpoint.trim_end_matches('/'); + let (logs, traces) = match endpoint.rsplit_once("logs") { + Some((before, after)) => (endpoint.to_owned(), format!("{before}traces{after}")), + None => (format!("{endpoint}/v1/logs"), format!("{endpoint}/v1/traces")), + }; + Target { project: None, logs, traces, token: None } +} + +#[cfg(test)] +pub(crate) mod tests { + use super::*; + + /// An unsigned JWT with these claims — enough for the core, which never verifies. + pub(crate) fn jwt(claims: serde_json::Value) -> String { + let engine = base64::engine::general_purpose::URL_SAFE_NO_PAD; + format!( + "{}.{}.sig", + engine.encode(br#"{"alg":"HS256"}"#), + engine.encode(claims.to_string()) + ) + } + + pub(crate) fn granted(ttl_secs: u64) -> String { + let exp = SystemTime::now().duration_since(UNIX_EPOCH).expect("clock").as_secs() + ttl_secs; + jwt(serde_json::json!({ "exp": exp, "observability": { "write": true } })) + } + + pub(crate) fn grantless(ttl_secs: u64) -> String { + let exp = SystemTime::now().duration_since(UNIX_EPOCH).expect("clock").as_secs() + ttl_secs; + jwt(serde_json::json!({ "exp": exp, "video": { "roomJoin": true } })) + } + + fn send(route: Route) -> Target { + match route { + Route::Send(target) => target, + other => panic!("expected a target, got {other:?}"), + } + } + + #[test] + fn only_livekit_cloud_hosts_get_an_ingest_url() { + let mut d = Destinations::default(); + let token = granted(600); + let host = d.set("wss://Proj.LiveKit.Cloud/rtc?access_token=x#f", &token, "s"); + assert_eq!(host.as_deref(), Some("proj.livekit.cloud")); + let target = send(d.route(host.as_deref(), Some("s"))); + assert_eq!(target.logs, "https://proj.livekit.cloud/observability/client/logs/otlp/v0"); + assert_eq!(target.traces, "https://proj.livekit.cloud/observability/client/traces/otlp/v0"); + assert_eq!(target.token.as_deref(), Some(token.as_str())); + + let staging = d.set("https://p.staging.livekit.cloud", &token, "s"); + assert!(matches!(d.route(staging.as_deref(), None), Route::Send(_)), "staging is Cloud"); + + for url in + ["ws://localhost:7880", "wss://livekit.example.com", "wss://livekit.cloud.evil.io"] + { + let host = d.set(url, &token, "s"); + assert_eq!(d.route(host.as_deref(), None), Route::Drop, "{url}: no ingest"); + } + assert_eq!(d.set("nonsense", &token, "s"), None); + let empty = d.set("wss:///rtc", &token, "s"); // the parser reads host `rtc` + assert_eq!(d.route(empty.as_deref(), None), Route::Drop); + } + + /// Whatever a string check would accept, the token only goes to a host the URL parser agrees is + /// `.livekit.cloud`, over TLS, on the default port, without userinfo — and the + /// endpoint is built from that host alone. + #[test] + fn look_alike_server_urls_never_get_the_token() { + let token = granted(600); + for url in [ + "wss://evil.example\\.livekit.cloud", + "wss://evil.example\\x.livekit.cloud/rtc", + "wss://user:pass@p.livekit.cloud", + "wss://evil.example@p.livekit.cloud", + "wss://p.livekit.cloud:8443", + "ws://p.livekit.cloud", + "http://p.livekit.cloud", + "wss://livekit.cloud", + "wss://.livekit.cloud", + "wss://p.livekit.cloud.evil.io", + "wss://p.livekit.cloud%2eevil.io", + "wss://evil.io#.livekit.cloud", + "wss://evil.io?.livekit.cloud", + "wss://[::1]", + "wss://1.2.3.4", + "wss://p.livekit.cloud\u{0}.evil", + ] { + let mut d = Destinations::default(); + let host = d.set(url, &token, "s"); + let route = d.route(host.as_deref(), Some("s")); + assert!(!matches!(route, Route::Send(_)), "{url} → {host:?} must not get the token"); + } + let mut d = Destinations::default(); + let host = d.set("WSS://P.LiveKit.Cloud:443/rtc?x=1#y", &token, "s"); + let target = send(d.route(host.as_deref(), Some("s"))); + assert_eq!(target.logs, "https://p.livekit.cloud/observability/client/logs/otlp/v0"); + } + + #[test] + fn a_token_without_the_grant_is_not_consent() { + let mut d = Destinations::default(); + let host = d.set("wss://p.livekit.cloud", &grantless(600), "s"); + assert_eq!(d.route(host.as_deref(), Some("s")), Route::Drop, "never opted in"); + assert_eq!(d.route(None, None), Route::Drop, "and no project takes process data"); + let host = d.set("wss://p.livekit.cloud", &granted(600), "s"); + assert!(matches!(d.route(host.as_deref(), Some("s")), Route::Send(_)), "a grant opts in"); + } + + #[tokio::test(start_paused = true)] + async fn a_refresh_that_drops_the_grant_keeps_the_granted_token_until_it_expires() { + let mut d = Destinations::default(); + let join = granted(60); + let host = d.set("wss://p.livekit.cloud", &join, "s"); + d.set("wss://p.livekit.cloud", &grantless(600), "s"); + assert_eq!(send(d.route(host.as_deref(), Some("s"))).token, Some(join), "still granted"); + tokio::time::advance(Duration::from_secs(61)).await; + assert_eq!(d.route(host.as_deref(), Some("s")), Route::Wait, "expired: hold, never drop"); + let refreshed = granted(600); + d.set("wss://p.livekit.cloud", &refreshed, "s"); + assert_eq!(send(d.route(host.as_deref(), Some("s"))).token, Some(refreshed)); + } + + #[test] + fn a_refused_token_is_never_sent_again() { + let mut d = Destinations::default(); + let first = granted(600); + let host = d.set("wss://p.livekit.cloud", &first, "s").expect("host"); + d.refuse(&first); + assert_eq!(d.route(Some(&host), Some("s")), Route::Wait); + d.set("wss://p.livekit.cloud", &first, "s"); + assert_eq!(d.route(Some(&host), Some("s")), Route::Wait, "handing it over again: still no"); + d.set("wss://p.livekit.cloud", &granted(900), "s"); + assert!(matches!(d.route(Some(&host), Some("s")), Route::Send(_))); + } + + #[test] + fn rooms_never_borrow_each_others_tokens() { + let mut d = Destinations::default(); + let (a, b, c) = (granted(600), granted(700), granted(800)); + let host_a = d.set("wss://a.livekit.cloud", &a, "room-a"); + let host_b = d.set("wss://b.livekit.cloud", &b, "room-b"); + d.set("wss://a.livekit.cloud", &c, "room-c"); // a second room on project a + let target = send(d.route(host_a.as_deref(), Some("room-a"))); + assert!(target.logs.starts_with("https://a.") && target.token == Some(a.clone())); + let target = send(d.route(host_a.as_deref(), Some("room-c"))); + assert_eq!(target.token, Some(c.clone()), "same project, its own token"); + let target = send(d.route(host_b.as_deref(), Some("room-b"))); + assert!(target.logs.starts_with("https://b.") && target.token == Some(b)); + d.refuse(&a); + assert_eq!(d.route(host_a.as_deref(), Some("room-a")), Route::Wait, "no borrowing c"); + let previous_launch = send(d.route(host_a.as_deref(), Some("gone"))); + assert_eq!(previous_launch.token, Some(c), "a previous launch's batch: project's latest"); + d.kill("b.livekit.cloud", Dead::Disabled); + assert!(send(d.route(None, None)).logs.starts_with("https://a."), "process: a live one"); + } + + /// A Room's batches keep the project they were captured for — after it reconnects to another + /// project they still go to the first with the first project's token — and records it captured + /// before it had a server never go to another Room's project. + #[test] + fn ownership_survives_a_room_changing_projects() { + let mut d = Destinations::default(); + let (a, b) = (granted(600), granted(700)); + let host_a = d.set("wss://a.livekit.cloud", &a, "room").expect("a"); + d.set("wss://b.livekit.cloud", &b, "room"); + let target = send(d.route(Some(&host_a), Some("room"))); + assert!(target.logs.starts_with("https://a.") && target.token == Some(a)); + + d.set("wss://c.livekit.cloud", &granted(800), "other"); + assert_eq!(d.route(None, Some("unconnected")), Route::Wait, "no borrowing c"); + let early = send(d.route(None, Some("room"))); + assert!(early.logs.starts_with("https://a."), "a Room's first project"); + assert!(matches!(d.route(None, Some(PROCESS_OWNER)), Route::Send(_))); + } + + /// A refusal belongs to the credential — no slot, Room, project or process route can make it + /// usable again — and credentials follow the live Rooms and their backlog, project-level copies + /// included. + #[test] + fn refusals_stick_and_credentials_follow_live_rooms() { + let mut d = Destinations::default(); + let tokens: Vec = (0..40).map(|n| granted(600 + n)).collect(); + for (n, token) in tokens.iter().enumerate() { + d.set("wss://p.livekit.cloud", token, &format!("room-{n}")); + d.refuse(token); + } + for (n, token) in tokens.iter().enumerate() { + let room = format!("room-{n}"); + d.set("wss://p.livekit.cloud", token, &room); // handed over again + assert_eq!(d.route(Some("p.livekit.cloud"), Some(&room)), Route::Wait, "{n}"); + d.set("wss://p.livekit.cloud", token, &format!("new-{n}")); // by a new Room + assert_eq!(d.route(Some("p.livekit.cloud"), Some(&format!("new-{n}"))), Route::Wait); + assert_eq!(d.route(None, Some(PROCESS_OWNER)), Route::Wait, "{n}: process route"); + assert_eq!(d.route(Some("p.livekit.cloud"), Some("gone")), Route::Wait, "{n}: restart"); + } + // Refuse A, let B become the latest, hand A over again: A stays refused everywhere. + let (a, b) = (granted(900), granted(901)); + d.set("wss://q.livekit.cloud", &a, "ra"); + d.refuse(&a); + d.set("wss://q.livekit.cloud", &b, "rb"); + d.set("wss://q.livekit.cloud", &a, "ra"); + let process = send(d.route(Some("q.livekit.cloud"), Some("gone"))); + assert_eq!(process.token, Some(b), "the project copy is b, never the refused a"); + + // Churn: 50 projects, their Rooms gone, nothing cached: nothing is retained. + for n in 0..50 { + d.set(&format!("wss://c{n}.livekit.cloud"), &granted(1000 + n), &format!("churn-{n}")); + } + let live: HashMap> = + [("rb".to_owned(), Some("q.livekit.cloud".to_owned()))].into(); + d.retain_owners(&live, &Default::default()); + assert_eq!(d.credentials(), 2, "rb's token and q's project copy, nothing else"); + let backlog = [(Some("q.livekit.cloud".to_owned()), "ra".to_owned())].into(); + d.set("wss://q.livekit.cloud", &granted(902), "ra"); + d.retain_owners(&Default::default(), &backlog); + assert_eq!(d.credentials(), 2, "a gone Room's backlog keeps its credential and q's copy"); + } + + #[test] + fn not_found_recovers_with_the_next_token_disabled_does_not() { + let mut d = Destinations::default(); + let host = d.set("wss://p.livekit.cloud", &granted(600), "s").expect("host"); + d.kill(&host, Dead::NotFound); + assert_eq!(d.route(Some(&host), Some("s")), Route::Drop); + d.set("wss://p.livekit.cloud", &granted(700), "s"); + assert!(matches!(d.route(Some(&host), Some("s")), Route::Send(_)), "a new connect retries"); + d.kill(&host, Dead::Disabled); + d.set("wss://p.livekit.cloud", &granted(800), "s"); + assert_eq!(d.route(Some(&host), Some("s")), Route::Drop, "the owner's switch stands"); + } + + #[test] + fn the_override_takes_everything_without_a_token() { + let d = Destinations::new(Some("http://localhost:4318/")); + let target = send(d.route(None, None)); + assert_eq!(target.logs, "http://localhost:4318/v1/logs"); + assert_eq!(target.traces, "http://localhost:4318/v1/traces"); + assert_eq!(target.token, None); + let full = Destinations::new(Some("https://h/observability/client/logs/otlp/v0")); + assert_eq!( + send(full.route(Some("ignored"), None)).traces, + "https://h/observability/client/traces/otlp/v0" + ); + let mut local = Destinations::new(Some("http://localhost:4318")); + let host = local.set("ws://localhost:7880", "not-a-jwt", "s"); + assert!(matches!(local.route(host.as_deref(), Some("s")), Route::Send(_))); + } + + /// A past, negative, fractional-past or malformed `exp` is expired (a hard hold); only a + /// missing one means no known expiry. + #[tokio::test(start_paused = true)] + async fn untrusted_expiry_claims_fail_closed() { + let now = SystemTime::now().duration_since(UNIX_EPOCH).expect("clock").as_secs_f64(); + let token = |exp: serde_json::Value| { + jwt(serde_json::json!({ "exp": exp, "observability": { "write": true } })) + }; + for exp in [ + serde_json::json!(-1), + serde_json::json!(0), + serde_json::json!(now - 0.5), + serde_json::json!(-1e308), + serde_json::json!("tomorrow"), + serde_json::json!(true), + serde_json::json!(null), + serde_json::json!({}), + ] { + let mut d = Destinations::default(); + let host = d.set("wss://p.livekit.cloud", &token(exp.clone()), "s"); + assert_eq!(d.route(host.as_deref(), Some("s")), Route::Wait, "exp {exp}: expired"); + } + let mut d = Destinations::default(); + let host = d.set("wss://p.livekit.cloud", &token(serde_json::json!(now + 60.9)), "s"); + assert!(matches!(d.route(host.as_deref(), Some("s")), Route::Send(_)), "fractional future"); + let host = d.set("wss://p.livekit.cloud", &token(serde_json::json!(1e300)), "s"); + assert!(matches!(d.route(host.as_deref(), Some("s")), Route::Send(_)), "extreme: clamped"); + } + + #[test] + fn tokens_are_read_not_verified() { + assert!(Token::parse(&granted(60)).grant); + let client_write = jwt(serde_json::json!({ "observability": { "clientWrite": true } })); + let parsed = Token::parse(&client_write); + assert!(parsed.grant && parsed.expires.is_none(), "no exp: no known expiry"); + assert!(!Token::parse("garbage").grant); + assert!(!Token::parse("a.!!!.c").grant); + } +} diff --git a/livekit-telemetry/src/lib.rs b/livekit-telemetry/src/lib.rs index 4620e0f31..9afce456d 100644 --- a/livekit-telemetry/src/lib.rs +++ b/livekit-telemetry/src/lib.rs @@ -51,9 +51,28 @@ mod otlp; /// OTLP protobuf types (re-exported from `opentelemetry-proto`). mod proto; +/// Transport seam: how encoded batches leave the device. +#[cfg_attr( + not(test), + expect( + dead_code, + reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet" + ) +)] +mod transport; + +/// Where batches go: server URL + token → ingest URL, grant, expiry, per-project routing. +#[expect( + dead_code, + reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet" +)] +mod destination; + +pub use destination::ENDPOINT_OVERRIDE_ENV; pub use event::*; pub use span::SpanOutcome; pub use stats::{TelemetryStats, TelemetryStatus}; +pub use transport::*; #[cfg(feature = "uniffi")] uniffi::setup_scaffolding!(); diff --git a/livekit-telemetry/src/transport.rs b/livekit-telemetry/src/transport.rs new file mode 100644 index 000000000..1d29281a5 --- /dev/null +++ b/livekit-telemetry/src/transport.rs @@ -0,0 +1,568 @@ +// 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. + +use std::collections::HashMap; + +use prost::Message; + +/// One fully composed OTLP/HTTP export request. +/// +/// The core fills in the URL, the headers (content type, auth, …) and the protobuf body; +/// a transport only moves the bytes. Non-HTTP transports (e.g. a data channel) may ignore +/// `url`/`headers` and forward `body`, which is a standard `ExportLogsServiceRequest`. +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +#[derive(Debug, Clone, PartialEq)] +pub struct ExportRequest { + pub url: String, + pub headers: HashMap, + pub body: Vec, +} + +/// What the collector answered. A transport moves bytes both ways and never interprets them: +/// status classification, `Retry-After`, the `google.rpc.Status` body and the Cloud "disabled" +/// contract are read once, here, for every platform ([`ExportError::from_response`]). +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +#[derive(Debug, Clone, PartialEq, Default)] +pub struct ExportResponse { + pub status: u16, + pub headers: HashMap, + pub body: Vec, +} + +impl ExportResponse { + /// `2xx`, nothing else: what an accepting collector answers. + pub fn accepted() -> Self { + Self { status: 200, ..Self::default() } + } +} + +/// LiveKit Cloud's answer when the project owner switched data recording off (401 or 403, plain +/// text, no machine-readable code yet). Matched as a phrase, not a word. +const DISABLED_ANSWER: &str = "data recording is disabled by owner"; + +/// How long a 429 that names no delay holds uploads: LiveKit Cloud's quota answer names none. +const THROTTLE_DEFAULT_MS: u64 = 60_000; + +/// Why a transport could not get an answer. A 4xx/5xx is an answer, not an error: return it as an +/// [`ExportResponse`] and the core classifies it. +#[cfg_attr(feature = "uniffi", derive(uniffi::Error))] +#[derive(Debug, Clone, PartialEq, thiserror::Error)] +pub enum ExportError { + /// No response: connection refused or reset, DNS or TLS failure, offline, timed out. The + /// batch stays cached and is retried with backoff. `retry_after_ms` is honored when set. + #[error("retryable export error: {reason}")] + Retryable { reason: String, retry_after_ms: Option }, + /// The request can never succeed (an invalid URL): the batch is dropped. + #[error("export rejected: {reason}")] + Rejected { reason: String }, + /// Telemetry is off for this project: the project goes silent. + #[error("telemetry disabled by the collector")] + Disabled, +} + +/// A foreign transport threw something its binding did not declare (a Swift `Error`, a Kotlin +/// `RuntimeException`): no answer, retried with backoff — never a panic in the exporter. +#[cfg(feature = "uniffi")] +impl From for ExportError { + fn from(error: uniffi::UnexpectedUniFFICallbackError) -> Self { + Self::Retryable { + reason: format!("transport threw: {}", error.reason), + retry_after_ms: None, + } + } +} + +/// What a collector's answer means for the batch: OTLP/HTTP failure semantics, plus the answers +/// LiveKit Cloud gives on top of them. +#[derive(Debug, Clone, PartialEq)] +pub(crate) enum Verdict { + /// 2xx. `rejected` counts the records an OTLP partial success refused (never retried). + Accepted { rejected: u64, reason: String }, + /// 400, 3xx (a credential-bearing request is never redirected), and every other 4xx/5xx the + /// OTLP spec does not call retryable: drop the batch. + Rejected(String), + /// 413: split the batch and retry the halves, down to single records. + TooLarge, + /// 401/403 that is not the "disabled" answer: the credential is the problem, not the data — + /// hold the project's batches until the platform hands over a new token. + Unauthorized(String), + /// 404: the host has no client ingest; the project goes silent. + NotFound, + /// 401/403 "data recording is disabled by owner": the project goes silent and its cache is + /// purged. + Disabled, + /// 429, or 503 naming a delay: pause every upload for `delay_ms`, keep collecting. + Throttled { delay_ms: u64, reason: String }, + /// 502/503/504, or any error carrying `RetryInfo`: the server failed transiently. Retry the + /// batch after `delay_ms` or a backoff; a batch that keeps failing is dropped eventually. + Retry { delay_ms: Option, reason: String }, +} + +impl Verdict { + /// Classify a collector response. + pub(crate) fn of(response: &ExportResponse) -> Self { + let status = response.status; + if (200..300).contains(&status) { + return match PartialSuccessResponse::decode(response.body.as_slice()) { + Ok(PartialSuccessResponse { partial_success: Some(p) }) if p.rejected > 0 => { + Self::Accepted { rejected: p.rejected as u64, reason: p.error_message } + } + _ => Self::Accepted { rejected: 0, reason: String::new() }, + }; + } + // OTLP/HTTP errors are a protobuf `google.rpc.Status`; Cloud's auth answers are text. + let rpc = RpcStatus::decode(response.body.as_slice()).ok(); + let text = match rpc.as_ref().map(|s| s.message.trim()).filter(|m| !m.is_empty()) { + Some(message) => message.to_owned(), + None => String::from_utf8_lossy(&response.body).trim().chars().take(200).collect(), + }; + // `reason`, not `message`: a UniFFI field named `message` collides with + // `Throwable.message` in Kotlin. + let reason = if text.is_empty() { + format!("HTTP {status}") + } else { + format!("HTTP {status}: {text}") + }; + if matches!(status, 401 | 403) { + return if text.to_ascii_lowercase().contains(DISABLED_ANSWER) { + Self::Disabled + } else { + Self::Unauthorized(reason) + }; + } + let header = response + .headers + .iter() + .find(|(name, _)| name.eq_ignore_ascii_case("retry-after")) + .and_then(|(_, value)| retry_after_ms(value)); + let retry_info = rpc.as_ref().and_then(retry_info_ms); + let delay = header.or(retry_info); + // The status decides whether a batch is retried; a delay hint only says when. A hint on a + // final status (400, 422, 501, a redirect) changes nothing. + match status { + 404 => Self::NotFound, + 413 => Self::TooLarge, + // A rate limit is an instruction to stop, so 429 always yields a wait: LiveKit Cloud + // answers a quota check with a bare `ResourceExhausted` — no `Retry-After`, no + // `RetryInfo` — and without one the exporter would spend its retries hammering the + // endpoint that just asked for quiet. + 429 => Self::Throttled { delay_ms: delay.unwrap_or(THROTTLE_DEFAULT_MS), reason }, + 503 if delay.is_some() => { + Self::Throttled { delay_ms: delay.unwrap_or_default(), reason } + } + 502..=504 => Self::Retry { delay_ms: delay, reason }, + // LiveKit Cloud's one retryable 500 carries `RetryInfo` (custom: OTLP retries only + // 429/502/503/504). + 500 if retry_info.is_some() => Self::Retry { delay_ms: retry_info, reason }, + _ => Self::Rejected(reason), + } + } +} + +/// `ExportLogsServiceResponse` and `ExportTraceServiceResponse` share one shape: field 1 is the +/// partial success, whose field 1 counts the rejected records and field 2 says why. +#[derive(Clone, PartialEq, prost::Message)] +struct PartialSuccessResponse { + #[prost(message, optional, tag = "1")] + partial_success: Option, +} + +#[derive(Clone, PartialEq, prost::Message)] +struct PartialSuccess { + #[prost(int64, tag = "1")] + rejected: i64, + #[prost(string, tag = "2")] + error_message: String, +} + +/// `google.rpc.Status`, the OTLP/HTTP error body. Two messages, three fields: not worth a +/// generated crate. +#[derive(Clone, PartialEq, prost::Message)] +struct RpcStatus { + #[prost(int32, tag = "1")] + code: i32, + #[prost(string, tag = "2")] + message: String, + #[prost(message, repeated, tag = "3")] + details: Vec, +} + +/// `google.rpc.RetryInfo`. +#[derive(Clone, PartialEq, prost::Message)] +struct RetryInfo { + #[prost(message, optional, tag = "1")] + retry_delay: Option, +} + +/// RFC 9110 §10.2.3 `Retry-After`: delay-seconds or an IMF-fixdate. Untrusted input: garbage is +/// ignored, a date in the past means now; the caller clamps the far end. +fn retry_after_ms(value: &str) -> Option { + let value = value.trim(); + // delay-seconds: digits only; more than a u64 holds saturates (the exporter clamps it). + if !value.is_empty() && value.bytes().all(|b| b.is_ascii_digit()) { + return Some(value.parse::().unwrap_or(u64::MAX).saturating_mul(1000)); + } + let at = imf_fixdate_secs(value)?; + let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).ok()?.as_secs(); + Some(at.saturating_sub(now).saturating_mul(1000)) +} + +/// `Sun, 06 Nov 1994 08:49:37 GMT` → unix seconds. The obsolete RFC 850 / asctime forms are not +/// accepted (RFC 9110 lets a recipient ignore them). +fn imf_fixdate_secs(value: &str) -> Option { + let mut parts = value.split_ascii_whitespace(); + let (_weekday, day, month, year, time, zone) = + (parts.next()?, parts.next()?, parts.next()?, parts.next()?, parts.next()?, parts.next()?); + if zone != "GMT" || parts.next().is_some() { + return None; + } + const MONTHS: [&str; 12] = + ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]; + let month = MONTHS.iter().position(|m| *m == month)? as i64 + 1; + let (day, year): (i64, i64) = (day.parse().ok()?, year.parse().ok()?); + let mut hms = time.split(':').map(|n| n.parse::().ok()); + let (h, m, s) = (hms.next()??, hms.next()??, hms.next()??); + // IMF-fixdate has a four-digit year; bounding it keeps the arithmetic below in range. + if !(1970..=9999).contains(&year) + || hms.next().is_some() + || !(1..=31).contains(&day) + || !(0..24).contains(&h) + || !(0..60).contains(&m) + || !(0..61).contains(&s) + { + return None; + } + // Days from the civil date (proleptic Gregorian), after Howard Hinnant's algorithm. + let y = if month <= 2 { year - 1 } else { year }; + let era = y.div_euclid(400); + let yoe = y - era * 400; + let doy = (153 * (month + if month > 2 { -3 } else { 9 }) + 2) / 5 + day - 1; + let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy; + let days = era * 146_097 + doe - 719_468; + u64::try_from(days * 86_400 + h * 3600 + m * 60 + s).ok() +} + +fn retry_info_ms(status: &RpcStatus) -> Option { + status + .details + .iter() + .filter(|detail| detail.type_url.ends_with("google.rpc.RetryInfo")) + .find_map(|detail| RetryInfo::decode(detail.value.as_slice()).ok()?.retry_delay) + .map(|delay| { + (delay.seconds.max(0) as u64) + .saturating_mul(1000) + .saturating_add(delay.nanos.clamp(0, 999_999_999) as u64 / 1_000_000) + }) +} + +/// Moves an encoded batch off the device and hands back whatever came back. +/// +/// Implemented in Rust ([`NetTransport`], feature `net`) or by the host platform through UniFFI. +/// The contract: +/// +/// - POST `url` with `headers` and `body` as given, and return the response whatever its status; +/// fail only when there was no response (connection, DNS, TLS, timeout) — +/// [`ExportError::Retryable`] — or the request cannot be made at all +/// ([`ExportError::Rejected`], e.g. an invalid URL). +/// - Never retry: the core owns the retry policy. +/// - Never forward the `Authorization` header across origins: do not follow a redirect to +/// another scheme, host or port with it (returning the 3xx is fine — the core drops the batch). +/// `NetTransport` over `livekit-net`'s native client strips it on cross-origin redirects. +/// - Keep the request off the call's critical path: lowest priority where the platform allows it. +#[cfg_attr(feature = "uniffi", uniffi::export(with_foreign))] +#[async_trait::async_trait] +pub trait TelemetryTransport: Send + Sync { + async fn send(&self, request: ExportRequest) -> Result; +} + +#[cfg(feature = "net")] +mod net { + use std::sync::Arc; + + use livekit_net::{Header, HttpClient, HttpClientExt, TransportError}; + + use super::{ExportError, ExportRequest, ExportResponse, TelemetryTransport}; + + /// Default transport: HTTP POST through a `livekit-net` client — the native backend, or + /// whatever the host registered with `livekit_net::set_http_client`. + pub struct NetTransport(Arc); + + impl NetTransport { + /// Post through `client`. + pub fn new(client: Arc) -> Self { + Self(client) + } + + /// Resolve the process-wide `livekit-net` client; `None` when none is available. + pub fn from_registry() -> Option { + livekit_net::http_client().map(Self) + } + } + + #[async_trait::async_trait] + impl TelemetryTransport for NetTransport { + async fn send(&self, request: ExportRequest) -> Result { + let headers = + request.headers.into_iter().map(|(name, value)| Header { name, value }).collect(); + match self.0.post(request.url, headers, request.body).await { + Ok(response) => Ok(ExportResponse { + status: response.status, + headers: response.headers.into_iter().map(|h| (h.name, h.value)).collect(), + body: response.body, + }), + // The client swallowed the body with the status; the core still classifies it. + Err(TransportError::Http { status }) => { + Ok(ExportResponse { status, ..Default::default() }) + } + Err(other) => { + Err(ExportError::Retryable { reason: other.to_string(), retry_after_ms: None }) + } + } + } + } +} + +#[cfg(feature = "net")] +pub use net::NetTransport; + +#[cfg(test)] +mod tests { + use prost::Message; + + use super::*; + + fn status_body(message: &str, retry: Option<(i64, i32)>) -> Vec { + let details = retry + .map(|(seconds, nanos)| prost_types::Any { + type_url: "type.googleapis.com/google.rpc.RetryInfo".into(), + value: RetryInfo { retry_delay: Some(prost_types::Duration { seconds, nanos }) } + .encode_to_vec(), + }) + .into_iter() + .collect(); + RpcStatus { code: 8, message: message.into(), details }.encode_to_vec() + } + + fn response(status: u16, headers: &[(&str, &str)], body: Vec) -> ExportResponse { + ExportResponse { + status, + headers: headers.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect(), + body, + } + } + + fn verdict(status: u16, headers: &[(&str, &str)], body: Vec) -> Verdict { + Verdict::of(&response(status, headers, body)) + } + + #[test] + fn success_and_partial_success() { + assert_eq!( + Verdict::of(&ExportResponse::accepted()), + Verdict::Accepted { rejected: 0, reason: String::new() } + ); + assert!(matches!(verdict(204, &[], vec![]), Verdict::Accepted { rejected: 0, .. })); + let partial = PartialSuccessResponse { + partial_success: Some(PartialSuccess { rejected: 3, error_message: "too old".into() }), + } + .encode_to_vec(); + assert_eq!( + verdict(200, &[], partial), + Verdict::Accepted { rejected: 3, reason: "too old".into() } + ); + assert!(matches!(verdict(200, &[], b"not protobuf".to_vec()), Verdict::Accepted { .. })); + } + + #[test] + fn client_errors() { + assert_eq!( + verdict(400, &[], status_body("bad batch", None)), + Verdict::Rejected("HTTP 400: bad batch".into()) + ); + assert_eq!(verdict(413, &[], vec![]), Verdict::TooLarge); + assert!(matches!(verdict(307, &[], vec![]), Verdict::Rejected(_)), "never follow"); + assert!(matches!(verdict(422, &[], vec![]), Verdict::Rejected(_))); + assert_eq!(verdict(404, &[], vec![]), Verdict::NotFound); + for body in ["invalid token", "operation requires observability write grant"] { + assert_eq!( + verdict(401, &[], body.as_bytes().to_vec()), + Verdict::Unauthorized(format!("HTTP 401: {body}")) + ); + } + assert!(matches!(verdict(403, &[], vec![]), Verdict::Unauthorized(_))); + } + + #[test] + fn disabled_by_owner() { + let text = b"project data recording is disabled by owner".to_vec(); + assert_eq!(verdict(401, &[], text), Verdict::Disabled); + let rpc = status_body("Project data recording is disabled by owner", None); + assert_eq!(verdict(403, &[], rpc), Verdict::Disabled); + // Only the documented phrase: another answer that merely mentions "disabled" is a + // credential problem, not a project switch. + for body in ["account disabled", "token disabled for this room"] { + assert!(matches!( + verdict(401, &[], body.as_bytes().to_vec()), + Verdict::Unauthorized(_) + )); + } + } + + /// The status decides; a delay hint on a final status is ignored. Only 429, 502–504 and Cloud's + /// 500-with-RetryInfo are retried. + #[test] + fn delay_hints_never_make_a_final_status_retryable() { + type Hint<'a> = (&'a [(&'a str, &'a str)], Vec); + let hints: [Hint; 3] = [ + (&[("Retry-After", "5")], vec![]), + (&[], status_body("later", Some((5, 0)))), + (&[("Retry-After", "5")], status_body("later", Some((5, 0)))), + ]; + for (headers, body) in &hints { + for status in [300, 302, 307, 400, 405, 409, 410, 422, 501, 505, 599] { + assert!( + matches!(verdict(status, headers, body.clone()), Verdict::Rejected(_)), + "{status} with {headers:?}" + ); + } + assert_eq!(verdict(413, headers, body.clone()), Verdict::TooLarge); + assert_eq!(verdict(404, headers, body.clone()), Verdict::NotFound); + assert!(matches!(verdict(429, headers, body.clone()), Verdict::Throttled { .. })); + assert!(matches!(verdict(503, headers, body.clone()), Verdict::Throttled { .. })); + for status in [502, 504] { + assert!(matches!( + verdict(status, headers, body.clone()), + Verdict::Retry { delay_ms: Some(5_000), .. } + )); + } + } + // 500: only with RetryInfo in the body, never on a Retry-After header alone. + assert!(matches!(verdict(500, &[("Retry-After", "5")], vec![]), Verdict::Rejected(_))); + assert!(matches!( + verdict(500, &[], status_body("later", Some((5, 0)))), + Verdict::Retry { delay_ms: Some(5_000), .. } + )); + } + + /// Untrusted time values never panic or wrap; oversized ones saturate. + #[test] + fn untrusted_time_values_are_bounded() { + assert_eq!(retry_after_ms("99999999999999999999999"), Some(u64::MAX)); + assert_eq!(retry_after_ms("18446744073709551615"), Some(u64::MAX)); + assert_eq!(retry_after_ms("-5"), None); + assert_eq!(retry_after_ms(""), None); + for date in [ + "Sun, 06 Nov 99999999999999 08:49:37 GMT", + "Sun, 06 Nov -9223372036854775808 08:49:37 GMT", + "Sun, 06 Nov 1969 08:49:37 GMT", + "Sun, 06 Nov 10000 08:49:37 GMT", + "Sun, 32 Nov 2026 08:49:37 GMT", + "Sun, 06 Nov 2026 08:49:37:99 GMT", + "Sun, 06 Nov 2026 99:49:37 GMT", + ] { + assert_eq!(imf_fixdate_secs(date), None, "{date}"); + } + assert_eq!(imf_fixdate_secs("Fri, 31 Dec 9999 23:59:59 GMT"), Some(253_402_300_799)); + let huge = verdict(429, &[("Retry-After", "99999999999999999999")], vec![]); + assert!(matches!(huge, Verdict::Throttled { delay_ms: u64::MAX, .. }), "clamped later"); + let far = verdict(429, &[], status_body("q", Some((i64::MAX, 999_999_999)))); + assert!(matches!(far, Verdict::Throttled { delay_ms: u64::MAX, .. })); + } + + #[test] + fn throttling_takes_the_header_then_the_body_then_the_default() { + let both = verdict(429, &[("Retry-After", "7")], status_body("quota", Some((30, 0)))); + assert!(matches!(both, Verdict::Throttled { delay_ms: 7_000, .. })); + let body = verdict(429, &[], status_body("quota", Some((30, 500_000_000)))); + assert!(matches!(body, Verdict::Throttled { delay_ms: 30_500, .. })); + // What LiveKit Cloud answers over quota: ResourceExhausted, no header, no RetryInfo. + assert_eq!( + verdict(429, &[], status_body("QuotaStatusExceeded", None)), + Verdict::Throttled { delay_ms: 60_000, reason: "HTTP 429: QuotaStatusExceeded".into() } + ); + let unavailable = verdict(503, &[("retry-after", " 12 ")], vec![]); + assert!(matches!(unavailable, Verdict::Throttled { delay_ms: 12_000, .. })); + let past = verdict(429, &[("Retry-After", "Wed, 21 Oct 2015 07:28:00 GMT")], vec![]); + assert!(matches!(past, Verdict::Throttled { delay_ms: 0, .. }), "a past date: now"); + let garbage = verdict(429, &[("Retry-After", "soon-ish")], vec![]); + assert!(matches!(garbage, Verdict::Throttled { delay_ms: 60_000, .. }), "ignored"); + let negative = verdict(429, &[], status_body("q", Some((-5, -1)))); + assert!(matches!(negative, Verdict::Throttled { delay_ms: 0, .. }), "clamped"); + } + + #[test] + fn retry_after_http_dates() { + assert_eq!(imf_fixdate_secs("Sun, 06 Nov 1994 08:49:37 GMT"), Some(784_111_777)); + assert_eq!(imf_fixdate_secs("Thu, 01 Jan 1970 00:00:00 GMT"), Some(0)); + assert_eq!(imf_fixdate_secs("Tue, 29 Feb 2028 12:00:00 GMT"), Some(1_835_438_400)); + assert_eq!(imf_fixdate_secs("Sunday, 06-Nov-94 08:49:37 GMT"), None, "obsolete form"); + assert_eq!(imf_fixdate_secs("Sun, 06 Nov 1994 08:49:37 PST"), None); + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("clock") + .as_secs(); + let date = |secs: u64| { + let days = secs / 86_400; + let (h, m, s) = ((secs % 86_400) / 3600, (secs % 3600) / 60, secs % 60); + // Civil from days (inverse of the parser's algorithm) for the fixture only. + let z = days as i64 + 719_468; + let era = z.div_euclid(146_097); + let doe = z - era * 146_097; + let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let d = doy - (153 * mp + 2) / 5 + 1; + let mo = if mp < 10 { mp + 3 } else { mp - 9 }; + let y = yoe + era * 400 + if mo <= 2 { 1 } else { 0 }; + const M: [&str; 12] = [ + "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec", + ]; + format!("Mon, {d:02} {} {y} {h:02}:{m:02}:{s:02} GMT", M[mo as usize - 1]) + }; + let ms = retry_after_ms(&date(now + 120)).expect("date"); + assert!((118_000..=120_000).contains(&ms), "{ms}"); + } + + #[test] + fn server_errors() { + for status in [502, 503, 504] { + assert!(matches!(verdict(status, &[], vec![]), Verdict::Retry { delay_ms: None, .. })); + } + assert!(matches!( + verdict(502, &[("Retry-After", "3")], vec![]), + Verdict::Retry { delay_ms: Some(3_000), .. } + )); + // Cloud's retryable 500 carries RetryInfo; a bare 500 is final, per OTLP/HTTP. + let internal = verdict(500, &[], status_body("try later", Some((5, 0)))); + assert!(matches!(internal, Verdict::Retry { delay_ms: Some(5_000), .. })); + assert!(matches!(verdict(500, &[], vec![]), Verdict::Rejected(_))); + assert!(matches!(verdict(501, &[], vec![]), Verdict::Rejected(_))); + } + + /// An exception the foreign transport did not declare becomes a retryable transport error. + /// Without the conversion UniFFI panics on the exporter's thread. + #[cfg(feature = "uniffi")] + #[test] + fn an_undeclared_foreign_exception_is_a_retryable_failure() { + let lifted = as uniffi::LiftReturn< + crate::UniFfiTag, + >>::handle_callback_unexpected_error( + uniffi::UnexpectedUniFFICallbackError::new("java.lang.IllegalStateException: boom"), + ); + assert!(matches!( + lifted, + Err(ExportError::Retryable { retry_after_ms: None, ref reason }) if reason.contains("boom") + )); + } +} From 5296abbf95dddabc7dba4d21d0c61da3597d1c58 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:30:15 +0200 Subject: [PATCH 2/3] docs(telemetry): specify destinations and credentials --- livekit-telemetry/SPEC.md | 49 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/livekit-telemetry/SPEC.md b/livekit-telemetry/SPEC.md index 1c58613d3..9e09f63e9 100644 --- a/livekit-telemetry/SPEC.md +++ b/livekit-telemetry/SPEC.md @@ -24,6 +24,55 @@ Set once per pipeline (`TelemetryConfig.resource`): | `device.model.identifier` | platform SDK | `iPhone16,1` | | `telemetry.sdk.name/language/version` | core | `livekit-telemetry`, `rust`, `0.1.0` | +## Pipeline, scopes and destination + +### Destination and credentials + +The platform passes exactly two things per room, through `Scope::set_server(url, token)`: the +LiveKit server URL the room connects to and the participant token it connects with — at connect, +and again with **every** refreshed token (the SFU sends one right after join, then every few +minutes). The call is cheap and idempotent. There is no client-side endpoint, header or sink. + +- **Ingest URL:** derived from the server URL's host, `https:///observability/client/{logs,traces}/otlp/v0`, + for LiveKit Cloud hosts only (`*.livekit.cloud`, including `*.staging.livekit.cloud`). Any other + host (self-hosted, OSS) has no ingest: one local warning, and that room's records are dropped + at the door instead of cached. +- **Token:** `Authorization: Bearer `. The core reads the token's *unverified* claims: the + observability grant (`observability.write` or `observability.clientWrite`) and `exp`. It never + sends a token known to be expired or refused; batches wait for the next token instead + (a hard hold). A project whose first token has no grant never opted in: nothing is collected + for it. A refreshed token without the grant (today's SFU drops it) does not replace a granted + one that is still valid. Tokens live in memory only — never in a batch, never on disk. +- **Server URL validation:** parsed with WHATWG URL rules, never string-matched. A token is only + sent to `https://.livekit.cloud/…` built from the parsed host alone: TLS scheme + (`wss`/`https`), a domain under the Cloud suffix with a label of its own, the default port, no + userinfo. +- **Ownership:** every record captures its owner when it is captured — its session and the + project that session is routed to at that moment — and keeps it: a Room that reconnects to + another project takes nothing queued or cached along. Credentials are keyed by (project, + session): a live Room uploads with its **own** latest token for that project, never another + Room's. A Room's records captured before it had a server go to its own first project, never to + another Room's. Process-level records (device state, pre-room errors, self-telemetry) go to + the project most recently handed a token, with that project's latest token; so do batches from + a previous launch, which wait — up to the 24 h age limit — for a token of the same project. + The answer to a request is attributed to the project it was sent to — its 404, disable or + pause never lands on another project. A session's credentials stay while the session is alive + (its Room, or records, windows or spans still referencing it) — for every project it was routed + to, so records captured for an earlier project can still be sent; that is the residual: a live + Room keeps one credential per project it has used — and while cached batches need them; + project-level copies only while a live session is routed there or the backlog has batches for + it. A token the collector refused is recorded by identity (a hash) and never sent again from + any slot, until it expires (a token without `exp` stays refused for the process; like the + per-host project table, which keeps one small entry per host ever seen, that grows only with + what a process meets — bounded in practice, not by a cap). A past, + negative or non-numeric `exp` counts as expired. +- **Waiting** for a first destination or a usable token is uncapped, bounded only by the cache. + +Local end-to-end tests point everything at an OpenTelemetry collector of their own with the +`LK_TELEMETRY_ENDPOINT` environment variable, read by the core at start (a base URL gets +`/v1/logs` and `/v1/traces`; a URL ending in `logs` is used as is). It is not part of any platform +API. With it, every batch goes there without Cloud rules or tokens. + ## Events An event with no `body` is exported with its name as the body as well as in `event_name`: log From 3e68d917447e6e4f93d6312a0a39eb0fd0c8a869 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 13:36:46 +0200 Subject: [PATCH 3/3] fix(telemetry): key non-Cloud URLs apart and tighten the transport docs A non-Cloud server URL now gets a routing key of its own (scheme, host and port), so a plaintext or userinfo URL on a Cloud host can no longer revive that project with its token or silence it. `alive` takes the session, so a Room without a server waits for its first project instead of being judged as process data. The test override swaps only a whole `logs` path segment. The project table drops entries that no live session or cached batch needs, except a project known to receive nothing (not Cloud, no grant, 404, disabled). A Cloud host must be a DNS hostname, and every other key escapes its host and port, so a key is a valid file name on every platform: it ends each cached batch's id. The `Server` and `set` docs describe both kinds of key. A Room's first project is recorded even when that project receives nothing, so what the Room captured before its first server is dropped with it instead of moving to a later project. The Cloud "disabled" phrase is matched on the whole error body, and only the logged reason is cut, to a named limit. `ExportError` loses `Disabled`, because a transport must not interpret the answer. `NetTransport::new` is crate-private, the public exports are listed, and `ExportResponse` and `Token` get `From` conversions. Doc comments now name the real classifier, the real body types and the real redirect behavior, and every public field has a doc line. The README names the transport and is now the crate doc, and a changeset records the stage. The `Retry-After` date parser checks the day against the length of the month, leap years included, and the retry verdict names 500 with `RetryInfo` as the only retryable 500. The native `livekit-net` HTTP client keeps its limit of 10 redirects but no longer follows one that changes the scheme, so `Authorization` never reaches a plaintext URL on the same host and port; a `livekit-net` changeset records it. In the code the crate first shipped with, the queue-full warning is logged after the queue lock, a pipeline attribute can no longer add a second `otel.event.name`, a record ships each attribute key once (of a key the caller repeats, the last value wins; the session's keys still replace the caller's), and report counts saturate at `i64::MAX` instead of wrapping. Finished spans wait in a `VecDeque`, `reserved` is `is_reserved`, the severity text is `Severity::as_str`, and `Spans::dropped` and `SpanKind` are crate-private. Every exported field and variant has a doc line, the limits and the size of a number are named constants, the duplicated wire values and the test hooks say why they exist, and the docs and SPEC state the real log floor, trace id, failure count and `queue_full` meaning. --- .changeset/livekit-net-redirect-scheme.md | 5 + .changeset/telemetry-transport.md | 5 + livekit-net/src/native/mod.rs | 61 +++++- livekit-telemetry/Cargo.toml | 4 +- livekit-telemetry/README.md | 3 + livekit-telemetry/SPEC.md | 39 ++-- livekit-telemetry/src/destination.rs | 236 +++++++++++++++++++--- livekit-telemetry/src/event.rs | 71 +++++-- livekit-telemetry/src/lib.rs | 25 ++- livekit-telemetry/src/otlp.rs | 90 +++++++-- livekit-telemetry/src/span.rs | 35 ++-- livekit-telemetry/src/stats.rs | 81 ++++++-- livekit-telemetry/src/store.rs | 40 +++- livekit-telemetry/src/transport.rs | 133 +++++++----- 14 files changed, 654 insertions(+), 174 deletions(-) create mode 100644 .changeset/livekit-net-redirect-scheme.md create mode 100644 .changeset/telemetry-transport.md diff --git a/.changeset/livekit-net-redirect-scheme.md b/.changeset/livekit-net-redirect-scheme.md new file mode 100644 index 000000000..6ccefaa99 --- /dev/null +++ b/.changeset/livekit-net-redirect-scheme.md @@ -0,0 +1,5 @@ +--- +livekit-net: patch +--- + +The native HTTP client no longer follows a redirect that changes the scheme, so an `Authorization` header never reaches a plaintext URL on the same host and port. diff --git a/.changeset/telemetry-transport.md b/.changeset/telemetry-transport.md new file mode 100644 index 000000000..b2b6ee119 --- /dev/null +++ b/.changeset/telemetry-transport.md @@ -0,0 +1,5 @@ +--- +livekit-telemetry: minor +--- + +Add the `TelemetryTransport` seam, a `net` HTTP transport over `livekit-net`, and LiveKit Cloud destination routing from a Room's server URL and token. diff --git a/livekit-net/src/native/mod.rs b/livekit-net/src/native/mod.rs index 6ca7ef78d..e0cb48bd6 100644 --- a/livekit-net/src/native/mod.rs +++ b/livekit-net/src/native/mod.rs @@ -42,10 +42,29 @@ pub struct NativeTransport { impl NativeTransport { pub(crate) fn new() -> Self { - Self { http: reqwest::Client::new() } + let http = reqwest::Client::builder() + .redirect(redirect_policy()) + .build() + // `reqwest::Client::new` panics the same way. + .expect("the TLS backend initializes"); + Self { http } } } +/// reqwest's default policy (at most 10 redirects), except that a redirect to another scheme is +/// returned instead of followed: reqwest drops `Authorization` only when the host or port +/// changes, so an `https` to `http` redirect on port 443 would send a credential in plaintext. +fn redirect_policy() -> reqwest::redirect::Policy { + reqwest::redirect::Policy::custom(|attempt| { + let from = attempt.previous().last().map(url::Url::scheme); + if from.is_some_and(|scheme| scheme != attempt.url().scheme()) { + attempt.stop() + } else { + reqwest::redirect::Policy::default().redirect(attempt) + } + }) +} + /// The shared default transport behind the registry fallback. Memoized because /// consumers resolve a client per request: a fresh transport per resolution /// would defeat its connection pooling. @@ -139,3 +158,43 @@ impl HttpClient for NativeTransport { } } } + +#[cfg(test)] +mod tests { + use std::{ + io::{Read, Write}, + net::TcpListener, + time::Duration, + }; + + use super::*; + + /// A redirect that changes only the scheme, on the same port, comes back as the 3xx: the + /// request with its `Authorization` header is not sent again under the new scheme. + #[tokio::test] + async fn a_redirect_to_another_scheme_is_not_followed() { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let port = listener.local_addr().expect("address").port(); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("the first request"); + let _ = stream.read(&mut [0; 4096]); + let answer = format!( + "HTTP/1.1 307 Temporary Redirect\r\nLocation: https://127.0.0.1:{port}/v1/logs\r\n\ + Content-Length: 0\r\nConnection: close\r\n\r\n" + ); + stream.write_all(answer.as_bytes()).expect("answer"); + drop(stream); + listener.set_nonblocking(true).expect("non-blocking"); + std::thread::sleep(Duration::from_millis(300)); + listener.accept().is_ok() + }); + let headers = vec![Header { name: "Authorization".into(), value: "Bearer token".into() }]; + let url = format!("http://127.0.0.1:{port}/v1/logs"); + let response = NativeTransport::new() + .request(HttpMethod::Post, url, headers, Some(b"batch".to_vec())) + .await + .expect("the redirect itself comes back"); + assert_eq!(response.status, 307); + assert!(!server.join().expect("server"), "nothing was sent to the new scheme"); + } +} diff --git a/livekit-telemetry/Cargo.toml b/livekit-telemetry/Cargo.toml index ad1fe9df6..29eb5d1d5 100644 --- a/livekit-telemetry/Cargo.toml +++ b/livekit-telemetry/Cargo.toml @@ -28,8 +28,8 @@ livekit-net = { workspace = true, optional = true } uniffi = { workspace = true, features = ["scaffolding-ffi-buffer-fns"], optional = true } [features] -# Default HTTP transport over the pluggable `livekit-net` client (native backend, or one -# the host registered with `livekit_net::set_http_client`). +# HTTP transport over a `livekit-net` client. The host registers one with +# `livekit_net::set_http_client`, or the build enables `livekit-net/native` and a TLS feature. net = ["dep:livekit-net"] uniffi = ["dep:uniffi"] diff --git a/livekit-telemetry/README.md b/livekit-telemetry/README.md index 83ca1fa3a..1128d148e 100644 --- a/livekit-telemetry/README.md +++ b/livekit-telemetry/README.md @@ -12,6 +12,9 @@ It currently holds the shared data model and its encoding: (trace) it belongs to; every drop is counted. - **Encoding.** [OTLP/HTTP](https://opentelemetry.io/docs/specs/otlp/) protobuf logs and traces, using the `opentelemetry-proto` message types only (no gRPC stack). +- **Transport and destinations.** `TelemetryTransport`, the seam that moves an encoded batch off + the device (`NetTransport` over `livekit-net` with the `net` feature), and the routing that + turns a Room's server URL and token into its LiveKit Cloud ingest URL and credential. The pipeline that batches, caches and uploads records is not part of the crate yet. diff --git a/livekit-telemetry/SPEC.md b/livekit-telemetry/SPEC.md index 9e09f63e9..ffe4c89e7 100644 --- a/livekit-telemetry/SPEC.md +++ b/livekit-telemetry/SPEC.md @@ -8,9 +8,11 @@ Every record carries a wall-clock timestamp: one without `timestamp_ns` is stamp captured (queued), never at export; an explicit `timestamp_ns` is kept. Its attributes, like its owner and timestamp, are taken at capture, its session's (`lk.room.*`, `lk.participant.*`, `session.id`) included, so a record captured before a disconnect keeps its room; only -pipeline-wide attributes are added at export. Attribute keys the SDK owns are `lk.*`, `otel.*`, -`code.*`, `session.id` and `error.type`: an app's attributes and custom events cannot set them, -and on a span only the core writes `lk.outcome` and `error.type` (see [Spans](#spans)). +pipeline-wide attributes are added at export. A record ships each key once: of a key it repeats, +the last value supplied wins, and its session's keys replace its own. Attribute keys the SDK +owns are `lk.*`, `otel.*`, `code.*`, `session.id` and `error.type`: an app's attributes and +custom events cannot set them, and on a span only the core writes `lk.outcome` and `error.type` +(see [Spans](#spans)). ## Resource attributes @@ -62,16 +64,22 @@ minutes). The call is cheap and idempotent. There is no client-side endpoint, he Room keeps one credential per project it has used — and while cached batches need them; project-level copies only while a live session is routed there or the backlog has batches for it. A token the collector refused is recorded by identity (a hash) and never sent again from - any slot, until it expires (a token without `exp` stays refused for the process; like the - per-host project table, which keeps one small entry per host ever seen, that grows only with - what a process meets — bounded in practice, not by a cap). A past, - negative or non-numeric `exp` counts as expired. + any slot, until it expires (a token without `exp` stays refused for the process). A past, + negative or non-numeric `exp` counts as expired. A project's entry goes once no live session + and no cached batch needs it. An entry known to receive nothing (not Cloud, no grant, 404, + disabled) stays for the process. +- **Routing key:** a Cloud project is keyed by its host, which must be a DNS hostname (letters, + digits, `-` and `.`). Any other URL is keyed by its scheme, host and port, so a plaintext or + userinfo URL on a Cloud host can neither feed that project a token nor silence it. Its host and + port are escaped (`%XX` for anything but a lowercase letter, a digit, `.`, `-` or `_`): the key + ends each cached batch's file name, which must be valid on every platform. - **Waiting** for a first destination or a usable token is uncapped, bounded only by the cache. Local end-to-end tests point everything at an OpenTelemetry collector of their own with the `LK_TELEMETRY_ENDPOINT` environment variable, read by the core at start (a base URL gets -`/v1/logs` and `/v1/traces`; a URL ending in `logs` is used as is). It is not part of any platform -API. With it, every batch goes there without Cloud rules or tokens. +`/v1/logs` and `/v1/traces`; a URL with a `logs` path segment is used as is, and its traces URL +replaces that segment). It is not part of any platform API. With it, every batch goes there +without Cloud rules or tokens. ## Events @@ -95,14 +103,14 @@ severity: info attributes: # counts since the previous report lk.telemetry.uploads.sent: int # batches accepted lk.telemetry.uploads.bytes: int # compressed bytes accepted — what telemetry cost the uplink - lk.telemetry.uploads.failed: int # attempts that failed transiently (no answer, 429, 5xx) + lk.telemetry.uploads.failed: int # attempts that failed transiently (a network error, 429 or 5xx, before the timeout) lk.telemetry.cache.batches: int # batches waiting in the cache right now (a gauge) # the rest only when non-zero: lk.telemetry.uploads.timeouts: int # attempts that hit export_timeout_ms lk.telemetry.uploads.unauthorized: int # 401/403 answers: tokens the collector refused lk.telemetry.holds.capped: int # soft holds that reached the 60 s cap lk.telemetry.cache.write_errors: int # batches the disk refused (full, gone), kept in memory - lk.telemetry.dropped.queue_full: int # records evicted from the in-memory queue + lk.telemetry.dropped.queue_full: int # records evicted from the in-memory queue, and spans dropped from a full span buffer lk.telemetry.dropped.cache_error: int # records no cache, not even memory, could take lk.telemetry.dropped.cache_full: int # records evicted by the cache's size / file-count bound lk.telemetry.dropped.expired: int # records in batches past the 24 h age limit @@ -219,8 +227,8 @@ A `TelemetryEvent` with an empty `name` is a plain log record (OTLP log without (semconv), `lk.log.source` (`sdk` | `ffi` | `webrtc`) and `lk.log.logger` (type, module or file). The platform hands the core a typed `LogRecord` via `log(record)`; the core applies the floor: WebRTC only at `error`, the SDK and the core at the configured `log_severity`, the core's own telemetry module -never. Only `warn` and `error` records -leave the device; `trace`/`debug`/`info` are dropped in `emit`. +never. Only plain records at or above `log_severity` (default `warn`) leave the device. Lower ones +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: @@ -237,8 +245,9 @@ lines. ## Spans A span is **one attempt** at an operation. The scope (one Room connection lifetime, across -reconnects) is the trace; its id is generated by the core when the pipeline starts and rides on -every span and log record. Spans are exported when they end — never a long-lived scope span. +reconnects) is the trace. The core generates its id in `begin_scope`, and it rides on every span +and log record. Records outside any Room carry the id of the process session, which the pipeline +makes at start. Spans are exported when they end — never a long-lived scope span. | Rule | Value | |---|---| diff --git a/livekit-telemetry/src/destination.rs b/livekit-telemetry/src/destination.rs index 55d86d198..33316be43 100644 --- a/livekit-telemetry/src/destination.rs +++ b/livekit-telemetry/src/destination.rs @@ -34,9 +34,9 @@ use tokio::time::Instant; /// The environment variable that points every upload at a collector of your own — a local /// OpenTelemetry collector for end-to-end tests. Not part of any platform API: set it in the -/// test process's environment before the pipeline starts. Either a base URL -/// (`http://localhost:4318`, OTLP paths `/v1/logs` and `/v1/traces` are appended) or a full logs -/// URL ending in `logs` (its traces URL is derived by replacing that segment). +/// test process's environment before the pipeline starts. Either a URL with a `logs` path +/// segment (its traces URL replaces that segment with `traces`) or a base URL +/// (`http://localhost:4318`, OTLP paths `/v1/logs` and `/v1/traces` are appended). pub const ENDPOINT_OVERRIDE_ENV: &str = "LK_TELEMETRY_ENDPOINT"; /// The longest a token is believed to stay valid, whatever its `exp` says. @@ -59,12 +59,16 @@ pub(crate) struct Target { /// The project the batch is sent to (`None` for the test override): what its answer is /// attributed to — a 404, a disable, a pause. pub project: Option, + /// The OTLP logs endpoint. pub logs: String, + /// The OTLP traces endpoint. pub traces: String, + /// The bearer token to send, `None` for the test override. pub token: Option, } impl Target { + /// The endpoint for `signal`. pub fn url(&self, signal: Signal) -> &str { match signal { Signal::Logs => &self.logs, @@ -86,8 +90,9 @@ pub(crate) struct Token { id: u64, } -impl Token { - pub fn parse(raw: &str) -> Self { +impl From<&str> for Token { + /// Read a raw participant token's claims; a malformed one carries no grant. + fn from(raw: &str) -> Self { let claims = raw .split('.') .nth(1) @@ -128,7 +133,9 @@ impl Token { }; Self { raw: raw.to_owned(), grant, expires, id } } +} +impl Token { fn expired(&self) -> bool { self.expires.is_some_and(|at| Instant::now() >= at) } @@ -237,6 +244,8 @@ pub(crate) struct Destinations { } impl Destinations { + /// No projects yet. With `endpoint_override` (see [`ENDPOINT_OVERRIDE_ENV`]) every batch goes + /// there instead. pub fn new(endpoint_override: Option<&str>) -> Self { Self { endpoint_override: endpoint_override.filter(|e| !e.is_empty()).map(override_target), @@ -250,17 +259,21 @@ impl Destinations { self.rooms.get(owner).cloned() } + /// Whether every batch goes to the test override. pub fn has_override(&self) -> bool { self.endpoint_override.is_some() } /// Session `owner`'s server URL and token (at connect, and again on every refresh). Returns - /// the project host its batches carry, or `None` when the URL has no host. + /// the project key its batches carry, or `None` when the URL has no host. pub fn set(&mut self, url: &str, token: &str, owner: &str) -> Option { let Server { key: host, cloud } = Server::parse(url)?; if self.endpoint_override.is_some() { return Some(host); } + // The first project is the Room's even when it receives nothing: what the Room captured + // before it had a server belongs there, and is dropped with it. + self.rooms.entry(owner.to_owned()).or_insert_with(|| host.clone()); let project = self.projects.entry(host.clone()).or_default(); if project.dead.is_none() && !cloud { project.dead = Some(Dead::NotCloud); @@ -269,7 +282,7 @@ impl Destinations { if matches!(project.dead, Some(Dead::NotCloud | Dead::Disabled)) { return Some(host); } - let token = Token::parse(token); + let token = Token::from(token); if project.dead == Some(Dead::NotFound) { log::debug!("new token for {host}: trying its ingest again"); project.dead = None; @@ -288,7 +301,6 @@ impl Destinations { let mut own = self.sessions.remove(&key); offer(&mut own, &token, &self.refused); self.sessions.extend(own.map(|own| (key, own))); - self.rooms.entry(owner.to_owned()).or_insert_with(|| host.clone()); self.latest = Some(host.clone()); Some(host) } @@ -352,16 +364,17 @@ impl Destinations { !self.projects.is_empty() && self.projects.values().all(|p| p.dead.is_some()) } - /// Whether a session routed to `host` should still collect: `false` once its project is - /// known to receive nothing, so its records are dropped at the door instead of cached. - pub fn alive(&self, host: Option<&str>) -> bool { - self.endpoint_override.is_some() || !matches!(self.route(host, None), Route::Drop) + /// Whether session `owner`, routed to `host`, should still collect: `false` once its + /// project is known to receive nothing, so its records are dropped at the door instead of + /// cached. A Room without a server yet waits for its first project, like [`route`](Self::route). + pub fn alive(&self, host: Option<&str>, owner: Option<&str>) -> bool { + !matches!(self.route(host, owner), Route::Drop) } /// The collector refused `token` (401/403): it is not sent again, from any slot; batches that /// would use it wait for the next one. pub fn refuse(&mut self, token: &str) { - if self.refused.add(&Token::parse(token)) { + if self.refused.add(&Token::from(token)) { log::warn!("the collector refused a token; waiting for a new one"); } } @@ -396,14 +409,27 @@ impl Destinations { .retain(|owner, _| live.contains_key(owner) || backlog.iter().any(|(_, o)| o == owner)); let unrouted = backlog.iter().any(|(host, _)| host.is_none()); let latest = self.latest.clone(); - for (host, project) in &mut self.projects { + // A kept Room credential or first project keeps its project's entry too. + let held: std::collections::HashSet = self + .sessions + .keys() + .map(|(host, _)| host) + .chain(self.rooms.values()) + .cloned() + .collect(); + // An entry nothing needs goes: the table follows the live sessions and the backlog. A + // dead entry stays for the process, so a later Room or process-level record is not + // collected for a project that takes nothing. + self.projects.retain(|host, project| { let needed = live.values().any(|route| route.as_deref() == Some(host)) || backlog.iter().any(|(h, _)| h.as_deref() == Some(host)) - || (unrouted && latest.as_deref() == Some(host)); + || (unrouted && latest.as_deref() == Some(host)) + || held.contains(host); if !needed { project.latest = None; } - } + needed || project.dead.is_some() + }); } /// How many credentials are held, Room and project copies together. @@ -438,8 +464,10 @@ impl Destinations { } /// A server URL, parsed (WHATWG URL rules, the same a browser or `URLSession` applies): its -/// routing key — the canonical host, with the port when one is given — and whether it is a -/// LiveKit Cloud project a token may be sent to. +/// routing key (a Cloud project's host, or `+` and the escaped host and port for any other +/// URL, for example `ws+localhost%3A7880`) and whether it is a LiveKit Cloud project a token may be +/// sent to. The key ends every cached batch's id, which is a file name: it holds no character a +/// file name cannot hold on any platform. struct Server { key: String, cloud: bool, @@ -454,20 +482,38 @@ impl Server { None => host.clone(), }; // Credentials only ever go to `https://.livekit.cloud/…`, built from the - // validated host alone: a TLS scheme, a domain under the Cloud suffix with a label of its - // own, the default port, no userinfo. Anything else — a look-alike that only a string - // check would accept (`evil.example\.livekit.cloud` is host `evil.example`), a - // plaintext scheme, an odd port — is not Cloud. + // validated host alone: a TLS scheme, a domain of letters, digits, `-` and `.` (a DNS + // hostname) under the Cloud suffix with a label of its own, the default port, no + // userinfo. Anything else — a look-alike that only a string check would accept + // (`evil.example\.livekit.cloud` is host `evil.example`), a plaintext scheme, an odd port + // — is not Cloud. let cloud = matches!(url.scheme(), "wss" | "https") && matches!(url.host(), Some(url::Host::Domain(_))) + && host.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'.') && host.strip_suffix(CLOUD_SUFFIX).is_some_and(|label| !label.is_empty()) && url.port().is_none() && url.username().is_empty() && url.password().is_none(); + // A non-Cloud URL never shares a key with a Cloud host: it can neither feed nor kill it. + let key = if cloud { key } else { format!("{}+{}", url.scheme(), file_safe(&key)) }; Some(Self { key, cloud }) } } +/// `text` with every byte other than a lowercase letter, a digit, `.`, `-` or `_` written as +/// `%XX`: a port's `:`, the brackets and colons of an IPv6 address and the `*` or `"` a URL host +/// may hold are not allowed in a Windows file name. +fn file_safe(text: &str) -> String { + let mut safe = String::with_capacity(text.len()); + for byte in text.bytes() { + match byte { + b'a'..=b'z' | b'0'..=b'9' | b'.' | b'-' | b'_' => safe.push(byte as char), + _ => safe.push_str(&format!("%{byte:02X}")), + } + } + safe +} + /// `https:///observability/client/{logs,traces}/otlp/v0`: the client table set. fn ingest_url(host: &str, signal: Signal) -> String { let signal = match signal { @@ -477,10 +523,19 @@ fn ingest_url(host: &str, signal: Signal) -> String { format!("https://{host}/observability/client/{signal}/otlp/v0") } +/// The targets of [`ENDPOINT_OVERRIDE_ENV`]. fn override_target(endpoint: &str) -> Target { let endpoint = endpoint.trim_end_matches('/'); - let (logs, traces) = match endpoint.rsplit_once("logs") { - Some((before, after)) => (endpoint.to_owned(), format!("{before}traces{after}")), + // A logs URL has a whole `logs` path segment; its traces URL swaps that segment. + let traces = url::Url::parse(endpoint).ok().and_then(|mut url| { + let mut segments: Vec = url.path_segments()?.map(str::to_owned).collect(); + let logs = segments.iter().rposition(|s| s == "logs")?; + segments[logs] = "traces".to_owned(); + url.set_path(&segments.join("/")); + Some(url.to_string()) + }); + let (logs, traces) = match traces { + Some(traces) => (endpoint.to_owned(), traces), None => (format!("{endpoint}/v1/logs"), format!("{endpoint}/v1/traces")), }; Target { project: None, logs, traces, token: None } @@ -727,6 +782,131 @@ pub(crate) mod tests { assert!(matches!(local.route(host.as_deref(), Some("s")), Route::Send(_))); } + #[test] + fn the_override_swaps_only_a_whole_logs_segment() { + let targets = |endpoint| { + let target = send(Destinations::new(Some(endpoint)).route(None, None)); + (target.logs, target.traces) + }; + assert_eq!( + targets("http://logs-collector:4318"), + ( + "http://logs-collector:4318/v1/logs".to_owned(), + "http://logs-collector:4318/v1/traces".to_owned() + ) + ); + assert_eq!( + targets("http://collector:4318/catalogs"), + ( + "http://collector:4318/catalogs/v1/logs".to_owned(), + "http://collector:4318/catalogs/v1/traces".to_owned() + ) + ); + } + + /// A plaintext or userinfo URL on a Cloud host gets a routing key of its own: it never + /// revives the Cloud project with a token from an insecure connect, and never silences it. + #[test] + fn non_cloud_urls_never_share_a_routing_key_with_cloud() { + let cloud = "p.livekit.cloud"; + let mut d = Destinations::default(); + d.set("wss://p.livekit.cloud", &grantless(600), "a"); + let plain = d.set("ws://p.livekit.cloud", &granted(600), "b"); + assert_ne!(plain.as_deref(), Some(cloud)); + assert_eq!(d.route(Some(cloud), Some("b")), Route::Drop, "still never opted in"); + assert_eq!(d.route(plain.as_deref(), Some("b")), Route::Drop, "not Cloud"); + + let mut d = Destinations::default(); + d.set("wss://p.livekit.cloud", &granted(600), "a"); + d.kill(cloud, Dead::NotFound); + d.set("ws://p.livekit.cloud", &granted(700), "b"); + assert_eq!(d.route(Some(cloud), Some("a")), Route::Drop, "only a Cloud connect revives"); + + for first in ["ws://p.livekit.cloud", "wss://user@p.livekit.cloud"] { + let mut d = Destinations::default(); + d.set(first, &granted(600), "a"); + let host = d.set("wss://p.livekit.cloud", &granted(600), "b"); + assert!(matches!(d.route(host.as_deref(), Some("b")), Route::Send(_)), "{first}"); + } + } + + #[test] + fn a_room_without_a_project_waits_for_its_first_one() { + let mut d = Destinations::default(); + d.set("ws://localhost:7880", &granted(600), "a"); + assert!(!d.alive(None, None), "process data: only dead projects so far"); + assert!(d.alive(None, Some("b")), "a Room without a server waits"); + } + + #[test] + fn the_project_table_follows_the_live_sessions() { + let mut d = Destinations::default(); + for n in 0..1000 { + d.set(&format!("wss://p{n}.livekit.cloud"), &granted(600), &format!("room-{n}")); + } + d.kill("p0.livekit.cloud", Dead::Disabled); + d.retain_owners(&Default::default(), &Default::default()); + assert_eq!(d.projects.len(), 1, "only the owner's switch stays"); + let route = d.route(Some("p0.livekit.cloud"), Some("room-0")); + assert_eq!(route, Route::Drop, "and keeps the project silent"); + } + + #[test] + fn routing_keys_are_file_names_on_every_platform() { + // What Windows refuses in a file name: a routing key ends every cached batch's id. + let refused = |c: char| c.is_control() || r#"<>:"/\|?*"#.contains(c); + let urls = [ + "ws://localhost:7880", + "wss://[::1]:7880", + "wss://a*b\"c.example", + "wss://x*y.livekit.cloud", + "https://p.livekit.cloud:8443", + "wss://p.livekit.cloud", + ]; + let mut d = Destinations::default(); + let keys: Vec = + urls.iter().map(|url| d.set(url, &granted(600), "s").expect(url)).collect(); + assert!(keys.iter().all(|key| !key.contains(refused)), "{keys:?}"); + let distinct: std::collections::HashSet<&String> = keys.iter().collect(); + assert_eq!(distinct.len(), keys.len(), "distinct servers keep distinct keys: {keys:?}"); + assert_eq!(keys[5], "p.livekit.cloud", "a Cloud project is keyed by its host"); + } + + #[test] + fn early_records_belong_to_a_first_project_that_is_not_cloud() { + let mut d = Destinations::default(); + let token = granted(600); + let first = d.set("ws://localhost:7880", &token, "room").expect("a host"); + d.set("wss://later.livekit.cloud", &token, "room"); + assert_eq!(d.first_project("room"), Some(first)); + assert_eq!(d.route(None, Some("room")), Route::Drop, "dropped with that project"); + } + + #[test] + fn early_records_belong_to_a_first_project_that_is_disabled() { + let mut d = Destinations::default(); + let token = granted(600); + d.set("wss://disabled.livekit.cloud", &token, "other"); + d.kill("disabled.livekit.cloud", Dead::Disabled); + d.set("wss://disabled.livekit.cloud", &token, "room"); + d.set("wss://later.livekit.cloud", &token, "room"); + assert_eq!(d.first_project("room").as_deref(), Some("disabled.livekit.cloud")); + assert_eq!(d.route(None, Some("room")), Route::Drop, "dropped with that project"); + } + + #[test] + fn a_dead_project_outlives_its_room() { + for (url, token) in + [("ws://localhost:7880", granted(600)), ("wss://p.livekit.cloud", grantless(600))] + { + let mut d = Destinations::default(); + d.set(url, &token, "room"); + d.retain_owners(&Default::default(), &Default::default()); + assert_eq!(d.route(None, Some(PROCESS_OWNER)), Route::Drop, "{url}"); + assert!(!d.alive(None, Some(PROCESS_OWNER)), "{url}: process data is not collected"); + } + } + /// A past, negative, fractional-past or malformed `exp` is expired (a hard hold); only a /// missing one means no known expiry. #[tokio::test(start_paused = true)] @@ -758,11 +938,11 @@ pub(crate) mod tests { #[test] fn tokens_are_read_not_verified() { - assert!(Token::parse(&granted(60)).grant); + assert!(Token::from(granted(60).as_str()).grant); let client_write = jwt(serde_json::json!({ "observability": { "clientWrite": true } })); - let parsed = Token::parse(&client_write); + let parsed = Token::from(client_write.as_str()); assert!(parsed.grant && parsed.expires.is_none(), "no exp: no known expiry"); - assert!(!Token::parse("garbage").grant); - assert!(!Token::parse("a.!!!.c").grant); + assert!(!Token::from("garbage").grant); + assert!(!Token::from("a.!!!.c").grant); } } diff --git a/livekit-telemetry/src/event.rs b/livekit-telemetry/src/event.rs index 79cfb1bcc..a0dd3cfb1 100644 --- a/livekit-telemetry/src/event.rs +++ b/livekit-telemetry/src/event.rs @@ -25,10 +25,12 @@ use std::time::{SystemTime, UNIX_EPOCH}; pub struct TelemetryEvent { /// Event name. LiveKit-defined events use the `lk.` prefix (e.g. `lk.ping`); see `SPEC.md`. pub name: String, + /// How serious it is (OTLP severity number and text). pub severity: Severity, /// Optional human-readable message (the OTLP log record body). #[cfg_attr(feature = "uniffi", uniffi(default))] pub body: Option, + /// The record's own attributes; the session's are added when it is captured. pub attributes: Vec, /// Wall-clock time in nanoseconds since the Unix epoch. `None` is stamped when the record is /// queued, never at export. @@ -43,14 +45,7 @@ pub struct TelemetryEvent { impl TelemetryEvent { /// An `Info` event without attributes, stamped when emitted. pub fn new(name: impl Into) -> Self { - Self { - name: name.into(), - severity: Severity::Info, - body: None, - attributes: Vec::new(), - timestamp_ns: None, - span_id: None, - } + Self { name: name.into(), ..Self::default() } } /// Link this record to an in-flight span. @@ -109,21 +104,42 @@ impl TelemetryEvent { #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Default)] pub enum Severity { + /// OTel `TRACE` (1). Trace, + /// OTel `DEBUG` (5). Debug, + /// OTel `INFO` (9). #[default] Info, + /// OTel `WARN` (13). Warn, + /// OTel `ERROR` (17). Error, } +impl Severity { + /// The OTel severity text (`WARN`). + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::Trace => "TRACE", + Self::Debug => "DEBUG", + Self::Info => "INFO", + Self::Warn => "WARN", + Self::Error => "ERROR", + } + } +} + /// 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))] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum LogSource { + /// The platform SDK's own code (Swift, Kotlin, Dart, JS). Sdk, + /// The Rust core, through the FFI. Ffi, + /// libwebrtc. WebRtc, } @@ -143,20 +159,28 @@ impl LogSource { #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] #[derive(Debug, Clone, PartialEq)] pub struct LogRecord { + /// How serious the line is; the core applies the source's floor to it. pub severity: Severity, + /// Who wrote the line: it picks the floor, and is exported as `lk.log.source`. pub source: LogSource, /// The line itself (the OTLP log body). Not `message`: AGENTS.md keeps that name off every /// exported record. pub body: String, - /// The logger: a type, module or file name (`Room`, `livekit::rtc_engine`, `sctp.cc`). + /// The logger: a type, module or file name (`Room`, `livekit::rtc_engine`, `sctp.cc`), + /// exported as `lk.log.logger` to group lines in the backend. #[cfg_attr(feature = "uniffi", uniffi(default))] pub logger: Option, + /// The function that logged, exported as `code.function.name`. #[cfg_attr(feature = "uniffi", uniffi(default))] pub function: Option, + /// The source file, exported as `code.file.path`. #[cfg_attr(feature = "uniffi", uniffi(default))] pub file: Option, + /// The line in `file`, exported as `code.line.number` (0 is no line). #[cfg_attr(feature = "uniffi", uniffi(default))] pub line: Option, + /// When the line was logged, in nanoseconds since the Unix epoch; `None` is stamped when the + /// record is queued. #[cfg_attr(feature = "uniffi", uniffi(default))] pub timestamp_ns: Option, /// The in-flight span this line was logged under, if any. @@ -180,7 +204,7 @@ impl From for TelemetryEvent { event = event.with_attribute("code.file.path", file); } if let Some(line) = record.line.filter(|l| *l > 0) { - event = event.with_attribute("code.line.number", line as i64); + event = event.with_attribute("code.line.number", i64::from(line)); } event.timestamp_ns = record.timestamp_ns; event.span_id = record.span_id; @@ -192,7 +216,9 @@ impl From for TelemetryEvent { #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] #[derive(Debug, Clone, PartialEq)] pub struct Attribute { + /// The attribute's name (`lk.room.sid`, `acme.checkout.step`). pub key: String, + /// Its value. pub value: AttributeValue, } @@ -207,9 +233,13 @@ impl Attribute { #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, PartialEq)] pub enum AttributeValue { + /// A string (OTLP `string_value`). Str(String), + /// A 64-bit integer (OTLP `int_value`). Int(i64), + /// A 64-bit float (OTLP `double_value`). Double(f64), + /// A boolean (OTLP `bool_value`). Bool(bool), } @@ -245,8 +275,10 @@ impl From for AttributeValue { /// Current wall-clock time in nanoseconds since the Unix epoch (0 if the clock is before 1970). pub(crate) fn now_unix_nanos() -> u64 { - let now = - SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_nanos() as u64).unwrap_or(0); + let now = SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |d| { + // Past the year 2554 a u64 of nanoseconds runs out: saturate rather than wrap. + u64::try_from(d.as_nanos()).unwrap_or(u64::MAX) + }); #[cfg(test)] let now = now.saturating_add_signed(CLOCK_JUMP_NS.with(|jump| jump.get())); now @@ -254,7 +286,9 @@ pub(crate) fn now_unix_nanos() -> u64 { #[cfg(test)] thread_local! { - /// How far a test moved the wall clock, on this (current-thread runtime) thread. + /// How far a test moved the wall clock, on this (current-thread runtime) thread: the + /// capture-time test (`queued_records_carry_their_capture_time_not_their_export_time`) moves + /// it between capture and export, which no real clock does on demand. pub(crate) static CLOCK_JUMP_NS: std::cell::Cell = const { std::cell::Cell::new(0) }; } @@ -263,7 +297,9 @@ thread_local! { /// truncated id silently collides with another. #[expect(dead_code, reason = "enforced by the pipeline, not in place yet")] pub(crate) const MAX_NAME_BYTES: usize = 128; +/// The longest attribute key an app may hand over, in bytes. pub(crate) const MAX_KEY_BYTES: usize = 128; +/// The longest string attribute value an app may hand over, in bytes. pub(crate) const MAX_VALUE_BYTES: usize = 1024; /// Custom attributes per room, and per custom event. pub(crate) const MAX_CUSTOM_ATTRIBUTES: usize = 64; @@ -271,7 +307,7 @@ pub(crate) const MAX_CUSTOM_ATTRIBUTES: usize = 64; /// Keys the SDK owns: an app can neither set nor override them — `lk.*` (room, participant, /// track, outcome), `otel.*` (OTel's own, `otel.event.name` among them), `code.*` (where a log /// line was written), `session.id` and `error.type`. -pub(crate) fn reserved(key: &str) -> bool { +pub(crate) fn is_reserved(key: &str) -> bool { key.starts_with("lk.") || key.starts_with("otel.") || key.starts_with("code.") @@ -285,7 +321,7 @@ pub(crate) fn valid_custom(key: &str, value: Option<&AttributeValue>) -> bool { Some(AttributeValue::Str(s)) => s.len() <= MAX_VALUE_BYTES, _ => true, }; - !key.is_empty() && key.len() <= MAX_KEY_BYTES && !reserved(key) && value_ok + !key.is_empty() && key.len() <= MAX_KEY_BYTES && !is_reserved(key) && value_ok } /// Encoded bytes of a log record beyond its strings: timestamps, severity, trace id and the @@ -294,6 +330,9 @@ pub(crate) fn valid_custom(key: &str, value: Option<&AttributeValue>) -> bool { const RECORD_OVERHEAD_BYTES: usize = 78; /// Encoded bytes of an attribute beyond its key and string value: tags, lengths and wrappers. const ATTRIBUTE_OVERHEAD_BYTES: usize = 8; +/// Encoded bytes of a number or boolean value: a varint takes 1 to 10 and a double 8, so 8 is +/// the round figure the encoder test calibrates. +const NUMBER_VALUE_BYTES: usize = 8; /// Rough encoded size of `attributes`. pub(crate) fn attributes_size_hint(attributes: &[Attribute]) -> usize { @@ -305,7 +344,7 @@ impl AttributeValue { fn size_hint(&self) -> usize { match self { AttributeValue::Str(s) => s.len(), - _ => 8, + _ => NUMBER_VALUE_BYTES, } } } diff --git a/livekit-telemetry/src/lib.rs b/livekit-telemetry/src/lib.rs index 9afce456d..56c68873b 100644 --- a/livekit-telemetry/src/lib.rs +++ b/livekit-telemetry/src/lib.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +#![doc = include_str!("../README.md")] + /// Event data model: what SDKs push in. mod event; @@ -29,12 +31,13 @@ mod store; )] mod stats; -/// Spans: one attempt at an operation, with explicit handles across the FFI. +/// Session identity: the trace id, attributes and route of one Room. #[expect( dead_code, reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet" )] mod scope; +/// Spans: one attempt at an operation, with explicit handles across the FFI. #[expect( dead_code, reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet" @@ -72,7 +75,9 @@ pub use destination::ENDPOINT_OVERRIDE_ENV; pub use event::*; pub use span::SpanOutcome; pub use stats::{TelemetryStats, TelemetryStatus}; -pub use transport::*; +#[cfg(feature = "net")] +pub use transport::NetTransport; +pub use transport::{ExportError, ExportRequest, ExportResponse, TelemetryTransport}; #[cfg(feature = "uniffi")] uniffi::setup_scaffolding!(); @@ -80,11 +85,17 @@ uniffi::setup_scaffolding!(); /// Warnings logged by the tests, so a test can assert the ones it caused. #[cfg(test)] pub(crate) mod test_log { + use std::cell::RefCell; use std::sync::{Mutex, Once}; use std::thread::{self, ThreadId}; static LINES: Mutex> = Mutex::new(Vec::new()); + thread_local! { + /// Asked at every line this thread logs; a `true` marks the line `held: `. + static HELD: RefCell bool>>> = const { RefCell::new(None) }; + } + struct Capture; impl log::Log for Capture { @@ -94,7 +105,9 @@ pub(crate) mod test_log { fn log(&self, record: &log::Record) { if self.enabled(record.metadata()) { - let line = (thread::current().id(), record.args().to_string()); + let held = HELD.with(|held| held.borrow().as_ref().is_some_and(|held| held())); + let line = format!("{}{}", if held { "held: " } else { "" }, record.args()); + let line = (thread::current().id(), line); LINES.lock().unwrap_or_else(|e| e.into_inner()).push(line); } } @@ -111,6 +124,12 @@ pub(crate) mod test_log { }); } + /// Mark every line this thread logs while `held` returns `true` (a lock taken, say) with a + /// `held: ` prefix: the host logger would have run at that moment. + pub fn mark_while(held: impl Fn() -> bool + 'static) { + HELD.with(|slot| *slot.borrow_mut() = Some(Box::new(held))); + } + /// The warnings logged on this thread since the last call: this test's own, since every /// test runs on a thread of its own. pub fn take() -> Vec { diff --git a/livekit-telemetry/src/otlp.rs b/livekit-telemetry/src/otlp.rs index e34e486f3..02539d1ec 100644 --- a/livekit-telemetry/src/otlp.rs +++ b/livekit-telemetry/src/otlp.rs @@ -29,6 +29,7 @@ use crate::{ Attribute, AttributeValue, Severity, SpanOutcome, }; +/// The `Content-Type` of an OTLP/HTTP protobuf request. pub(crate) const CONTENT_TYPE: &str = "application/x-protobuf"; fn resource(attributes: &[Attribute]) -> Option { @@ -133,14 +134,20 @@ fn log_record( // duplicates `EventName` for backends that do not read the field yet (Loki); an app cannot // set it (`otel.*` is reserved). let body = event.body.or_else(|| (!event.name.is_empty()).then(|| event.name.clone())); + // The core's copy is the only one: a pipeline attribute of that name never ships. + event.attributes.retain(|a| a.key != "otel.event.name"); if !event.name.is_empty() { event.attributes.push(Attribute::new("otel.event.name", event.name.clone())); } + last_value_wins(&mut event.attributes); LogRecord { time_unix_nano, + // When the collecting side saw the record: for one recorded in process, OTel SDKs set + // it equal to the timestamp (9 bytes). observed_time_unix_nano: time_unix_nano, severity_number: SeverityNumber::from(event.severity) as i32, - severity_text: severity_text(event.severity).to_owned(), + // The text as well as the number, as OTel SDKs send them: a log view shows the text. + severity_text: event.severity.as_str().to_owned(), body: body.map(|text| AnyValue { value: Some(any_value::Value::StringValue(text)) }), attributes: event.attributes.iter().map(KeyValue::from).collect(), event_name: event.name, @@ -151,14 +158,13 @@ fn log_record( } fn otlp_span(mut record: SpanRecord, global: &[Attribute]) -> Span { - let session = record.session.clone(); fill_in_global(&mut record.attributes, global); // The outcome is the core's, as `session.id` is: a span's own copies never ship. record.attributes.retain(|a| a.key != "lk.outcome" && a.key != "error.type"); let mut attributes: Vec = record.attributes.iter().map(KeyValue::from).collect(); attributes.extend(record.outcome_attributes().iter().map(KeyValue::from)); Span { - trace_id: session.trace_id.to_vec(), + trace_id: record.session.trace_id.to_vec(), span_id: record.span_id.to_be_bytes().to_vec(), parent_span_id: record.parent_span_id.map(|p| p.to_be_bytes().to_vec()).unwrap_or_default(), name: record.name, @@ -196,6 +202,21 @@ fn fill_in_global(own: &mut Vec, global: &[Attribute]) { } } +/// One attribute per key, as OTLP requires (a receiver's choice among repeats is unpredictable): +/// the last value supplied wins, in the place of that last copy. Only a caller's own list can +/// repeat a key, so the precedence above holds: the session's attributes, `session.id` and +/// `otel.event.name` replaced every copy of their keys, and the rest only fill in missing keys. +fn last_value_wins(attributes: &mut Vec) { + let mut i = 0; + while i < attributes.len() { + if attributes[i + 1..].iter().any(|a| a.key == attributes[i].key) { + attributes.remove(i); + } else { + i += 1; + } + } +} + impl From for span::SpanKind { /// The OTLP value of a span kind. fn from(kind: SpanKind) -> Self { @@ -218,16 +239,6 @@ impl From for SeverityNumber { } } -fn severity_text(severity: Severity) -> &'static str { - match severity { - Severity::Trace => "TRACE", - Severity::Debug => "DEBUG", - Severity::Info => "INFO", - Severity::Warn => "WARN", - Severity::Error => "ERROR", - } -} - impl From<&Attribute> for KeyValue { fn from(attribute: &Attribute) -> Self { let value = match &attribute.value { @@ -249,6 +260,59 @@ mod tests { use super::*; use crate::TelemetryEvent; + /// A pipeline attribute named `otel.event.name` never ships a second copy of the key: the + /// core's own, from the event's name, is the only one. + #[test] + fn otel_event_name_is_sent_once() { + let session = crate::scope::ScopeState::new(); + let global = [Attribute::new("otel.event.name", "sdk-meta")]; + let queued = vec![QueuedEvent::new(TelemetryEvent::new("lk.ping"), session)]; + let decoded = ExportLogsServiceRequest::decode(&encode_logs(&[], &global, queued)[..]) + .expect("valid OTLP"); + let record = &decoded.resource_logs[0].scope_logs[0].log_records[0]; + let names: Vec<_> = + record.attributes.iter().filter(|kv| kv.key == "otel.event.name").collect(); + assert_eq!(names.len(), 1, "{names:?}"); + let value = names[0].value.as_ref().and_then(|v| v.value.as_ref()); + assert_eq!(value, Some(&any_value::Value::StringValue("lk.ping".into()))); + } + + /// OTLP keys are unique: of a caller's repeated keys the last value ships, and a session key + /// the caller repeats still ships the session's value. + #[test] + fn repeated_caller_keys_ship_once_with_the_last_value() { + let session = crate::scope::ScopeState::with_trace_id([7u8; 16]); + let custom = TelemetryEvent::custom( + "checkout", + vec![Attribute::new("app.step", "first"), Attribute::new("app.step", "second")], + ); + let sdk = TelemetryEvent::new("lk.ping") + .with_attribute("session.id", "spoofed") + .with_attribute("session.id", "spoofed again"); + let queued = + vec![QueuedEvent::new(custom, session.clone()), QueuedEvent::new(sdk, session)]; + let decoded = ExportLogsServiceRequest::decode(&encode_logs(&[], &[], queued)[..]) + .expect("valid OTLP"); + let records = &decoded.resource_logs[0].scope_logs[0].log_records; + let values = |record: &LogRecord, key: &str| -> Vec<_> { + let values = record.attributes.iter().filter(|kv| kv.key == key); + values.filter_map(|kv| kv.value.clone()?.value).collect() + }; + assert_eq!( + values(&records[0], "app.step"), + [any_value::Value::StringValue("second".into())] + ); + let session_id = any_value::Value::StringValue("07".repeat(16)); + assert_eq!(values(&records[1], "session.id"), [session_id]); + for record in records { + let mut keys: Vec<_> = record.attributes.iter().map(|kv| &kv.key).collect(); + let count = keys.len(); + keys.sort_unstable(); + keys.dedup(); + assert_eq!(keys.len(), count, "unique keys: {:?}", record.attributes); + } + } + #[test] fn encodes_events_as_otlp_log_records() { let resource = [Attribute::new("service.name", "test")]; diff --git a/livekit-telemetry/src/span.rs b/livekit-telemetry/src/span.rs index 5987d7d60..53dee7655 100644 --- a/livekit-telemetry/src/span.rs +++ b/livekit-telemetry/src/span.rs @@ -23,7 +23,7 @@ use crate::{ /// OTel span kind, restricted to what client operations need; implied by [`crate::SpanName`]. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SpanKind { +pub(crate) enum SpanKind { /// An operation inside the SDK (publish, subscribe). Internal, /// A call to the SFU that waits for its answer (connect, reconnect). @@ -36,8 +36,11 @@ pub enum SpanKind { #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum SpanOutcome { + /// It did what it set out to do (OTel status `Ok`). Ok, + /// It failed; `error.type` says how (OTel status `Error`). Error, + /// It was called off before it finished, which is not a failure (OTel status `Unset`). Cancelled, } @@ -89,8 +92,9 @@ const SPAN_OVERHEAD_BYTES: usize = 94; /// Encoded bytes of a span event beyond its name: its timestamp and framing. const SPAN_EVENT_OVERHEAD_BYTES: usize = 13; -/// OTel default span limits. +/// Checkpoints one span keeps: OTel's default span limit. const MAX_EVENTS_PER_SPAN: usize = 128; +/// Attributes one span keeps: OTel's default span limit. const MAX_ATTRIBUTES_PER_SPAN: usize = 128; /// Spans a host can leave open before the oldest is abandoned (counted as dropped). const MAX_OPEN_SPANS: usize = 256; @@ -104,7 +108,7 @@ pub(crate) struct Spans { open: HashMap, /// Insertion order of `open`, to abandon the oldest when the cap is hit. open_order: VecDeque, - finished: Vec, + finished: VecDeque, finished_capacity: usize, /// Which session every recent span belongs to — open, finished or already exported — so a /// log record that arrives after its span ended (a warning logged right before a failing @@ -114,7 +118,7 @@ pub(crate) struct Spans { /// open span keeps its session however many spans start after it. session_order: VecDeque, next_id: u64, - pub dropped: u64, + dropped: u64, warned: Warned, /// The opt-out, checked under the lock that guards the registry (see `Store`). revoked: Arc, @@ -150,13 +154,13 @@ impl Spans { Self { open: HashMap::new(), open_order: VecDeque::new(), - finished: Vec::new(), + finished: VecDeque::new(), finished_capacity: finished_capacity.max(1), sessions: HashMap::new(), session_order: VecDeque::new(), - // Span ids must be non-zero (OTLP treats all-zero as absent); start at 1 and mix in - // randomness so ids from two pipelines in one process never collide. 63 bits: a - // platform whose integers are signed (Dart) must be able to hand an id back. + // A random odd 63-bit start: never zero (OTLP treats all-zero as absent), and two + // pipelines in one process are very unlikely to overlap. 63 bits: a platform whose + // integers are signed (Dart) must be able to hand an id back. next_id: (rand::random::() >> 1) | 1, dropped: 0, warned: Warned::default(), @@ -174,7 +178,8 @@ impl Spans { self.revoked.load(std::sync::atomic::Ordering::SeqCst) } - /// Open a span in `session`'s trace. + /// Open a span in `session`'s trace. After the opt-out nothing opens: returns 0, which is + /// never a span id. pub fn begin_in( &mut self, name: &str, @@ -216,8 +221,6 @@ impl Spans { id } - /// Whether a span with one of these names is still open (the exporter holds uploads while - /// `lk.connect` / `lk.reconnect` are). /// The session a recent span belongs to — open, ended or exported (log records emitted inside /// a span are filed there, and they may arrive after the span ended). pub fn scope_of(&self, id: u64) -> Option> { @@ -234,6 +237,8 @@ impl Spans { self.open.len() } + /// Whether a span with one of these names is still open (the exporter holds uploads while + /// `lk.connect` / `lk.reconnect` are). pub fn any_open(&self, names: &[&str]) -> bool { self.open.values().any(|span| names.contains(&span.name.as_str())) } @@ -302,10 +307,10 @@ impl Spans { span.attributes = attributes; self.retire(id); if self.finished.len() >= self.finished_capacity { - self.finished.remove(0); + self.finished.pop_front(); self.count_drop("finished", self.finished_capacity); } - self.finished.push(span); + self.finished.push_back(span); } /// A span ended or was abandoned: its session stays resolvable until [`REMEMBERED_SPANS`] @@ -329,8 +334,8 @@ impl Spans { } /// Take the finished spans, oldest first. - /// At most `max` spans and about `max_bytes` (always at least one, so an oversized span - /// still ships). + /// At most `max` spans and about `max_bytes` (at least one when `max > 0`, so an oversized + /// span still ships). pub fn drain(&mut self, max: usize, max_bytes: usize) -> Vec { self.warned = Warned::default(); let (mut n, mut bytes) = (0, 0); diff --git a/livekit-telemetry/src/stats.rs b/livekit-telemetry/src/stats.rs index 26a5d759e..f2a10ea4e 100644 --- a/livekit-telemetry/src/stats.rs +++ b/livekit-telemetry/src/stats.rs @@ -50,7 +50,8 @@ macro_rules! counters { } counters! { - /// Records evicted from the in-memory queue (`max_queue_size`). + /// Records evicted from the in-memory queue (`max_queue_size`), and spans dropped from a full + /// span buffer (open spans abandoned, finished ones past its capacity). queue_full, /// Records lost because no cache, not even memory, could take their batch. cache_error, @@ -101,18 +102,13 @@ impl Counters { impl Snapshot { /// Everything that was lost, by any reason. pub fn dropped(&self) -> u64 { - self.queue_full - + self.cache_error - + self.cache_full - + self.expired - + self.corrupt - + self.invalid - + self.rejected - + self.oversized - + self.throttled - + self.disabled - + self.rate_limited - + self.purged + self.problem_drops() + self.disabled + self.purged + } + + /// Upload attempts that did not get through: failures and timeouts together, the "failed" + /// of the report's body (the `lk.telemetry.uploads.failed` attribute counts failures only). + pub fn failed_attempts(&self) -> u64 { + self.upload_failures + self.upload_timeouts } /// Records lost other than by policy (a project that receives nothing, an opt-out). @@ -150,14 +146,14 @@ impl Snapshot { "telemetry: {} batches sent ({} B), {} failed, {} dropped, {} cached", self.uploads_sent, self.upload_bytes, - self.upload_failures + self.upload_timeouts, + self.failed_attempts(), self.problem_drops(), cached_batches )) - .with_attribute("lk.telemetry.uploads.sent", self.uploads_sent as i64) - .with_attribute("lk.telemetry.uploads.bytes", self.upload_bytes as i64) - .with_attribute("lk.telemetry.uploads.failed", self.upload_failures as i64) - .with_attribute("lk.telemetry.cache.batches", cached_batches as i64); + .with_attribute("lk.telemetry.uploads.sent", int(self.uploads_sent)) + .with_attribute("lk.telemetry.uploads.bytes", int(self.upload_bytes)) + .with_attribute("lk.telemetry.uploads.failed", int(self.upload_failures)) + .with_attribute("lk.telemetry.cache.batches", int(cached_batches)); for (key, value) in [ ("lk.telemetry.uploads.timeouts", self.upload_timeouts), ("lk.telemetry.uploads.unauthorized", self.auth_denied), @@ -175,13 +171,18 @@ impl Snapshot { ("lk.telemetry.dropped.rate_limited", self.rate_limited), ] { if value > 0 { - event = event.with_attribute(key, value as i64); + event = event.with_attribute(key, int(value)); } } event } } +/// A count as an OTLP integer: past `i64::MAX` it saturates rather than wraps negative. +fn int(count: u64) -> i64 { + i64::try_from(count).unwrap_or(i64::MAX) +} + /// What the upload policy is doing right now. #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -224,17 +225,29 @@ pub struct TelemetryStats { pub status: TelemetryStatus, /// Records lost for any reason (sum of the `dropped_*` fields). pub dropped: u64, + /// Records evicted from the in-memory queue, and spans dropped from a full span buffer. pub dropped_queue_full: u64, + /// Records lost because no cache, not even memory, could take their batch. pub dropped_cache_error: u64, + /// Records evicted from the cache by its size or file-count bound. pub dropped_cache_full: u64, + /// Records in cached batches past the 24 h age limit. pub dropped_expired: u64, + /// Records in cached batches that failed their integrity check. pub dropped_corrupt: u64, + /// Custom events and attributes refused at the door, over the limits. pub dropped_invalid: u64, + /// Records the collector rejected (a final 4xx/5xx, or a partial success). pub dropped_rejected: u64, + /// Single records larger than the collector accepts. pub dropped_oversized: u64, + /// Records evicted from the cache while the collector held uploads off. pub dropped_throttled: u64, + /// Records for a project that receives nothing (self-hosted, no grant, disabled, 404). pub dropped_disabled: u64, + /// Discrete events dropped by the flood guard. pub dropped_rate_limited: u64, + /// Records deleted by the opt-out. pub dropped_purged: u64, /// Batches the collector accepted. pub uploads_sent: u64, @@ -254,8 +267,9 @@ pub struct TelemetryStats { pub cached_batches: u64, } -/// One line for a debug console: status, throughput, one backlog number, one loss number, then -/// the loss breakdown. +/// One line for a debug console: status, throughput, one backlog number, "failed" (failures and +/// timeouts together, [`TelemetryStats::failed_attempts`]), one loss number, then the loss +/// breakdown. impl std::fmt::Display for TelemetryStats { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!( @@ -267,7 +281,7 @@ impl std::fmt::Display for TelemetryStats { self.uploads_sent, self.upload_bytes, self.cached_batches, - self.upload_failures + self.upload_timeouts, + self.failed_attempts(), self.dropped, self.dropped_queue_full, self.dropped_cache_full + self.dropped_cache_error, @@ -288,6 +302,12 @@ impl std::fmt::Display for TelemetryStats { } impl TelemetryStats { + /// Upload attempts that did not get through: `upload_failures` and `upload_timeouts` + /// together, the "failed" of the `Display` line and of the report's body. + pub fn failed_attempts(&self) -> u64 { + self.upload_failures + self.upload_timeouts + } + pub(crate) fn new(snapshot: Snapshot, cached_batches: u64, status: TelemetryStatus) -> Self { Self { status, @@ -315,3 +335,20 @@ impl TelemetryStats { } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::AttributeValue; + + /// A count past `i64::MAX` reaches the report as `i64::MAX`, never as a negative number. + #[test] + fn report_counts_saturate_instead_of_wrapping() { + let snapshot = Snapshot { upload_failures: u64::MAX, ..Snapshot::default() }; + let report = snapshot.report(u64::MAX); + let value = + |key: &str| report.attributes.iter().find(|a| a.key == key).map(|a| a.value.clone()); + assert_eq!(value("lk.telemetry.uploads.failed"), Some(AttributeValue::Int(i64::MAX))); + assert_eq!(value("lk.telemetry.cache.batches"), Some(AttributeValue::Int(i64::MAX))); + } +} diff --git a/livekit-telemetry/src/store.rs b/livekit-telemetry/src/store.rs index 075c9d4a8..44120ea39 100644 --- a/livekit-telemetry/src/store.rs +++ b/livekit-telemetry/src/store.rs @@ -59,7 +59,8 @@ pub(crate) struct Store { queue: Mutex, /// The opt-out, checked under the queue lock `clear` also takes: nothing lands after a purge. revoked: Arc, - /// Test-only: runs at the start of `push`, before the lock (to hold a producer there). + /// Test-only: runs at the start of `push`, before the lock, so the opt-out race test + /// (`a_capture_racing_the_opt_out_leaves_nothing_behind`) can hold a producer exactly there. #[cfg(test)] pub(crate) pause: Mutex>>, capacity: usize, @@ -110,23 +111,27 @@ impl Store { if self.revoked.load(Ordering::SeqCst) { return false; } + let mut warn = false; if queue.events.len() >= self.capacity { if let Some(oldest) = queue.events.pop_front() { queue.bytes = queue.bytes.saturating_sub(oldest.event.size_hint()); } Counters::add(&self.counters.queue_full, 1); - if !queue.full_warned { - queue.full_warned = true; - log::warn!( - "queue full ({} records): dropping oldest until the exporter drains", - self.capacity - ); - } + warn = !std::mem::replace(&mut queue.full_warned, true); } let before = queue.bytes; queue.bytes += queued.event.size_hint(); queue.events.push_back(queued); - before < self.flush_threshold && queue.bytes >= self.flush_threshold + let crossed = before < self.flush_threshold && queue.bytes >= self.flush_threshold; + // Logged after the lock: a log line may call into the host. + drop(queue); + if warn { + log::warn!( + "queue full ({} records): dropping oldest until the exporter drains", + self.capacity + ); + } + crossed } /// Remove and return the oldest events: at most `min(max, len)` of them and about `max_bytes` @@ -172,6 +177,23 @@ mod tests { QueuedEvent::new(event, ScopeState::new()) } + /// The queue-full warning reaches the logger once the queue's lock is released: a host logger + /// that calls back into the pipeline cannot block on it. + #[test] + fn the_queue_full_warning_is_logged_outside_the_queue_lock() { + crate::test_log::capture(); + let store = Arc::new(Store::new(1, usize::MAX, Arc::new(Counters::default()))); + let held = Arc::downgrade(&store); + crate::test_log::mark_while(move || { + held.upgrade().is_some_and(|store| store.queue.try_lock().is_err()) + }); + store.push(queued(TelemetryEvent::new("a"))); + store.push(queued(TelemetryEvent::new("b"))); + let lines = crate::test_log::take(); + assert!(lines.iter().any(|line| line.contains("queue full")), "{lines:?}"); + assert!(lines.iter().all(|line| !line.starts_with("held: ")), "{lines:?}"); + } + #[test] fn drops_oldest_when_full() { crate::test_log::capture(); diff --git a/livekit-telemetry/src/transport.rs b/livekit-telemetry/src/transport.rs index 1d29281a5..fdc22d273 100644 --- a/livekit-telemetry/src/transport.rs +++ b/livekit-telemetry/src/transport.rs @@ -19,24 +19,30 @@ use prost::Message; /// One fully composed OTLP/HTTP export request. /// /// The core fills in the URL, the headers (content type, auth, …) and the protobuf body; -/// a transport only moves the bytes. Non-HTTP transports (e.g. a data channel) may ignore -/// `url`/`headers` and forward `body`, which is a standard `ExportLogsServiceRequest`. +/// a transport only moves the bytes. The body is an `ExportLogsServiceRequest` or an +/// `ExportTraceServiceRequest`, as `url` says. #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] #[derive(Debug, Clone, PartialEq)] pub struct ExportRequest { + /// Where to POST: the OTLP logs or traces endpoint of the batch's project. pub url: String, + /// Headers to send exactly as given: content type, content encoding, `Authorization`. pub headers: HashMap, + /// The encoded request body, sent as is. pub body: Vec, } /// What the collector answered. A transport moves bytes both ways and never interprets them: /// status classification, `Retry-After`, the `google.rpc.Status` body and the Cloud "disabled" -/// contract are read once, here, for every platform ([`ExportError::from_response`]). +/// contract are read once, here, for every platform (`Verdict::of`). #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] #[derive(Debug, Clone, PartialEq, Default)] pub struct ExportResponse { + /// The HTTP status code. pub status: u16, + /// The response headers; the core reads `Retry-After` from them. pub headers: HashMap, + /// The response body: an OTLP partial success, a `google.rpc.Status` or plain text. pub body: Vec, } @@ -51,6 +57,10 @@ impl ExportResponse { /// text, no machine-readable code yet). Matched as a phrase, not a word. const DISABLED_ANSWER: &str = "data recording is disabled by owner"; +/// The most characters of an error body that a verdict's reason keeps: enough to read the cause in +/// a log line, short enough that a large HTML error page does not fill it. +const MAX_REASON_CHARS: usize = 200; + /// How long a 429 that names no delay holds uploads: LiveKit Cloud's quota answer names none. const THROTTLE_DEFAULT_MS: u64 = 60_000; @@ -60,15 +70,20 @@ const THROTTLE_DEFAULT_MS: u64 = 60_000; #[derive(Debug, Clone, PartialEq, thiserror::Error)] pub enum ExportError { /// No response: connection refused or reset, DNS or TLS failure, offline, timed out. The - /// batch stays cached and is retried with backoff. `retry_after_ms` is honored when set. + /// batch stays cached and is retried with backoff. #[error("retryable export error: {reason}")] - Retryable { reason: String, retry_after_ms: Option }, + Retryable { + /// What went wrong, for the log. + reason: String, + /// How long to wait before the next try, when known. `None` uses the backoff. + retry_after_ms: Option, + }, /// The request can never succeed (an invalid URL): the batch is dropped. #[error("export rejected: {reason}")] - Rejected { reason: String }, - /// Telemetry is off for this project: the project goes silent. - #[error("telemetry disabled by the collector")] - Disabled, + Rejected { + /// What went wrong, for the log. + reason: String, + }, } /// A foreign transport threw something its binding did not declare (a Swift `Error`, a Kotlin @@ -89,8 +104,8 @@ impl From for ExportError { pub(crate) enum Verdict { /// 2xx. `rejected` counts the records an OTLP partial success refused (never retried). Accepted { rejected: u64, reason: String }, - /// 400, 3xx (a credential-bearing request is never redirected), and every other 4xx/5xx the - /// OTLP spec does not call retryable: drop the batch. + /// 400, 3xx (a redirect the transport did not follow), and every other 4xx/5xx the OTLP spec + /// does not call retryable: drop the batch. Rejected(String), /// 413: split the batch and retry the halves, down to single records. TooLarge, @@ -104,8 +119,8 @@ pub(crate) enum Verdict { Disabled, /// 429, or 503 naming a delay: pause every upload for `delay_ms`, keep collecting. Throttled { delay_ms: u64, reason: String }, - /// 502/503/504, or any error carrying `RetryInfo`: the server failed transiently. Retry the - /// batch after `delay_ms` or a backoff; a batch that keeps failing is dropped eventually. + /// 502/503/504, or 500 carrying `RetryInfo`: the server failed transiently. Retry the batch + /// after `delay_ms` or a backoff; a batch that keeps failing is dropped eventually. Retry { delay_ms: Option, reason: String }, } @@ -125,8 +140,13 @@ impl Verdict { let rpc = RpcStatus::decode(response.body.as_slice()).ok(); let text = match rpc.as_ref().map(|s| s.message.trim()).filter(|m| !m.is_empty()) { Some(message) => message.to_owned(), - None => String::from_utf8_lossy(&response.body).trim().chars().take(200).collect(), + None => String::from_utf8_lossy(&response.body).trim().to_owned(), }; + // The whole answer is searched for the phrase; only the logged reason is cut. + if matches!(status, 401 | 403) && text.to_ascii_lowercase().contains(DISABLED_ANSWER) { + return Self::Disabled; + } + let text: String = text.chars().take(MAX_REASON_CHARS).collect(); // `reason`, not `message`: a UniFFI field named `message` collides with // `Throwable.message` in Kotlin. let reason = if text.is_empty() { @@ -135,11 +155,7 @@ impl Verdict { format!("HTTP {status}: {text}") }; if matches!(status, 401 | 403) { - return if text.to_ascii_lowercase().contains(DISABLED_ANSWER) { - Self::Disabled - } else { - Self::Unauthorized(reason) - }; + return Self::Unauthorized(reason); } let header = response .headers @@ -231,12 +247,19 @@ fn imf_fixdate_secs(value: &str) -> Option { ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]; let month = MONTHS.iter().position(|m| *m == month)? as i64 + 1; let (day, year): (i64, i64) = (day.parse().ok()?, year.parse().ok()?); + let leap = year % 4 == 0 && (year % 100 != 0 || year % 400 == 0); + let month_days = match month { + 2 if leap => 29, + 2 => 28, + 4 | 6 | 9 | 11 => 30, + _ => 31, + }; let mut hms = time.split(':').map(|n| n.parse::().ok()); let (h, m, s) = (hms.next()??, hms.next()??, hms.next()??); // IMF-fixdate has a four-digit year; bounding it keeps the arithmetic below in range. if !(1970..=9999).contains(&year) || hms.next().is_some() - || !(1..=31).contains(&day) + || !(1..=month_days).contains(&day) || !(0..24).contains(&h) || !(0..60).contains(&m) || !(0..61).contains(&s) @@ -268,7 +291,7 @@ fn retry_info_ms(status: &RpcStatus) -> Option { /// Moves an encoded batch off the device and hands back whatever came back. /// -/// Implemented in Rust ([`NetTransport`], feature `net`) or by the host platform through UniFFI. +/// Implemented in Rust (`NetTransport`, feature `net`) or by the host platform through UniFFI. /// The contract: /// /// - POST `url` with `headers` and `body` as given, and return the response whatever its status; @@ -278,11 +301,13 @@ fn retry_info_ms(status: &RpcStatus) -> Option { /// - Never retry: the core owns the retry policy. /// - Never forward the `Authorization` header across origins: do not follow a redirect to /// another scheme, host or port with it (returning the 3xx is fine — the core drops the batch). -/// `NetTransport` over `livekit-net`'s native client strips it on cross-origin redirects. +/// `NetTransport` over the native `livekit-net` client follows at most 10 redirects, never one +/// that changes the scheme, and drops `Authorization` when a redirect changes the host or port. /// - Keep the request off the call's critical path: lowest priority where the platform allows it. #[cfg_attr(feature = "uniffi", uniffi::export(with_foreign))] #[async_trait::async_trait] pub trait TelemetryTransport: Send + Sync { + /// Send `request` once and return the collector's answer, whatever its status. async fn send(&self, request: ExportRequest) -> Result; } @@ -290,7 +315,7 @@ pub trait TelemetryTransport: Send + Sync { mod net { use std::sync::Arc; - use livekit_net::{Header, HttpClient, HttpClientExt, TransportError}; + use livekit_net::{Header, HttpClient, HttpClientExt, HttpResponse, TransportError}; use super::{ExportError, ExportRequest, ExportResponse, TelemetryTransport}; @@ -300,13 +325,23 @@ mod net { impl NetTransport { /// Post through `client`. - pub fn new(client: Arc) -> Self { + pub(crate) fn new(client: Arc) -> Self { Self(client) } /// Resolve the process-wide `livekit-net` client; `None` when none is available. pub fn from_registry() -> Option { - livekit_net::http_client().map(Self) + livekit_net::http_client().map(Self::new) + } + } + + impl From for ExportResponse { + fn from(response: HttpResponse) -> Self { + Self { + status: response.status, + headers: response.headers.into_iter().map(|h| (h.name, h.value)).collect(), + body: response.body, + } } } @@ -316,11 +351,7 @@ mod net { let headers = request.headers.into_iter().map(|(name, value)| Header { name, value }).collect(); match self.0.post(request.url, headers, request.body).await { - Ok(response) => Ok(ExportResponse { - status: response.status, - headers: response.headers.into_iter().map(|h| (h.name, h.value)).collect(), - body: response.body, - }), + Ok(response) => Ok(response.into()), // The client swallowed the body with the status; the core still classifies it. Err(TransportError::Http { status }) => { Ok(ExportResponse { status, ..Default::default() }) @@ -409,6 +440,13 @@ mod tests { assert_eq!(verdict(401, &[], text), Verdict::Disabled); let rpc = status_body("Project data recording is disabled by owner", None); assert_eq!(verdict(403, &[], rpc), Verdict::Disabled); + // The phrase counts wherever it sits; only the logged reason is cut short. + let long = format!("{} data recording is disabled by owner", "x".repeat(250)); + assert_eq!(verdict(401, &[], long.into_bytes()), Verdict::Disabled); + let Verdict::Unauthorized(reason) = verdict(401, &[], "y".repeat(500).into_bytes()) else { + panic!("a credential answer") + }; + assert_eq!(reason.len(), "HTTP 401: ".len() + MAX_REASON_CHARS); // Only the documented phrase: another answer that merely mentions "disabled" is a // credential problem, not a project switch. for body in ["account disabled", "token disabled for this room"] { @@ -501,6 +539,17 @@ mod tests { assert!(matches!(negative, Verdict::Throttled { delay_ms: 0, .. }), "clamped"); } + #[test] + fn impossible_calendar_dates_are_ignored() { + assert_eq!(imf_fixdate_secs("Tue, 30 Feb 2027 12:00:00 GMT"), None); + assert_eq!(imf_fixdate_secs("Mon, 29 Feb 2027 12:00:00 GMT"), None, "not a leap year"); + assert_eq!(imf_fixdate_secs("Thu, 31 Apr 2027 12:00:00 GMT"), None); + assert_eq!(imf_fixdate_secs("Tue, 29 Feb 2100 12:00:00 GMT"), None, "a century"); + assert!(imf_fixdate_secs("Tue, 29 Feb 2000 12:00:00 GMT").is_some(), "every 400 years"); + let ignored = verdict(429, &[("Retry-After", "Tue, 30 Feb 2027 12:00:00 GMT")], vec![]); + assert!(matches!(ignored, Verdict::Throttled { delay_ms: 60_000, .. }), "the default"); + } + #[test] fn retry_after_http_dates() { assert_eq!(imf_fixdate_secs("Sun, 06 Nov 1994 08:49:37 GMT"), Some(784_111_777)); @@ -508,30 +557,14 @@ mod tests { assert_eq!(imf_fixdate_secs("Tue, 29 Feb 2028 12:00:00 GMT"), Some(1_835_438_400)); assert_eq!(imf_fixdate_secs("Sunday, 06-Nov-94 08:49:37 GMT"), None, "obsolete form"); assert_eq!(imf_fixdate_secs("Sun, 06 Nov 1994 08:49:37 PST"), None); + // A far-future date: the delay is the distance from now, whatever now is. let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .expect("clock") .as_secs(); - let date = |secs: u64| { - let days = secs / 86_400; - let (h, m, s) = ((secs % 86_400) / 3600, (secs % 3600) / 60, secs % 60); - // Civil from days (inverse of the parser's algorithm) for the fixture only. - let z = days as i64 + 719_468; - let era = z.div_euclid(146_097); - let doe = z - era * 146_097; - let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; - let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); - let mp = (5 * doy + 2) / 153; - let d = doy - (153 * mp + 2) / 5 + 1; - let mo = if mp < 10 { mp + 3 } else { mp - 9 }; - let y = yoe + era * 400 + if mo <= 2 { 1 } else { 0 }; - const M: [&str; 12] = [ - "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec", - ]; - format!("Mon, {d:02} {} {y} {h:02}:{m:02}:{s:02} GMT", M[mo as usize - 1]) - }; - let ms = retry_after_ms(&date(now + 120)).expect("date"); - assert!((118_000..=120_000).contains(&ms), "{ms}"); + let expected = (253_402_300_799 - now) * 1000; + let ms = retry_after_ms("Fri, 31 Dec 9999 23:59:59 GMT").expect("date"); + assert!(expected.abs_diff(ms) <= 2_000, "{ms} vs {expected}"); } #[test]