livekit_telemetry/
lib.rs

1// Copyright 2026 LiveKit, Inc.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15#![doc = include_str!("../README.md")]
16
17/// Event data model: what SDKs push in.
18mod event;
19
20/// Bounded in-memory queue between `emit` and the exporter.
21#[expect(
22    dead_code,
23    reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet"
24)]
25mod store;
26
27/// Pipeline health counters and the `lk.telemetry.report` event.
28#[expect(
29    dead_code,
30    reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet"
31)]
32mod stats;
33
34/// Host-reported device state and the cadence policy derived from it.
35#[cfg_attr(
36    not(test),
37    expect(
38        dead_code,
39        reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet"
40    )
41)]
42mod device;
43
44/// RTC stats samples and their on-device windowing.
45#[cfg_attr(
46    not(test),
47    expect(
48        dead_code,
49        reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet"
50    )
51)]
52mod rtc;
53
54/// Session identity: the trace id, attributes and route of one Room.
55#[expect(
56    dead_code,
57    reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet"
58)]
59mod scope;
60/// Spans: one attempt at an operation, with explicit handles across the FFI.
61#[expect(
62    dead_code,
63    reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet"
64)]
65mod span;
66
67/// OTLP/HTTP protobuf encoding of a batch.
68#[expect(
69    dead_code,
70    reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet"
71)]
72mod otlp;
73
74/// Queue of encoded batches between the exporter and the transport (memory or disk).
75mod cache;
76
77/// OTLP protobuf types (re-exported from `opentelemetry-proto`).
78mod proto;
79
80/// Transport seam: how encoded batches leave the device.
81#[cfg_attr(
82    not(test),
83    expect(
84        dead_code,
85        reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet"
86    )
87)]
88mod transport;
89
90/// Where batches go: server URL + token → ingest URL, grant, expiry, per-project routing.
91#[expect(
92    dead_code,
93    reason = "consumed by the pipeline (`Telemetry`, `Exporter`), not in place yet"
94)]
95mod destination;
96
97pub use cache::{BatchCache, FileCache, MemoryCache};
98pub use destination::ENDPOINT_OVERRIDE_ENV;
99pub use device::*;
100pub use event::*;
101pub use rtc::{RtcStat, RtcStatsSample, StreamDirection, TrackKind};
102pub use span::SpanOutcome;
103pub use stats::{TelemetryStats, TelemetryStatus};
104#[cfg(feature = "net")]
105pub use transport::NetTransport;
106pub use transport::{ExportError, ExportRequest, ExportResponse, TelemetryTransport};
107
108#[cfg(feature = "uniffi")]
109uniffi::setup_scaffolding!();
110
111/// Warnings logged by the tests, so a test can assert the ones it caused.
112#[cfg(test)]
113pub(crate) mod test_log {
114    use std::cell::RefCell;
115    use std::sync::{Mutex, Once};
116    use std::thread::{self, ThreadId};
117
118    static LINES: Mutex<Vec<(ThreadId, String)>> = Mutex::new(Vec::new());
119
120    thread_local! {
121        /// Asked at every line this thread logs; a `true` marks the line `held: `.
122        static HELD: RefCell<Option<Box<dyn Fn() -> bool>>> = const { RefCell::new(None) };
123    }
124
125    struct Capture;
126
127    impl log::Log for Capture {
128        fn enabled(&self, metadata: &log::Metadata) -> bool {
129            metadata.level() <= log::Level::Warn
130        }
131
132        fn log(&self, record: &log::Record) {
133            if self.enabled(record.metadata()) {
134                let held = HELD.with(|held| held.borrow().as_ref().is_some_and(|held| held()));
135                let line = format!("{}{}", if held { "held: " } else { "" }, record.args());
136                let line = (thread::current().id(), line);
137                LINES.lock().unwrap_or_else(|e| e.into_inner()).push(line);
138            }
139        }
140
141        fn flush(&self) {}
142    }
143
144    /// Start capturing; `log` takes one logger per process, so the first test installs it.
145    pub fn capture() {
146        static INSTALL: Once = Once::new();
147        INSTALL.call_once(|| {
148            log::set_logger(&Capture).expect("the only logger in tests");
149            log::set_max_level(log::LevelFilter::Warn);
150        });
151    }
152
153    /// Mark every line this thread logs while `held` returns `true` (a lock taken, say) with a
154    /// `held: ` prefix: the host logger would have run at that moment.
155    pub fn mark_while(held: impl Fn() -> bool + 'static) {
156        HELD.with(|slot| *slot.borrow_mut() = Some(Box::new(held)));
157    }
158
159    /// The warnings logged on this thread since the last call: this test's own, since every
160    /// test runs on a thread of its own.
161    pub fn take() -> Vec<String> {
162        let me = thread::current().id();
163        let mut lines = LINES.lock().unwrap_or_else(|e| e.into_inner());
164        let (mine, others): (Vec<_>, Vec<_>) =
165            std::mem::take(&mut *lines).into_iter().partition(|(id, _)| *id == me);
166        *lines = others;
167        mine.into_iter().map(|(_, line)| line).collect()
168    }
169}
livekit_telemetry/lib.rs

livekit_telemetry/
lib.rs
