feat(telemetry): add the transport seam and LiveKit Cloud destinations

.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.

.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.

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");
+    }
+}

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]
+# 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"]
 
+[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]

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.
 

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
 
@@ -24,6 +26,61 @@ 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://<host>/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 <token>`. 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://<project>.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). 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 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
 
 An event with no `body` is exported with its name as the body as well as in `event_name`: log
@@ -46,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
@@ -170,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:
@@ -188,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 |
 |---|---|