diff --git a/.changeset/protocol-semconv.md b/.changeset/protocol-semconv.md new file mode 100644 index 000000000..dc9299926 --- /dev/null +++ b/.changeset/protocol-semconv.md @@ -0,0 +1,5 @@ +--- +livekit-protocol: patch +--- + +Add `semconv`, the telemetry names and enum values generated from the protocol's Weaver registry, and move the protocol submodule to `003613b7`. diff --git a/.changeset/telemetry-semconv.md b/.changeset/telemetry-semconv.md new file mode 100644 index 000000000..437312f21 --- /dev/null +++ b/.changeset/telemetry-semconv.md @@ -0,0 +1,5 @@ +--- +livekit-telemetry: minor +--- + +Take every telemetry name from the protocol registry (`livekit_protocol::semconv`) and send its spelling: `lk.room_sid`, `lk.participant_id`, `lk.track_sid`, `lk.pii.room_name`, `lk.pii.participant_identity`, `lk.pii.remote_participant_identity`, `lk.track_source` with the RTC FFI `TrackSource` names, and `lk.disconnect_reason` with the protobuf `DisconnectReason` names. diff --git a/Cargo.lock b/Cargo.lock index 91650ba92..b09f9df16 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4022,6 +4022,7 @@ dependencies = [ "base64 0.22.1", "flate2", "livekit-net", + "livekit-protocol", "log", "opentelemetry-proto", "prost 0.14.4", diff --git a/livekit-protocol/generate_proto.sh b/livekit-protocol/generate_proto.sh index 901adedd7..586d34fc9 100755 --- a/livekit-protocol/generate_proto.sh +++ b/livekit-protocol/generate_proto.sh @@ -16,6 +16,10 @@ # dependencies: cargo install protoc-gen-prost@0.5.0 protoc-gen-prost-serde@0.4.0 +# The telemetry names (`src/lib.rs`, `mod semconv`) are not generated here: Weaver writes them in +# the protocol repository (`otel/generate.sh`) and the crate reads `protocol/otel/gen/rust/semconv.rs` +# straight from the submodule. + PROTOCOL=${PROTOCOL:-protocol/protobufs} OUT_RUST=src diff --git a/livekit-protocol/protocol b/livekit-protocol/protocol index 3fbe50985..003613b76 160000 --- a/livekit-protocol/protocol +++ b/livekit-protocol/protocol @@ -1 +1 @@ -Subproject commit 3fbe509850cb24aa90a0bc20134936698333dceb +Subproject commit 003613b76adfcd57f5142be6395b7978ffca917e diff --git a/livekit-protocol/src/lib.rs b/livekit-protocol/src/lib.rs index 0f6566c46..818a9b5d4 100644 --- a/livekit-protocol/src/lib.rs +++ b/livekit-protocol/src/lib.rs @@ -49,3 +49,10 @@ pub mod agent { pub mod logger { include!("logger.rs"); } + +/// The telemetry names and enum values, generated by Weaver in `livekit/protocol` +/// (`otel/gen/rust/semconv.rs`) and read straight from the submodule, so the +/// submodule pointer is the version. +#[path = "../protocol/otel/gen/rust/semconv.rs"] +#[rustfmt::skip] +pub mod semconv; diff --git a/livekit-telemetry/Cargo.toml b/livekit-telemetry/Cargo.toml index c74521015..66dce5b34 100644 --- a/livekit-telemetry/Cargo.toml +++ b/livekit-telemetry/Cargo.toml @@ -26,6 +26,8 @@ flate2 = "1" # OTLP message types only (`gen-tonic-messages` = prost structs, no tonic). Same prost as # livekit-protocol so a single prost is linked. opentelemetry-proto = { version = "0.32", default-features = false, features = ["logs", "trace", "gen-tonic-messages"] } +# The registered telemetry names (`semconv`), generated from the protocol's Weaver registry. +livekit-protocol = { workspace = true } livekit-net = { workspace = true, optional = true } uniffi = { workspace = true, features = ["scaffolding-ffi-buffer-fns"], optional = true } diff --git a/livekit-telemetry/SPEC.md b/livekit-telemetry/SPEC.md index 9fea4a4c3..8386cd6fe 100644 --- a/livekit-telemetry/SPEC.md +++ b/livekit-telemetry/SPEC.md @@ -1,12 +1,18 @@ # Client telemetry spec -Source of truth for event names, attributes and cadences emitted by LiveKit client SDKs. -Additive-only by convention; LiveKit-defined names carry the `lk.` prefix, everything else -follows [OpenTelemetry semantic conventions](https://github.com/open-telemetry/semantic-conventions). +Event names, attributes and cadences emitted by LiveKit client SDKs. Every fixed name and every +enum value that the protocol's telemetry registry (`otel/` in livekit/protocol) defines comes from +the generated `livekit_protocol::semconv`, and `src/wire_tests.rs` pins what the core sends. Four +values stay literals until the registry defines them: the `network.connection.type` values `vpn`, +`bluetooth` and `other`, and the `error.type` value `timed_out`. Names that the app chooses, such +as custom event names, and the checkpoint name patterns are not registry constants. The registry +records each rename as `renamed_from` on the new name. LiveKit-defined names +carry the `lk.` prefix, everything else follows +[OpenTelemetry semantic conventions](https://github.com/open-telemetry/semantic-conventions). Every record carries a wall-clock timestamp: one without `timestamp_ns` is stamped when it is 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.*`, +owner and timestamp, are taken at capture, its session's (`lk.room_sid`, `lk.participant_id`, …, `session.id`) included, so a record captured before a disconnect keeps its room; only 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 @@ -30,8 +36,8 @@ Set once per pipeline (`TelemetryConfig.resource`): One pipeline per process — started at SDK init, so audio pre-initialization, permission failures and connect attempts that never reach a server are captured — and one **scope** per room (one -call). A scope is a trace id plus the attributes attached to its records (`lk.room.sid`, -`lk.participant.identity`, …); spans, RTC windows and events are filed under the scope that +call). A scope is a trace id plus the attributes attached to its records (`lk.room_sid`, +`lk.pii.participant_identity`, …); spans, RTC windows and events are filed under the scope that produced them, and `session.id` (OTel semconv) is written on every record as an attribute. A log record emitted inside a room's span is filed under that room's scope; anything emitted outside a scope — device state, pre-room errors, self-telemetry — belongs to the pipeline's own process @@ -240,7 +246,7 @@ cadence: one per track and direction per stats window (default 60 s, stretched b (`stats_poll_interval_ms`). Keyed by session and track: two rooms receiving the same track keep two windows. attributes: - lk.track.sid: string + lk.track_sid: string lk.track.kind: enum(audio | video) lk.track.direction: enum(inbound | outbound) lk.rtc.codec: string # mimeType, when known @@ -483,10 +489,10 @@ exempt. ```yaml event: lk.room.disconnected area: session -severity: info (client_initiated) | warn (anything else) +severity: info (CLIENT_INITIATED) | warn (anything else) cadence: once, when the Room leaves connected for good — never on a reconnect attributes: - lk.disconnect.reason: enum(client_initiated | duplicate_identity | server_shutdown | participant_removed | room_deleted | state_mismatch | join_failure | migration | signal_close | room_closed | user_unavailable | user_rejected | sip_trunk_failure | connection_timeout | media_failure | agent_error | reconnect_failed | unknown) # the protocol's DisconnectReason, plus the client giving up + lk.disconnect_reason: enum(CLIENT_INITIATED | DUPLICATE_IDENTITY | SERVER_SHUTDOWN | PARTICIPANT_REMOVED | ROOM_DELETED | STATE_MISMATCH | JOIN_FAILURE | MIGRATION | SIGNAL_CLOSE | ROOM_CLOSED | USER_UNAVAILABLE | USER_REJECTED | SIP_TRUNK_FAILURE | CONNECTION_TIMEOUT | MEDIA_FAILURE | AGENT_ERROR | RECONNECT_FAILED | UNKNOWN_REASON) # the protobuf DisconnectReason names, as Agents sends them, plus RECONNECT_FAILED (the client giving up); the body keeps snake case (`disconnected: client_initiated`) platforms: all ``` @@ -540,8 +546,8 @@ parent: the ambient span, when any; a pre-connect publish (a microphone publishe connect completes) is its own span in the session's trace — `lk.connect` has ended by then attributes: lk.track.kind: enum(audio | video) - lk.track.source: enum(camera | microphone | screen_share | screen_share_audio | unknown) - lk.track.sid: string # on success + lk.track_source: enum(SOURCE_CAMERA | SOURCE_MICROPHONE | SOURCE_SCREENSHARE | SOURCE_SCREENSHARE_AUDIO | SOURCE_UNKNOWN) # the RTC FFI TrackSource names, as Agents sends them + lk.track_sid: string # on success outcome: ok | error (error.type) | cancelled ``` @@ -560,10 +566,10 @@ ends: at first media (the first inbound stats reading with bytes; the core see owner: the core (`Scope::subscribe_started / subscribed / track_ended / subscribe_failed`); an SDK only reports the remote track's lifecycle attributes: - lk.track.sid: string + lk.track_sid: string lk.track.kind: enum(audio | video) - lk.track.source: enum(camera | microphone | screen_share | screen_share_audio | unknown) - lk.participant.remote_identity: string + lk.track_source: enum(SOURCE_CAMERA | SOURCE_MICROPHONE | SOURCE_SCREENSHARE | SOURCE_SCREENSHARE_AUDIO | SOURCE_UNKNOWN) + lk.pii.remote_participant_identity: string checkpoints: subscribed, first_media ``` @@ -580,10 +586,10 @@ the policy. `Attribute { key, value }` survives only as the open bag: `emit_cust | `telemetry_disable()` | the opt-out, in effect when it returns: capture stops; everything unsent is then deleted | | `TelemetryConfig.sdk: TelemetryResource { sdk: Sdk, sdk_version, os_name, os_version, device_model }` | `service.name = livekit-client-`, `service.version`, `os.*`, `device.model.identifier`, plus `telemetry.sdk.*` | | `log(LogRecord { severity, source: LogSource, body, logger, function, file, line, timestamp_ns, span_id })` | a record with `code.function.name`, `code.file.path`, `code.line.number`, `lk.log.source`, `lk.log.logger`; the per-source floor (WebRTC at `error`, own module never) | -| `Scope::set_room(RoomIdentity { sid, name, participant_sid, participant_identity })` | `lk.room.*`, `lk.participant.*` on every record the scope captures from then on | +| `Scope::set_room(RoomIdentity { sid, name, participant_sid, participant_identity })` | `lk.room_sid`, `lk.pii.room_name`, `lk.participant_id`, `lk.pii.participant_identity` on every record the scope captures from then on | | `Scope::start(SpanName, parent) -> Span`; `Span::detached(name)` | an OTLP span (`lk.connect` / `lk.reconnect` are `client`, the rest `internal`); `Reconnect { reason }` sets `lk.reconnect.reason` | | `Span::step(SpanStep)` | a span event named `ws_open` … `room_connected`, `subscribed`, `first_media`, `attempt N quick|full` (which also sets `lk.reconnect.attempts` / `.mode`) | -| `Span::set_track(SpanTrack { sid, kind, source, remote_identity })` | `lk.track.sid`, `lk.track.kind`, `lk.track.source`, `lk.participant.remote_identity` | +| `Span::set_track(SpanTrack { sid, kind, source, remote_identity })` | `lk.track_sid`, `lk.track.kind`, `lk.track_source`, `lk.pii.remote_participant_identity` | | `Span::end(outcome, error)` / `fail(error)` / `cancel()` | status, `lk.outcome`, and `error.type` when it failed; ending twice is a no-op | | `Span::describe()` | `lk.connect: ws_open +1.49s, signal +0.03s, total 1.83s, ok` — the console line, identical on every platform | | `Span::context()` | `TraceContext { trace_id, span_id }` for log correlation; `None` when detached | @@ -595,7 +601,7 @@ the policy. `Attribute { key, value }` survives only as the open bag: `emit_cust | `Scope::record_stats_report(sid, kind, direction, Vec, ts)` | the same mapping for one track's report (platforms that poll per track) | | `DisconnectReason::from(i32)`, `ReconnectReason::from(i32)` | the protocol numbers → the shared enums (an unknown number is `Unknown`) | | `Scope::log(LogRecord)` | a record filed under the session without an ambient span (Dart has no task-local outside a zone); same floor and filters as `Telemetry::log` | -| `Scope::disconnected(DisconnectReason)` | `lk.room.disconnected` with `lk.disconnect.reason`, once — info when the client hung up, warn otherwise; pending subscribes end, the session's RTC windows ship and its state is retired; a repeated call is a no-op until a new `lk.connect` | +| `Scope::disconnected(DisconnectReason)` | `lk.room.disconnected` with `lk.disconnect_reason`, once — info when the client hung up, warn otherwise; pending subscribes end, the session's RTC windows ship and its state is retired; a repeated call is a no-op until a new `lk.connect` | | `RtcStatsSample.layer` (rid, ssrc or stats id) | simulcast layers folded into one monotonic series per track before windowing | Timing rule: span calls are synchronous and stamp the clock inside the core, so the only skew is diff --git a/livekit-telemetry/src/device.rs b/livekit-telemetry/src/device.rs index feb8ea019..5ca12f440 100644 --- a/livekit-telemetry/src/device.rs +++ b/livekit-telemetry/src/device.rs @@ -12,6 +12,13 @@ // See the License for the specific language governing permissions and // limitations under the License. +use livekit_protocol::semconv::{ + self, event, lk_device_app_state as app_state, lk_device_audio_interruption as interruption, + lk_device_audio_route_reason as route_reason, lk_device_capture_device as capture_device, + lk_device_capture_reason as capture_reason, lk_device_memory_pressure as memory, + lk_device_thermal_state as thermal, upstream, upstream::network_connection_type as network, +}; + use crate::TelemetryEvent; /// Thermal pressure as the OS reports it (`ProcessInfo.thermalState`, `PowerManager` thermal @@ -201,9 +208,9 @@ impl DeviceState { && previous.is_none_or(|p| p.thermal != self.thermal) { events.push( - TelemetryEvent::new("lk.device.thermal.changed") + TelemetryEvent::new(event::LK_DEVICE_THERMAL_CHANGED) .with_body(format!("thermal: {}", self.thermal.as_str())) - .with_attribute("lk.device.thermal.state", self.thermal.as_str()), + .with_attribute(semconv::LK_DEVICE_THERMAL_STATE, self.thermal.as_str()), ); } if let Some(low_power) = self @@ -211,23 +218,23 @@ impl DeviceState { .filter(|_| previous.is_none_or(|p| p.low_power_mode != self.low_power_mode)) { events.push( - TelemetryEvent::new("lk.device.low_power.changed") + TelemetryEvent::new(event::LK_DEVICE_LOW_POWER_CHANGED) .with_body(format!("low power mode: {}", if low_power { "on" } else { "off" })) - .with_attribute("lk.device.low_power.enabled", low_power), + .with_attribute(semconv::LK_DEVICE_LOW_POWER_ENABLED, low_power), ); } if previous.is_none_or(|p| p.app_state != self.app_state) { events.push( - TelemetryEvent::new("lk.device.app_state.changed") + TelemetryEvent::new(event::LK_DEVICE_APP_STATE_CHANGED) .with_body(format!("app: {}", self.app_state.as_str())) - .with_attribute("lk.device.app_state", self.app_state.as_str()), + .with_attribute(semconv::LK_DEVICE_APP_STATE, self.app_state.as_str()), ); } if previous.is_none_or(|p| p.memory != self.memory) { events.push( - TelemetryEvent::new("lk.device.memory.changed") + TelemetryEvent::new(event::LK_DEVICE_MEMORY_CHANGED) .with_body(format!("memory pressure: {}", self.memory.as_str())) - .with_attribute("lk.device.memory.pressure", self.memory.as_str()), + .with_attribute(semconv::LK_DEVICE_MEMORY_PRESSURE, self.memory.as_str()), ); } if previous.is_none_or(|p| { @@ -235,16 +242,19 @@ impl DeviceState { != (self.network, self.network_expensive, self.network_constrained) }) { events.push( - TelemetryEvent::new("lk.device.network.changed") + TelemetryEvent::new(event::LK_DEVICE_NETWORK_CHANGED) .with_body(format!( "network: {}{}{}", self.network.as_str(), if self.network_expensive { ", expensive" } else { "" }, if self.network_constrained { ", constrained" } else { "" } )) - .with_attribute("network.connection.type", self.network.as_str()) - .with_attribute("lk.device.network.expensive", self.network_expensive) - .with_attribute("lk.device.network.constrained", self.network_constrained), + .with_attribute(upstream::NETWORK_CONNECTION_TYPE, self.network.as_str()) + .with_attribute(semconv::LK_DEVICE_NETWORK_EXPENSIVE, self.network_expensive) + .with_attribute( + semconv::LK_DEVICE_NETWORK_CONSTRAINED, + self.network_constrained, + ), ); } // Per-percent updates are noise; report charging flips and bucket crossings. @@ -256,13 +266,13 @@ impl DeviceState { }) }) { events.push( - TelemetryEvent::new("lk.device.battery.changed") + TelemetryEvent::new(event::LK_DEVICE_BATTERY_CHANGED) .with_body(format!( "battery: {level}%{}", if self.battery_charging { ", charging" } else { "" } )) - .with_attribute("lk.device.battery.level", i64::from(level)) - .with_attribute("lk.device.battery.charging", self.battery_charging), + .with_attribute(semconv::LK_DEVICE_BATTERY_LEVEL, i64::from(level)) + .with_attribute(semconv::LK_DEVICE_BATTERY_CHARGING, self.battery_charging), ); } events @@ -272,11 +282,12 @@ impl DeviceState { impl ThermalState { fn as_str(self) -> &'static str { match self { + // Never sent: an unknown reading makes no change event. ThermalState::Unknown => "unknown", - ThermalState::Nominal => "nominal", - ThermalState::Fair => "fair", - ThermalState::Serious => "serious", - ThermalState::Critical => "critical", + ThermalState::Nominal => thermal::NOMINAL, + ThermalState::Fair => thermal::FAIR, + ThermalState::Serious => thermal::SERIOUS, + ThermalState::Critical => thermal::CRITICAL, } } } @@ -284,8 +295,8 @@ impl ThermalState { impl AppState { fn as_str(self) -> &'static str { match self { - AppState::Foreground => "foreground", - AppState::Background => "background", + AppState::Foreground => app_state::FOREGROUND, + AppState::Background => app_state::BACKGROUND, } } } @@ -293,9 +304,9 @@ impl AppState { impl MemoryPressure { fn as_str(self) -> &'static str { match self { - MemoryPressure::Normal => "normal", - MemoryPressure::Warning => "warning", - MemoryPressure::Critical => "critical", + MemoryPressure::Normal => memory::NORMAL, + MemoryPressure::Warning => memory::WARNING, + MemoryPressure::Critical => memory::CRITICAL, } } } @@ -303,14 +314,15 @@ impl MemoryPressure { impl NetworkType { fn as_str(self) -> &'static str { match self { - NetworkType::Unknown => "unknown", - NetworkType::Wifi => "wifi", - NetworkType::Cell => "cell", - NetworkType::Wired => "wired", + NetworkType::Unknown => network::UNKNOWN, + NetworkType::Wifi => network::WIFI, + NetworkType::Cell => network::CELL, + NetworkType::Wired => network::WIRED, + // Not OTel members; the registry notes that clients send them. NetworkType::Vpn => "vpn", NetworkType::Bluetooth => "bluetooth", NetworkType::Other => "other", - NetworkType::Unavailable => "unavailable", + NetworkType::Unavailable => network::UNAVAILABLE, } } } @@ -418,9 +430,46 @@ pub enum DeviceEvent { }, } -/// A variant's wire value: its `Debug` name in snake case (`OldDeviceUnavailable` → -/// `old_device_unavailable`). Variant names are therefore the wire contract: renaming a variant -/// changes what ships, and the tests pin every value to `SPEC.md`. +impl AudioRouteReason { + fn as_str(self) -> &'static str { + match self { + Self::NewDevice => route_reason::NEW_DEVICE, + Self::OldDeviceUnavailable => route_reason::OLD_DEVICE_UNAVAILABLE, + Self::CategoryChange => route_reason::CATEGORY_CHANGE, + Self::Override => route_reason::OVERRIDE, + Self::WakeFromSleep => route_reason::WAKE_FROM_SLEEP, + Self::NoSuitableRoute => route_reason::NO_SUITABLE_ROUTE, + Self::RouteConfigurationChange => route_reason::ROUTE_CONFIGURATION_CHANGE, + Self::Unknown => route_reason::UNKNOWN, + } + } +} + +impl CaptureDevice { + fn as_str(self) -> &'static str { + match self { + Self::Camera => capture_device::CAMERA, + Self::Microphone => capture_device::MICROPHONE, + Self::ScreenShare => capture_device::SCREEN_SHARE, + } + } +} + +impl CaptureFailure { + fn as_str(self) -> &'static str { + match self { + Self::PermissionDenied => capture_reason::PERMISSION_DENIED, + Self::NotFound => capture_reason::NOT_FOUND, + Self::InUse => capture_reason::IN_USE, + Self::Disconnected => capture_reason::DISCONNECTED, + Self::Other => capture_reason::OTHER, + } + } +} + +/// A variant's `Debug` name in snake case (`WiredHeadset` → `wired_headset`): the wire value of +/// an [`AudioOutput`], which the registry does not enumerate, and the display bodies. Renaming +/// such a variant changes what ships; the tests pin every value to `SPEC.md`. pub(crate) fn snake_case(debug: impl std::fmt::Debug) -> String { let mut out = String::new(); for (i, c) in format!("{debug:?}").chars().enumerate() { @@ -438,25 +487,25 @@ impl From for TelemetryEvent { match event { DeviceEvent::AudioRouteChanged { outputs, reason } => { let outputs = outputs.into_iter().map(snake_case).collect::>().join(","); - let reason = snake_case(reason); - TelemetryEvent::new("lk.device.audio_route.changed") + let reason = reason.as_str(); + TelemetryEvent::new(event::LK_DEVICE_AUDIO_ROUTE_CHANGED) .with_body(format!("audio route: {outputs} ({reason})")) - .with_attribute("lk.device.audio_route.outputs", outputs) - .with_attribute("lk.device.audio_route.reason", reason) + .with_attribute(semconv::LK_DEVICE_AUDIO_ROUTE_OUTPUTS, outputs) + .with_attribute(semconv::LK_DEVICE_AUDIO_ROUTE_REASON, reason) } DeviceEvent::AudioInterruption { began } => { - let phase = if began { "began" } else { "ended" }; - TelemetryEvent::new("lk.device.audio.interruption") + let phase = if began { interruption::BEGAN } else { interruption::ENDED }; + TelemetryEvent::new(event::LK_DEVICE_AUDIO_INTERRUPTION) .with_body(format!("audio interruption {phase}")) - .with_attribute("lk.device.audio.interruption", phase) + .with_attribute(semconv::LK_DEVICE_AUDIO_INTERRUPTION, phase) } DeviceEvent::CaptureFailed { device, reason } => { - let (device, reason) = (snake_case(device), snake_case(reason)); - TelemetryEvent::new("lk.device.capture.failed") + let (device, reason) = (device.as_str(), reason.as_str()); + TelemetryEvent::new(event::LK_DEVICE_CAPTURE_FAILED) .with_severity(crate::Severity::Warn) .with_body(format!("capture failed: {device} ({reason})")) - .with_attribute("lk.device.capture.device", device) - .with_attribute("lk.device.capture.reason", reason) + .with_attribute(semconv::LK_DEVICE_CAPTURE_DEVICE, device) + .with_attribute(semconv::LK_DEVICE_CAPTURE_REASON, reason) } } } @@ -598,7 +647,7 @@ mod tests { R::Unknown, ]; assert_eq!( - names(reasons.into_iter().map(snake_case).collect()), + names(reasons.into_iter().map(|r| r.as_str().to_owned()).collect()), "new_device | old_device_unavailable | category_change | override | wake_from_sleep \ | no_suitable_route | route_configuration_change | unknown" ); @@ -622,13 +671,13 @@ mod tests { let devices = [CaptureDevice::Camera, CaptureDevice::Microphone, CaptureDevice::ScreenShare]; assert_eq!( - names(devices.into_iter().map(snake_case).collect()), + names(devices.into_iter().map(|d| d.as_str().to_owned()).collect()), "camera | microphone | screen_share" ); use CaptureFailure as F; let failures = [F::PermissionDenied, F::NotFound, F::InUse, F::Disconnected, F::Other]; assert_eq!( - names(failures.into_iter().map(snake_case).collect()), + names(failures.into_iter().map(|f| f.as_str().to_owned()).collect()), "permission_denied | not_found | in_use | disconnected | other" ); } diff --git a/livekit-telemetry/src/device_tests.rs b/livekit-telemetry/src/device_tests.rs index 79fc1e390..8dc65203c 100644 --- a/livekit-telemetry/src/device_tests.rs +++ b/livekit-telemetry/src/device_tests.rs @@ -294,7 +294,7 @@ async fn a_subscribe_past_its_deadline_stays_timed_out_through_disconnect() { .find(|s| { s.attributes .iter() - .any(|a| a.key == "lk.track.sid" && format!("{:?}", a.value).contains(sid)) + .any(|a| a.key == "lk.track_sid" && format!("{:?}", a.value).contains(sid)) }) .expect("span"); span.attributes.iter().find(|a| a.key == "lk.outcome").map(|a| format!("{:?}", a.value)) @@ -420,12 +420,12 @@ async fn custom_data_is_room_scoped_validated_and_snapshotted() { other.emit_custom("elsewhere", vec![]); let long = "x".repeat(1025); - room.set_attribute("lk.room.sid", Some("spoof".into())); + room.set_attribute("lk.room_sid", Some("spoof".into())); room.set_attribute("session.id", Some("spoof".into())); room.set_attribute("app.long", Some(long.clone().into())); room.set_attribute(&"k".repeat(129), Some("v".into())); room.emit_custom(&"n".repeat(129), vec![]); - room.emit_custom("bad", vec![Attribute::new("lk.track.sid", "spoof")]); + room.emit_custom("bad", vec![Attribute::new("lk.track_sid", "spoof")]); room.emit_custom("bad", vec![Attribute::new("v", long.as_str())]); for n in 0..63 { room.set_attribute(&format!("app.k{n}"), Some("v".into())); diff --git a/livekit-telemetry/src/event.rs b/livekit-telemetry/src/event.rs index 9d51ef29f..72fe51f23 100644 --- a/livekit-telemetry/src/event.rs +++ b/livekit-telemetry/src/event.rs @@ -14,6 +14,8 @@ use std::time::{SystemTime, UNIX_EPOCH}; +use livekit_protocol::semconv::{self, lk_log_source as source, upstream}; + /// A discrete telemetry event. /// /// Exported as one OTLP log record whose `event_name` is [`name`](Self::name), following the @@ -90,9 +92,9 @@ impl TelemetryEvent { /// when it doubles as the body. /// /// Counts the attributes the record carries: once queued, that includes its session's - /// (`lk.room.*`, `lk.participant.*`, `session.id`) and the app's correlation attributes, all - /// taken at capture. Pipeline-wide attributes are added at export: the exporter's drains count - /// them for every record (`Store::drain`, `Spans::drain`). + /// (`lk.room_sid`, `lk.participant_id`, …, `session.id`) and the app's correlation + /// attributes, all taken at capture. Pipeline-wide attributes are added at export: the + /// exporter's drains count them for every record (`Store::drain`, `Spans::drain`). pub fn size_hint(&self) -> usize { RECORD_OVERHEAD_BYTES + 2 * self.name.len() @@ -160,9 +162,9 @@ pub enum LogSource { impl LogSource { fn as_str(self) -> &'static str { match self { - Self::Sdk => "sdk", - Self::Ffi => "ffi", - Self::WebRtc => "webrtc", + Self::Sdk => source::SDK, + Self::Ffi => source::FFI, + Self::WebRtc => source::WEBRTC, } } } @@ -208,18 +210,18 @@ impl From for TelemetryEvent { let mut event = TelemetryEvent::default() .with_severity(record.severity) .with_body(record.body) - .with_attribute("lk.log.source", record.source.as_str()); + .with_attribute(semconv::LK_LOG_SOURCE, record.source.as_str()); if let Some(logger) = record.logger.filter(|s| !s.is_empty()) { - event = event.with_attribute("lk.log.logger", logger); + event = event.with_attribute(semconv::LK_LOG_LOGGER, logger); } if let Some(function) = record.function.filter(|s| !s.is_empty()) { - event = event.with_attribute("code.function.name", function); + event = event.with_attribute(upstream::CODE_FUNCTION_NAME, function); } if let Some(file) = record.file.filter(|s| !s.is_empty()) { - event = event.with_attribute("code.file.path", file); + event = event.with_attribute(upstream::CODE_FILE_PATH, file); } if let Some(line) = record.line.filter(|l| *l > 0) { - event = event.with_attribute("code.line.number", i64::from(line)); + event = event.with_attribute(upstream::CODE_LINE_NUMBER, i64::from(line)); } event.timestamp_ns = record.timestamp_ns; event.span_id = record.span_id; @@ -326,9 +328,9 @@ pub(crate) fn is_reserved(key: &str) -> bool { key.starts_with("lk.") || key.starts_with("otel.") || key.starts_with("code.") - || key == "session.id" - || key == "error.type" - || key == "network.connection.type" + || key == upstream::SESSION_ID + || key == upstream::ERROR_TYPE + || key == upstream::NETWORK_CONNECTION_TYPE } /// Whether an app-provided attribute is within the limits and outside the SDK's namespace. diff --git a/livekit-telemetry/src/exporter.rs b/livekit-telemetry/src/exporter.rs index c695aa959..dff421747 100644 --- a/livekit-telemetry/src/exporter.rs +++ b/livekit-telemetry/src/exporter.rs @@ -22,6 +22,7 @@ use std::{ }; use flate2::{write::GzEncoder, Compression}; +use livekit_protocol::semconv::span; use tokio::sync::{mpsc, oneshot}; use tokio::time::{sleep_until, timeout, Instant}; @@ -68,7 +69,7 @@ const MAX_AGE: Duration = Duration::from_secs(24 * 60 * 60); /// anyway: the cap that bounds the policy when its signals lie. Hard holds have no such escape. const MAX_HOLD: Duration = Duration::from_secs(60); /// While one of these is open the uplink belongs to signaling and ICE/DTLS. -pub(crate) const SENSITIVE_SPANS: &[&str] = &["lk.connect", "lk.reconnect"]; +pub(crate) const SENSITIVE_SPANS: &[&str] = &[span::LK_CONNECT, span::LK_RECONNECT]; /// RFC 9218 request priority: lowest urgency, not incremental. A hint for HTTP/2+ hops that /// implement it; the host transport marks the local traffic class (see `TelemetryTransport`). const PRIORITY: &str = "u=7"; @@ -1758,12 +1759,12 @@ pub(crate) mod tests { #[test] fn gzip_shrinks_a_batch_of_records_at_least_this_much() { let session = ScopeState::new(); - session.set_attribute("lk.room.sid", Some("RM_UpE3vYzFmJp8".into())); + session.set_attribute("lk.room_sid", Some("RM_UpE3vYzFmJp8".into())); let records = (0..100) .map(|i| { let event = TelemetryEvent::new("lk.rtc.stats.sample") .with_body(format!("video inbound: {i} kbps, rtt 48 ms, 30 fps")) - .with_attribute("lk.track.sid", "TR_VCd8tCgNdE9h") + .with_attribute("lk.track_sid", "TR_VCd8tCgNdE9h") .with_attribute("lk.rtc.bytes", i as i64 * 1_000); QueuedEvent::new(event, session.clone()) }) diff --git a/livekit-telemetry/src/lib.rs b/livekit-telemetry/src/lib.rs index 43548196a..4b3cec7a6 100644 --- a/livekit-telemetry/src/lib.rs +++ b/livekit-telemetry/src/lib.rs @@ -79,6 +79,9 @@ mod backend_tests; /// The device contract, end to end through the pipeline and the process-wide opt-out. #[cfg(test)] mod device_tests; +/// Every name and enum value on the wire, pinned to the registry. +#[cfg(test)] +mod wire_tests; #[cfg(feature = "uniffi")] uniffi::setup_scaffolding!(); diff --git a/livekit-telemetry/src/otlp.rs b/livekit-telemetry/src/otlp.rs index 9716de74b..84a5d2191 100644 --- a/livekit-telemetry/src/otlp.rs +++ b/livekit-telemetry/src/otlp.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use livekit_protocol::semconv::{self, upstream}; use prost::Message; use crate::span::SpanKind; @@ -135,9 +136,9 @@ fn log_record( // 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"); + event.attributes.retain(|a| a.key != upstream::OTEL_EVENT_NAME); if !event.name.is_empty() { - event.attributes.push(Attribute::new("otel.event.name", event.name.clone())); + event.attributes.push(Attribute::new(upstream::OTEL_EVENT_NAME, event.name.clone())); } last_value_wins(&mut event.attributes); LogRecord { @@ -160,7 +161,7 @@ fn log_record( fn otlp_span(mut record: SpanRecord, global: &[Attribute]) -> Span { 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"); + record.attributes.retain(|a| a.key != semconv::LK_OUTCOME && a.key != upstream::ERROR_TYPE); let mut attributes: Vec = record.attributes.iter().map(KeyValue::from).collect(); attributes.extend(record.outcome_attributes().iter().map(KeyValue::from)); Span { @@ -393,10 +394,10 @@ mod tests { /// The attributes `Scope::set_room` gives a session in a call. fn in_room(session: &crate::scope::ScopeState) { for (key, value) in [ - ("lk.room.sid", "RM_UpE3vYzFmJp8"), - ("lk.room.name", "standup-2026-10-08"), - ("lk.participant.sid", "PA_kQ2Fv8sWbE3n"), - ("lk.participant.identity", "user-4f1c2a"), + ("lk.room_sid", "RM_UpE3vYzFmJp8"), + ("lk.pii.room_name", "standup-2026-10-08"), + ("lk.participant_id", "PA_kQ2Fv8sWbE3n"), + ("lk.pii.participant_identity", "user-4f1c2a"), ] { session.set_attribute(key, Some(value.into())); } @@ -416,16 +417,16 @@ mod tests { let session = crate::scope::ScopeState::new(); in_room(&session); let spoofed = - TelemetryEvent::new("lk.room.disconnected").with_attribute("lk.room.sid", "x"); + TelemetryEvent::new("lk.room.disconnected").with_attribute("lk.room_sid", "x"); let queued = QueuedEvent::new(spoofed, session.clone()); // The platform clears the room right after the disconnect, before the next export. - session.set_attribute("lk.room.sid", None); - let global = [Attribute::new("lk.room.sid", "global"), Attribute::new("acme.tenant", "t1")]; + session.set_attribute("lk.room_sid", None); + let global = [Attribute::new("lk.room_sid", "global"), Attribute::new("acme.tenant", "t1")]; let bytes = encode_logs(&[], &global, vec![queued]); let decoded = ExportLogsServiceRequest::decode(&bytes[..]).expect("valid OTLP"); let record = &decoded.resource_logs[0].scope_logs[0].log_records[0]; let text = |s: &str| any_value::Value::StringValue(s.into()); - assert_eq!(values(&record.attributes, "lk.room.sid"), [text("RM_UpE3vYzFmJp8")]); + assert_eq!(values(&record.attributes, "lk.room_sid"), [text("RM_UpE3vYzFmJp8")]); assert_eq!(values(&record.attributes, "session.id"), [text(session.hex())]); assert_eq!(values(&record.attributes, "acme.tenant"), [text("t1")], "globals fill in"); } @@ -436,13 +437,13 @@ mod tests { in_room(&session); let mut spans = crate::span::Spans::new(1); let id = spans.begin_in("lk.publish", SpanKind::Internal, None, session.clone()); - spans.end(id, SpanOutcome::Ok, None, vec![Attribute::new("lk.room.sid", "x")]); - session.set_attribute("lk.room.sid", None); + spans.end(id, SpanOutcome::Ok, None, vec![Attribute::new("lk.room_sid", "x")]); + session.set_attribute("lk.room_sid", None); let bytes = encode_spans(&[], &[], spans.drain(1, usize::MAX, 0)); let decoded = ExportTraceServiceRequest::decode(&bytes[..]).expect("valid OTLP"); let span = &decoded.resource_spans[0].scope_spans[0].spans[0]; let text = |s: &str| any_value::Value::StringValue(s.into()); - assert_eq!(values(&span.attributes, "lk.room.sid"), [text("RM_UpE3vYzFmJp8")]); + assert_eq!(values(&span.attributes, "lk.room_sid"), [text("RM_UpE3vYzFmJp8")]); assert_eq!(values(&span.attributes, "session.id"), [text(session.hex())]); } diff --git a/livekit-telemetry/src/rtc.rs b/livekit-telemetry/src/rtc.rs index 8ea218285..8eb1af868 100644 --- a/livekit-telemetry/src/rtc.rs +++ b/livekit-telemetry/src/rtc.rs @@ -18,6 +18,9 @@ use std::{ time::Duration, }; +use livekit_protocol::semconv::{ + self, event, lk_track_direction as direction, lk_track_kind as kind, +}; use tokio::time::Instant; use crate::{ @@ -460,12 +463,13 @@ impl Gauge { (self.n > 0).then(|| self.sum / self.n as f64) } - fn attach(&self, event: TelemetryEvent, key: &str) -> TelemetryEvent { + /// The `.min`, `.max` and `.avg` keys, in that order. + fn attach(&self, event: TelemetryEvent, [min, max, avg_key]: [&str; 3]) -> TelemetryEvent { let Some(avg) = self.avg() else { return event }; event - .with_attribute(format!("{key}.min"), self.min) - .with_attribute(format!("{key}.max"), self.max) - .with_attribute(format!("{key}.avg"), avg) + .with_attribute(min, self.min) + .with_attribute(max, self.max) + .with_attribute(avg_key, avg) } } @@ -573,44 +577,64 @@ impl Window { let window_ms = end_ns.saturating_sub(self.start_ns) / 1_000_000; let body = self.summary(); let last = self.last; - let mut event = TelemetryEvent::new("lk.rtc.stats.sample") + let mut event = TelemetryEvent::new(event::LK_RTC_STATS_SAMPLE) .with_body(body) - .with_attribute("lk.track.sid", last.track_sid) - .with_attribute("lk.track.kind", last.kind.as_str()) - .with_attribute("lk.track.direction", last.direction.as_str()) - .with_attribute("lk.rtc.window_ms", window_ms as i64) - .with_attribute("lk.rtc.samples", self.samples as i64); + .with_attribute(semconv::LK_TRACK_SID, last.track_sid) + .with_attribute(semconv::LK_TRACK_KIND, last.kind.as_str()) + .with_attribute(semconv::LK_TRACK_DIRECTION, last.direction.as_str()) + .with_attribute(semconv::LK_RTC_WINDOW_MS, window_ms as i64) + .with_attribute(semconv::LK_RTC_SAMPLES, self.samples as i64); if let Some(codec) = last.codec { - event = event.with_attribute("lk.rtc.codec", codec); + event = event.with_attribute(semconv::LK_RTC_CODEC, codec); } let counters = [ - ("lk.rtc.bytes", last.bytes), - ("lk.rtc.packets", last.packets), - ("lk.rtc.packets_lost", last.packets_lost), - ("lk.rtc.freeze_count", last.freeze_count), - ("lk.rtc.freezes_duration_ms", last.freezes_duration_ms), - ("lk.rtc.concealed_samples", last.concealed_samples), - ("lk.rtc.concealment_events", last.concealment_events), - ("lk.rtc.jitter_buffer_delay_ms", last.jitter_buffer_delay_ms), - ("lk.rtc.jitter_buffer_emitted_count", last.jitter_buffer_emitted_count), - ("lk.rtc.quality_limitation.bandwidth_ms", last.quality_limitation_bandwidth_ms), - ("lk.rtc.quality_limitation.cpu_ms", last.quality_limitation_cpu_ms), - ("lk.rtc.quality_limitation.other_ms", last.quality_limitation_other_ms), - ("lk.rtc.pause_count", last.pause_count), - ("lk.rtc.pauses_duration_ms", last.pauses_duration_ms), - ("lk.rtc.silent_concealed_samples", last.silent_concealed_samples), - ("lk.rtc.interruption_count", last.interruption_count), - ("lk.rtc.interruptions_duration_ms", last.interruptions_duration_ms), + (semconv::LK_RTC_BYTES, last.bytes), + (semconv::LK_RTC_PACKETS, last.packets), + (semconv::LK_RTC_PACKETS_LOST, last.packets_lost), + (semconv::LK_RTC_FREEZE_COUNT, last.freeze_count), + (semconv::LK_RTC_FREEZES_DURATION_MS, last.freezes_duration_ms), + (semconv::LK_RTC_CONCEALED_SAMPLES, last.concealed_samples), + (semconv::LK_RTC_CONCEALMENT_EVENTS, last.concealment_events), + (semconv::LK_RTC_JITTER_BUFFER_DELAY_MS, last.jitter_buffer_delay_ms), + (semconv::LK_RTC_JITTER_BUFFER_EMITTED_COUNT, last.jitter_buffer_emitted_count), + (semconv::LK_RTC_QUALITY_LIMITATION_BANDWIDTH_MS, last.quality_limitation_bandwidth_ms), + (semconv::LK_RTC_QUALITY_LIMITATION_CPU_MS, last.quality_limitation_cpu_ms), + (semconv::LK_RTC_QUALITY_LIMITATION_OTHER_MS, last.quality_limitation_other_ms), + (semconv::LK_RTC_PAUSE_COUNT, last.pause_count), + (semconv::LK_RTC_PAUSES_DURATION_MS, last.pauses_duration_ms), + (semconv::LK_RTC_SILENT_CONCEALED_SAMPLES, last.silent_concealed_samples), + (semconv::LK_RTC_INTERRUPTION_COUNT, last.interruption_count), + (semconv::LK_RTC_INTERRUPTIONS_DURATION_MS, last.interruptions_duration_ms), ]; for (key, value) in counters { if let Some(value) = value { event = event.with_attribute(key, value as i64); } } - event = self.jitter.attach(event, "lk.rtc.jitter_ms"); - event = self.rtt.attach(event, "lk.rtc.rtt_ms"); - event = self.fps.attach(event, "lk.rtc.fps"); - event = self.audio_level.attach(event, "lk.rtc.audio_level"); + event = self.jitter.attach( + event, + [ + semconv::LK_RTC_JITTER_MS_MIN, + semconv::LK_RTC_JITTER_MS_MAX, + semconv::LK_RTC_JITTER_MS_AVG, + ], + ); + event = self.rtt.attach( + event, + [semconv::LK_RTC_RTT_MS_MIN, semconv::LK_RTC_RTT_MS_MAX, semconv::LK_RTC_RTT_MS_AVG], + ); + event = self.fps.attach( + event, + [semconv::LK_RTC_FPS_MIN, semconv::LK_RTC_FPS_MAX, semconv::LK_RTC_FPS_AVG], + ); + event = self.audio_level.attach( + event, + [ + semconv::LK_RTC_AUDIO_LEVEL_MIN, + semconv::LK_RTC_AUDIO_LEVEL_MAX, + semconv::LK_RTC_AUDIO_LEVEL_AVG, + ], + ); event.timestamp_ns = Some(end_ns); event } @@ -919,10 +943,10 @@ fn test_session() -> Arc { } impl TrackKind { - fn as_str(self) -> &'static str { + pub(crate) fn as_str(self) -> &'static str { match self { - TrackKind::Audio => "audio", - TrackKind::Video => "video", + TrackKind::Audio => kind::AUDIO, + TrackKind::Video => kind::VIDEO, } } } @@ -930,8 +954,8 @@ impl TrackKind { impl StreamDirection { fn as_str(self) -> &'static str { match self { - StreamDirection::Inbound => "inbound", - StreamDirection::Outbound => "outbound", + StreamDirection::Inbound => direction::INBOUND, + StreamDirection::Outbound => direction::OUTBOUND, } } } diff --git a/livekit-telemetry/src/scope.rs b/livekit-telemetry/src/scope.rs index c8035588d..82ae294c9 100644 --- a/livekit-telemetry/src/scope.rs +++ b/livekit-telemetry/src/scope.rs @@ -19,6 +19,7 @@ use std::{ time::Duration, }; +use livekit_protocol::semconv::{self, event, lk_disconnect_reason as reason, upstream}; use tokio::time::Instant; use crate::{ @@ -28,12 +29,12 @@ use crate::{ }; /// One session's identity: the trace id every one of its records carries, and the attributes -/// copied into them when they are captured (`lk.room.sid`, `lk.participant.identity`, …). +/// copied into them when they are captured (`lk.room_sid`, `lk.pii.participant_identity`, …). pub(crate) struct ScopeState { pub trace_id: [u8; 16], /// The trace id as hex, formatted once: the session's owner in `Destinations`. hex: String, - /// SDK-owned attributes (`lk.room.*`, `lk.participant.*`), copied into each record when it is + /// SDK-owned attributes (the room and local participant), copied into each record when it is /// captured (see [`ScopeState::decorate`]). attributes: Mutex>, /// The app's correlation attributes, copied into each record when it is captured, so a @@ -278,8 +279,8 @@ impl ScopeState { own.push(attribute.clone()); } drop(session); - own.retain(|a| a.key != "session.id"); - own.push(Attribute::new("session.id", self.hex())); + own.retain(|a| a.key != upstream::SESSION_ID); + own.push(Attribute::new(upstream::SESSION_ID, self.hex())); } } @@ -307,16 +308,16 @@ impl fmt::Debug for ScopeState { #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct RoomIdentity { - /// The room's sid (`RM_…`), `lk.room.sid`. + /// The room's sid (`RM_…`), `lk.room_sid`. #[cfg_attr(feature = "uniffi", uniffi(default))] pub sid: Option, - /// The room's name, `lk.room.name`. + /// The room's name, `lk.pii.room_name`. #[cfg_attr(feature = "uniffi", uniffi(default))] pub name: Option, - /// The local participant's sid (`PA_…`), `lk.participant.sid`. + /// The local participant's sid (`PA_…`), `lk.participant_id`. #[cfg_attr(feature = "uniffi", uniffi(default))] pub participant_sid: Option, - /// The local participant's identity, `lk.participant.identity`. + /// The local participant's identity, `lk.pii.participant_identity`. #[cfg_attr(feature = "uniffi", uniffi(default))] pub participant_identity: Option, } @@ -363,6 +364,33 @@ pub enum DisconnectReason { ReconnectFailed, } +impl DisconnectReason { + /// The protobuf `DisconnectReason` name, the spelling Agents sends; `RECONNECT_FAILED` is the + /// one client-only value. + fn as_str(self) -> &'static str { + match self { + Self::Unknown => reason::UNKNOWN_REASON, + Self::ClientInitiated => reason::CLIENT_INITIATED, + Self::DuplicateIdentity => reason::DUPLICATE_IDENTITY, + Self::ServerShutdown => reason::SERVER_SHUTDOWN, + Self::ParticipantRemoved => reason::PARTICIPANT_REMOVED, + Self::RoomDeleted => reason::ROOM_DELETED, + Self::StateMismatch => reason::STATE_MISMATCH, + Self::JoinFailure => reason::JOIN_FAILURE, + Self::Migration => reason::MIGRATION, + Self::SignalClose => reason::SIGNAL_CLOSE, + Self::RoomClosed => reason::ROOM_CLOSED, + Self::UserUnavailable => reason::USER_UNAVAILABLE, + Self::UserRejected => reason::USER_REJECTED, + Self::SipTrunkFailure => reason::SIP_TRUNK_FAILURE, + Self::ConnectionTimeout => reason::CONNECTION_TIMEOUT, + Self::MediaFailure => reason::MEDIA_FAILURE, + Self::AgentError => reason::AGENT_ERROR, + Self::ReconnectFailed => reason::RECONNECT_FAILED, + } + } +} + impl From for DisconnectReason { /// The protocol's `DisconnectReason` number, so no SDK keeps its own switch; a number this SDK /// does not know is `Unknown`. @@ -467,14 +495,13 @@ impl Scope { } else { crate::Severity::Warn }; - let reason = crate::device::snake_case(reason); // The body repeats the reason on purpose: it is the display line a log view shows (like // the device events' bodies), and it takes the place of the name a body-less event shows. self.emit( - TelemetryEvent::new("lk.room.disconnected") + TelemetryEvent::new(event::LK_ROOM_DISCONNECTED) .with_severity(severity) - .with_body(format!("disconnected: {reason}")) - .with_attribute("lk.disconnect.reason", reason), + .with_body(format!("disconnected: {}", crate::device::snake_case(reason))) + .with_attribute(semconv::LK_DISCONNECT_REASON, reason.as_str()), ); } @@ -676,14 +703,15 @@ impl Scope { Span::bound(name, parent, self.telemetry.clone(), &self.state) } - /// The room and local participant, as `lk.room.*` / `lk.participant.*` on every record - /// captured from then on. + /// The room and local participant, as `lk.room_sid`, `lk.pii.room_name`, + /// `lk.participant_id` and `lk.pii.participant_identity` on every record captured from then + /// on. pub fn set_room(&self, room: RoomIdentity) { for (key, value) in [ - ("lk.room.sid", room.sid), - ("lk.room.name", room.name), - ("lk.participant.sid", room.participant_sid), - ("lk.participant.identity", room.participant_identity), + (semconv::LK_ROOM_SID, room.sid), + (semconv::LK_PII_ROOM_NAME, room.name), + (semconv::LK_PARTICIPANT_ID, room.participant_sid), + (semconv::LK_PII_PARTICIPANT_IDENTITY, room.participant_identity), ] { // Identifiers, not free text: an over-long one is not attached (counted as invalid). if value.as_ref().is_some_and(|v| v.len() > crate::event::MAX_VALUE_BYTES) { @@ -928,11 +956,11 @@ mod tests { .map(DisconnectReason::from) .chain([DisconnectReason::ReconnectFailed, DisconnectReason::from(0)]); assert_eq!( - reasons.map(crate::device::snake_case).collect::>().join(" | "), - "client_initiated | duplicate_identity | server_shutdown | participant_removed \ - | room_deleted | state_mismatch | join_failure | migration | signal_close \ - | room_closed | user_unavailable | user_rejected | sip_trunk_failure \ - | connection_timeout | media_failure | agent_error | reconnect_failed | unknown" + reasons.map(DisconnectReason::as_str).collect::>().join(" | "), + "CLIENT_INITIATED | DUPLICATE_IDENTITY | SERVER_SHUTDOWN | PARTICIPANT_REMOVED \ + | ROOM_DELETED | STATE_MISMATCH | JOIN_FAILURE | MIGRATION | SIGNAL_CLOSE \ + | ROOM_CLOSED | USER_UNAVAILABLE | USER_REJECTED | SIP_TRUNK_FAILURE \ + | CONNECTION_TIMEOUT | MEDIA_FAILURE | AGENT_ERROR | RECONNECT_FAILED | UNKNOWN_REASON" ); assert_eq!(DisconnectReason::from(99), DisconnectReason::Unknown); } @@ -940,7 +968,7 @@ mod tests { #[test] fn custom_attributes_reject_sdk_keys_and_fit_replacements_at_capacity() { for key in [ - "lk.room.sid", + "lk.room_sid", "otel.event.name", "otel.scope.name", "code.function.name", diff --git a/livekit-telemetry/src/span.rs b/livekit-telemetry/src/span.rs index b4fa52e66..f252e20a0 100644 --- a/livekit-telemetry/src/span.rs +++ b/livekit-telemetry/src/span.rs @@ -15,6 +15,8 @@ use std::collections::{HashMap, VecDeque}; use std::sync::{Arc, Weak}; +use livekit_protocol::semconv::{self, lk_outcome as outcome, upstream}; + use crate::{ event::{attributes_size_hint, now_unix_nanos}, scope::ScopeState, @@ -47,9 +49,9 @@ pub enum SpanOutcome { impl SpanOutcome { pub(crate) fn as_str(self) -> &'static str { match self { - SpanOutcome::Ok => "ok", - SpanOutcome::Error => "error", - SpanOutcome::Cancelled => "cancelled", + SpanOutcome::Ok => outcome::OK, + SpanOutcome::Error => outcome::ERROR, + SpanOutcome::Cancelled => outcome::CANCELLED, } } } @@ -284,7 +286,7 @@ impl Spans { /// Its attributes are taken now, like a log record's when it is queued: `attributes`, then /// the app's correlation attributes, cut to [`MAX_ATTRIBUTES_PER_SPAN`] — the app's go first, /// then the tail of `attributes` — counted in [`SpanRecord::dropped_attributes`] and warned - /// about once per drain. The session's own (`lk.room.*`, `session.id`) and the outcome are + /// about once per drain. The session's own (`lk.room_sid`, `session.id`, …) and the outcome are /// added past the cap and always ship. pub fn end( &mut self, @@ -397,9 +399,10 @@ impl SpanRecord { /// `lk.outcome`, plus `error.type` when the span failed: the attributes every span carries /// beyond the caller's. pub(crate) fn outcome_attributes(&self) -> Vec { - let mut attributes = vec![Attribute::new("lk.outcome", self.outcome.as_str())]; + let mut attributes = vec![Attribute::new(semconv::LK_OUTCOME, self.outcome.as_str())]; if let (SpanOutcome::Error, Some(error_type)) = (self.outcome, &self.error_type) { - attributes.push(Attribute::new("error.type", AttributeValue::Str(error_type.clone()))); + let error_type = AttributeValue::Str(error_type.clone()); + attributes.push(Attribute::new(upstream::ERROR_TYPE, error_type)); } attributes } @@ -509,7 +512,7 @@ mod tests { crate::test_log::capture(); let mut spans = Spans::new(8); let session = ScopeState::new(); - session.set_attribute("lk.room.sid", Some("RM_1".into())); + session.set_attribute("lk.room_sid", Some("RM_1".into())); assert!(session.set_custom("app.call_id", Some("c1".into()))); let overflow = |spans: &mut Spans| { let id = spans.begin_in("lk.publish", SpanKind::Internal, None, session.clone()); @@ -526,7 +529,7 @@ mod tests { assert_eq!(span.dropped_attributes_count, 1, "the app's call id"); let keys: Vec<_> = span.attributes.iter().map(|kv| kv.key.as_str()).collect(); assert!(!keys.contains(&"app.call_id"), "the app's own go first"); - for key in ["lk.publish.k127", "lk.room.sid", "session.id", "lk.outcome"] { + for key in ["lk.publish.k127", "lk.room_sid", "session.id", "lk.outcome"] { assert!(keys.contains(&key), "{key} survives the cap"); } } diff --git a/livekit-telemetry/src/stats.rs b/livekit-telemetry/src/stats.rs index ae198dcf9..83e38b2c7 100644 --- a/livekit-telemetry/src/stats.rs +++ b/livekit-telemetry/src/stats.rs @@ -14,6 +14,8 @@ use std::sync::atomic::{AtomicU64, Ordering}; +use livekit_protocol::semconv::{self, event}; + use crate::TelemetryEvent; /// Declares the pipeline's health counters once: the shared atomics ([`Counters`]), their @@ -143,7 +145,7 @@ impl Snapshot { /// persisted on its own, never an extra request — plus one at shutdown, so every session /// leaves a summary the fleet's success rates can be computed from. pub fn report(&self, cached_batches: u64) -> TelemetryEvent { - let mut event = TelemetryEvent::new("lk.telemetry.report") + let mut event = TelemetryEvent::new(event::LK_TELEMETRY_REPORT) .with_body(format!( "telemetry: {} batches sent ({} B), {} failed, {} dropped, {} cached", self.uploads_sent, @@ -152,25 +154,25 @@ impl Snapshot { self.problem_drops(), cached_batches )) - .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)); + .with_attribute(semconv::LK_TELEMETRY_UPLOADS_SENT, int(self.uploads_sent)) + .with_attribute(semconv::LK_TELEMETRY_UPLOADS_BYTES, int(self.upload_bytes)) + .with_attribute(semconv::LK_TELEMETRY_UPLOADS_FAILED, int(self.upload_failures)) + .with_attribute(semconv::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), - ("lk.telemetry.holds.capped", self.hold_cap_hits), - ("lk.telemetry.cache.write_errors", self.cache_write_errors), - ("lk.telemetry.dropped.queue_full", self.queue_full), - ("lk.telemetry.dropped.cache_error", self.cache_error), - ("lk.telemetry.dropped.cache_full", self.cache_full), - ("lk.telemetry.dropped.expired", self.expired), - ("lk.telemetry.dropped.corrupt", self.corrupt), - ("lk.telemetry.dropped.invalid", self.invalid), - ("lk.telemetry.dropped.rejected", self.rejected), - ("lk.telemetry.dropped.oversized", self.oversized), - ("lk.telemetry.dropped.throttled", self.throttled), - ("lk.telemetry.dropped.rate_limited", self.rate_limited), + (semconv::LK_TELEMETRY_UPLOADS_TIMEOUTS, self.upload_timeouts), + (semconv::LK_TELEMETRY_UPLOADS_UNAUTHORIZED, self.auth_denied), + (semconv::LK_TELEMETRY_HOLDS_CAPPED, self.hold_cap_hits), + (semconv::LK_TELEMETRY_CACHE_WRITE_ERRORS, self.cache_write_errors), + (semconv::LK_TELEMETRY_DROPPED_QUEUE_FULL, self.queue_full), + (semconv::LK_TELEMETRY_DROPPED_CACHE_ERROR, self.cache_error), + (semconv::LK_TELEMETRY_DROPPED_CACHE_FULL, self.cache_full), + (semconv::LK_TELEMETRY_DROPPED_EXPIRED, self.expired), + (semconv::LK_TELEMETRY_DROPPED_CORRUPT, self.corrupt), + (semconv::LK_TELEMETRY_DROPPED_INVALID, self.invalid), + (semconv::LK_TELEMETRY_DROPPED_REJECTED, self.rejected), + (semconv::LK_TELEMETRY_DROPPED_OVERSIZED, self.oversized), + (semconv::LK_TELEMETRY_DROPPED_THROTTLED, self.throttled), + (semconv::LK_TELEMETRY_DROPPED_RATE_LIMITED, self.rate_limited), ] { if value > 0 { event = event.with_attribute(key, int(value)); diff --git a/livekit-telemetry/src/store.rs b/livekit-telemetry/src/store.rs index aafc33877..bcadecd02 100644 --- a/livekit-telemetry/src/store.rs +++ b/livekit-telemetry/src/store.rs @@ -240,7 +240,7 @@ mod tests { #[test] fn queued_bytes_count_the_sessions_attributes() { let session = ScopeState::new(); - session.set_attribute("lk.room.name", Some("r".repeat(100).into())); + session.set_attribute("lk.pii.room_name", Some("r".repeat(100).into())); let store = Store::new(10, usize::MAX, Arc::default()); store.push(QueuedEvent::new(TelemetryEvent::new("lk.ping"), session)); let bare = TelemetryEvent::new("lk.ping").size_hint(); diff --git a/livekit-telemetry/src/telemetry.rs b/livekit-telemetry/src/telemetry.rs index 1300f7352..d1be2be5e 100644 --- a/livekit-telemetry/src/telemetry.rs +++ b/livekit-telemetry/src/telemetry.rs @@ -20,6 +20,7 @@ use std::{ time::Duration, }; +use livekit_protocol::semconv::{self, upstream}; use tokio::sync::{mpsc, oneshot}; use tokio::time::{timeout, Instant}; @@ -206,7 +207,7 @@ impl TelemetryConfig { /// let room = telemetry.begin_scope(); /// room.set_server("wss://my-project.livekit.cloud", ""); /// room.emit_custom("checkout", vec![Attribute::new("plan", "pro")]); -/// telemetry.emit(TelemetryEvent::new("lk.ping")); +/// telemetry.emit(TelemetryEvent::new(livekit_protocol::semconv::event::LK_PING)); /// telemetry.shutdown().await; /// # } /// ``` @@ -737,7 +738,8 @@ impl Telemetry { /// the core sets on every record or span (`session.id`, `error.type`, `lk.outcome`), or 64 /// keys are already set. pub fn set_attribute(&self, key: &str, value: Option) { - let core_owned = matches!(key, "session.id" | "error.type" | "lk.outcome"); + let core_owned = + matches!(key, upstream::SESSION_ID | upstream::ERROR_TYPE | semconv::LK_OUTCOME); if core_owned || !within_limits(key, value.as_ref()) { return self.count_invalid(); } @@ -977,13 +979,13 @@ pub struct TelemetryResource { impl TelemetryResource { pub(crate) fn attributes(&self) -> Vec { let mut out = vec![ - Attribute::new("service.name", format!("livekit-client-{}", self.sdk.as_str())), - Attribute::new("service.version", self.sdk_version.as_str()), - Attribute::new("os.name", self.os_name.as_str()), - Attribute::new("os.version", self.os_version.as_str()), + Attribute::new(upstream::SERVICE_NAME, format!("livekit-client-{}", self.sdk.as_str())), + Attribute::new(upstream::SERVICE_VERSION, self.sdk_version.as_str()), + Attribute::new(upstream::OS_NAME, self.os_name.as_str()), + Attribute::new(upstream::OS_VERSION, self.os_version.as_str()), ]; if let Some(model) = &self.device_model { - out.push(Attribute::new("device.model.identifier", model.as_str())); + out.push(Attribute::new(upstream::DEVICE_MODEL_IDENTIFIER, model.as_str())); } out } @@ -998,10 +1000,10 @@ fn add_sdk_resource(resource: &mut Vec, sdk: Option<&TelemetryResourc } } let defaults = [ - ("service.name", "livekit-client"), - ("telemetry.sdk.name", env!("CARGO_PKG_NAME")), - ("telemetry.sdk.language", "rust"), - ("telemetry.sdk.version", env!("CARGO_PKG_VERSION")), + (upstream::SERVICE_NAME, "livekit-client"), + (upstream::TELEMETRY_SDK_NAME, env!("CARGO_PKG_NAME")), + (upstream::TELEMETRY_SDK_LANGUAGE, upstream::telemetry_sdk_language::RUST), + (upstream::TELEMETRY_SDK_VERSION, env!("CARGO_PKG_VERSION")), ]; for (key, value) in defaults { if !resource.iter().any(|a| a.key == key) { @@ -1199,7 +1201,7 @@ pub(crate) mod tests { .iter() .find(|s| { s.attributes.iter().any(|kv| { - kv.key == "lk.track.sid" + kv.key == "lk.track_sid" && kv.value.as_ref().and_then(|v| v.value.clone()) == Some(Value::StringValue(sid.into())) }) @@ -1228,7 +1230,7 @@ pub(crate) mod tests { "cancelled exports too" ); assert!( - !spans.iter().any(|s| s.attributes.iter().any(|kv| kv.key == "lk.track.sid" + !spans.iter().any(|s| s.attributes.iter().any(|kv| kv.key == "lk.track_sid" && kv.value.as_ref().and_then(|v| v.value.clone()) == Some(Value::StringValue("TR_d".into())))), "still open" @@ -1259,8 +1261,8 @@ pub(crate) mod tests { let sent = transport.sent(); let logs: Vec = sent.iter().flat_map(records).collect(); let with_room = - logs.iter().find(|r| attribute(r, "lk.room.sid").is_some()).expect("room record"); - assert_eq!(attribute(with_room, "lk.room.sid"), Some(Value::StringValue("RM_a".into()))); + logs.iter().find(|r| attribute(r, "lk.room_sid").is_some()).expect("room record"); + assert_eq!(attribute(with_room, "lk.room_sid"), Some(Value::StringValue("RM_a".into()))); assert!(logs.iter().any(|r| r.body.as_ref().and_then(|b| b.value.clone()) == Some(Value::StringValue("audio interruption began".into())))); let decoded = @@ -1942,13 +1944,13 @@ pub(crate) mod tests { sent.iter().filter(|r| r.url.ends_with("logs")).flat_map(records).collect(); assert_eq!(logs.len(), 4); assert_eq!(hex(&logs[0].trace_id), a.trace_id()); - assert_eq!(attribute(&logs[0], "lk.room.sid"), Some(Value::StringValue("RM_a".into()))); + assert_eq!(attribute(&logs[0], "lk.room_sid"), Some(Value::StringValue("RM_a".into()))); assert_eq!(attribute(&logs[0], "session.id"), Some(Value::StringValue(a.trace_id()))); assert_eq!(hex(&logs[1].trace_id), a.trace_id(), "resolved through the span"); assert_eq!(hex(&logs[2].trace_id), b.trace_id()); - assert_eq!(attribute(&logs[2], "lk.room.sid"), Some(Value::StringValue("RM_b".into()))); + assert_eq!(attribute(&logs[2], "lk.room_sid"), Some(Value::StringValue("RM_b".into()))); assert_eq!(hex(&logs[3].trace_id), telemetry.trace_id(), "device state: process session"); - assert_eq!(attribute(&logs[3], "lk.room.sid"), None); + assert_eq!(attribute(&logs[3], "lk.room_sid"), None); assert!( logs.iter() .all(|r| attribute(r, "acme.tenant") == Some(Value::StringValue("t1".into()))), @@ -1959,7 +1961,7 @@ pub(crate) mod tests { ExportTraceServiceRequest::decode(&gunzip(&traces_request.body)[..]).expect("otlp"); let otlp_span = &traces.resource_spans[0].scope_spans[0].spans[0]; assert_eq!(hex(&otlp_span.trace_id), a.trace_id()); - assert!(otlp_span.attributes.iter().any(|a| a.key == "lk.room.sid")); + assert!(otlp_span.attributes.iter().any(|a| a.key == "lk.room_sid")); // A record that names a span which has already ended — and been exported — is still that // session's: the SDK's log path hops threads, the span does not wait for it. @@ -2101,21 +2103,21 @@ pub(crate) mod tests { async fn session_attributes_are_attached_to_every_record() { let transport = FakeTransport::scripted([]); let telemetry = pipeline(transport.clone()); - telemetry.set_attribute("lk.room.sid", Some("RM_1".into())); + telemetry.set_attribute("lk.room_sid", Some("RM_1".into())); telemetry.emit(TelemetryEvent::new("lk.ping")); - telemetry.emit(TelemetryEvent::new("lk.ping").with_attribute("lk.room.sid", "RM_override")); + telemetry.emit(TelemetryEvent::new("lk.ping").with_attribute("lk.room_sid", "RM_override")); telemetry.flush().await; let first = records(&transport.sent()[0]); - assert_eq!(attribute(&first[0], "lk.room.sid"), Some(Value::StringValue("RM_1".into()))); + assert_eq!(attribute(&first[0], "lk.room_sid"), Some(Value::StringValue("RM_1".into()))); assert_eq!( - attribute(&first[1], "lk.room.sid"), + attribute(&first[1], "lk.room_sid"), Some(Value::StringValue("RM_override".into())), "an explicit attribute wins" ); - telemetry.set_attribute("lk.room.sid", None); + telemetry.set_attribute("lk.room_sid", None); telemetry.emit(TelemetryEvent::new("lk.ping")); telemetry.flush().await; - assert_eq!(attribute(&records(&transport.sent()[1])[0], "lk.room.sid"), None); + assert_eq!(attribute(&records(&transport.sent()[1])[0], "lk.room_sid"), None); } #[tokio::test(start_paused = true)] diff --git a/livekit-telemetry/src/trace.rs b/livekit-telemetry/src/trace.rs index b5d27afae..16eb5d1d6 100644 --- a/livekit-telemetry/src/trace.rs +++ b/livekit-telemetry/src/trace.rs @@ -22,9 +22,12 @@ use std::{ time::Duration, }; +use livekit_protocol::semconv::{ + self, checkpoint, lk_reconnect_mode as mode, lk_reconnect_reason as reconnect_reason, + lk_track_source as track_source, span, +}; use tokio::time::Instant; -use crate::device::snake_case; use crate::span::SpanKind; use crate::{ scope::ScopeState, telemetry::Shared, Attribute, AttributeValue, SpanOutcome, Telemetry, @@ -54,6 +57,21 @@ pub enum ReconnectReason { Unknown, } +impl ReconnectReason { + fn as_str(self) -> &'static str { + match self { + Self::SignalDisconnected => reconnect_reason::SIGNAL_DISCONNECTED, + Self::PublisherFailed => reconnect_reason::PUBLISHER_FAILED, + Self::SubscriberFailed => reconnect_reason::SUBSCRIBER_FAILED, + Self::TransportFailed => reconnect_reason::TRANSPORT_FAILED, + Self::SwitchCandidate => reconnect_reason::SWITCH_CANDIDATE, + Self::NetworkChanged => reconnect_reason::NETWORK_CHANGED, + Self::Debug => reconnect_reason::DEBUG, + Self::Unknown => reconnect_reason::UNKNOWN, + } + } +} + impl From for ReconnectReason { /// The protocol's `ReconnectReason` number (`RR_*`); a number this SDK does not know is /// `Unknown`. @@ -95,10 +113,10 @@ impl SpanName { /// The span name on the wire (`lk.connect`, …). pub fn label(&self) -> &str { match self { - Self::Connect => "lk.connect", - Self::Reconnect { .. } => "lk.reconnect", - Self::Publish => "lk.publish", - Self::Subscribe => "lk.subscribe", + Self::Connect => span::LK_CONNECT, + Self::Reconnect { .. } => span::LK_RECONNECT, + Self::Publish => span::LK_PUBLISH, + Self::Subscribe => span::LK_SUBSCRIBE, Self::Custom { name } => name, } } @@ -113,10 +131,10 @@ impl SpanName { fn attributes(&self) -> Vec { match self { Self::Reconnect { reason } => { - vec![Attribute::new("lk.reconnect.reason", snake_case(reason))] + vec![Attribute::new(semconv::LK_RECONNECT_REASON, reason.as_str())] } // The user-initiated connect is attempt 1; a platform that retries sets it again. - Self::Connect => vec![Attribute::new("lk.connect.attempt", 1i64)], + Self::Connect => vec![Attribute::new(semconv::LK_CONNECT_ATTEMPT, 1i64)], _ => Vec::new(), } } @@ -166,20 +184,19 @@ pub enum SpanStep { impl SpanStep { fn label(&self) -> String { match self { - Self::WsOpen => "ws_open".into(), - Self::Signal => "signal".into(), - Self::JoinRecv => "join_recv".into(), - Self::PcCreated => "pc_created".into(), - Self::OfferSent => "offer_sent".into(), - Self::AnswerSent => "answer_sent".into(), - Self::Engine => "engine".into(), - Self::PcConnected => "pc_connected".into(), - Self::RoomConnected => "room_connected".into(), - Self::Subscribed => "subscribed".into(), - Self::FirstMedia => "first_media".into(), - Self::Attempt { number, full } => { - format!("attempt {number} {}", if *full { "full" } else { "quick" }) - } + Self::WsOpen => checkpoint::WS_OPEN.into(), + Self::Signal => checkpoint::SIGNAL.into(), + Self::JoinRecv => checkpoint::JOIN_RECV.into(), + Self::PcCreated => checkpoint::PC_CREATED.into(), + Self::OfferSent => checkpoint::OFFER_SENT.into(), + Self::AnswerSent => checkpoint::ANSWER_SENT.into(), + Self::Engine => checkpoint::ENGINE.into(), + Self::PcConnected => checkpoint::PC_CONNECTED.into(), + Self::RoomConnected => checkpoint::ROOM_CONNECTED.into(), + Self::Subscribed => checkpoint::SUBSCRIBED.into(), + Self::FirstMedia => checkpoint::FIRST_MEDIA.into(), + // The registry's `attempt ` checkpoint pattern. + Self::Attempt { number, full } => format!("attempt {number} {}", reconnect_mode(*full)), Self::Custom { name } => name.clone(), } } @@ -187,15 +204,25 @@ impl SpanStep { fn attributes(&self) -> Vec { match self { Self::Attempt { number, full } => vec![ - Attribute::new("lk.reconnect.attempts", *number as i64), - Attribute::new("lk.reconnect.mode", if *full { "full" } else { "quick" }), + Attribute::new(semconv::LK_RECONNECT_ATTEMPTS, *number as i64), + Attribute::new(semconv::LK_RECONNECT_MODE, reconnect_mode(*full)), ], _ => Vec::new(), } } } -/// The protocol's `TrackSource`. +/// The `lk.reconnect.mode` value of an attempt. +fn reconnect_mode(full: bool) -> &'static str { + if full { + mode::FULL + } else { + mode::QUICK + } +} + +/// What produces a track. Sent as the RTC FFI `TrackSource` name (`SOURCE_CAMERA`, …), the +/// spelling Agents sends. #[cfg_attr(feature = "uniffi", derive(uniffi::Enum))] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum TrackSource { @@ -211,6 +238,18 @@ pub enum TrackSource { Unknown, } +impl TrackSource { + fn as_str(self) -> &'static str { + match self { + Self::Camera => track_source::SOURCE_CAMERA, + Self::Microphone => track_source::SOURCE_MICROPHONE, + Self::ScreenShare => track_source::SOURCE_SCREENSHARE, + Self::ScreenShareAudio => track_source::SOURCE_SCREENSHARE_AUDIO, + Self::Unknown => track_source::SOURCE_UNKNOWN, + } + } +} + /// The track a publish or subscribe span is about. #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] #[derive(Debug, Clone, PartialEq)] @@ -230,14 +269,17 @@ pub struct SpanTrack { impl SpanTrack { fn attributes(&self) -> Vec { let mut out = vec![ - Attribute::new("lk.track.kind", snake_case(self.kind)), - Attribute::new("lk.track.source", snake_case(self.source)), + Attribute::new(semconv::LK_TRACK_KIND, self.kind.as_str()), + Attribute::new(semconv::LK_TRACK_SOURCE, self.source.as_str()), ]; if let Some(sid) = &self.sid { - out.push(Attribute::new("lk.track.sid", sid.as_str())); + out.push(Attribute::new(semconv::LK_TRACK_SID, sid.as_str())); } if let Some(identity) = &self.remote_identity { - out.push(Attribute::new("lk.participant.remote_identity", identity.as_str())); + out.push(Attribute::new( + semconv::LK_PII_REMOTE_PARTICIPANT_IDENTITY, + identity.as_str(), + )); } out } @@ -589,7 +631,7 @@ mod tests { let invalid = || telemetry.shared.counters.snapshot().invalid; let room = telemetry.begin_scope(); let app = room.start(SpanName::Custom { name: "work".into() }, None); - app.set_attribute("lk.room.sid".into(), AttributeValue::Str("RM_x".into())); + app.set_attribute("lk.room_sid".into(), AttributeValue::Str("RM_x".into())); app.set_attribute("otel.scope.name".into(), AttributeValue::Str("x".into())); app.set_attribute("app.step".into(), AttributeValue::Int(1)); assert_eq!(invalid(), 2); @@ -647,8 +689,9 @@ mod tests { let attributes = span.lock().attributes.clone(); let get = |key: &str| attributes.iter().find(|a| a.key == key).map(|a| a.value.clone()); assert_eq!(get("lk.track.kind"), Some(AttributeValue::Str("video".into()))); - assert_eq!(get("lk.track.sid"), Some(AttributeValue::Str("TR_1".into()))); - assert_eq!(attributes.iter().filter(|a| a.key == "lk.track.source").count(), 1); + assert_eq!(get("lk.track_sid"), Some(AttributeValue::Str("TR_1".into()))); + assert_eq!(get("lk.track_source"), Some(AttributeValue::Str("SOURCE_CAMERA".into()))); + assert_eq!(attributes.iter().filter(|a| a.key == "lk.track_source").count(), 1); } /// What a span retains before export is bounded, however long it lives. diff --git a/livekit-telemetry/src/wire_tests.rs b/livekit-telemetry/src/wire_tests.rs new file mode 100644 index 000000000..1338c3adc --- /dev/null +++ b/livekit-telemetry/src/wire_tests.rs @@ -0,0 +1,472 @@ +// 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. + +//! Every name and enum value the core sends, frozen. The names come from the protocol's registry +//! (`livekit_protocol::semconv`), so a registry change that renames one fails here before it +//! reaches a dashboard. + +use std::{ + collections::{BTreeMap, BTreeSet}, + time::Duration, +}; + +use prost::Message; + +use crate::{ + proto::opentelemetry::proto::{ + collector::{logs::v1::ExportLogsServiceRequest, trace::v1::ExportTraceServiceRequest}, + common::v1::{any_value::Value, KeyValue}, + trace::v1::span::SpanKind, + }, + stats::Snapshot, + telemetry::tests::{gunzip, start_with_collector, test_config, FakeTransport}, + AppState, AudioOutput, AudioRouteReason, CaptureDevice, CaptureFailure, DeviceEvent, + DeviceState, DisconnectReason, LogRecord, LogSource, MemoryPressure, NetworkType, + ReconnectReason, RoomIdentity, RtcStatsSample, Scope, Sdk, Severity, SpanName, SpanOutcome, + SpanStep, SpanTrack, StreamDirection, TelemetryConfig, TelemetryResource, ThermalState, + TrackKind, TrackSource, +}; + +/// On every record and span of a session; listed once rather than under each record. +const SESSION_KEYS: &[&str] = &[ + "session.id", + "otel.event.name", + "lk.room_sid", + "lk.pii.room_name", + "lk.participant_id", + "lk.pii.participant_identity", +]; + +/// Keys whose values the core picks from a closed set: their values are pinned too. +const ENUM_KEYS: &[&str] = &[ + "service.name", + "telemetry.sdk.language", + "lk.device.thermal.state", + "lk.device.app_state", + "lk.device.memory.pressure", + "network.connection.type", + "lk.device.audio_route.reason", + "lk.device.audio_route.outputs", + "lk.device.audio.interruption", + "lk.device.capture.device", + "lk.device.capture.reason", + "lk.track.kind", + "lk.track.direction", + "lk.track_source", + "lk.reconnect.reason", + "lk.reconnect.mode", + "lk.disconnect_reason", + "lk.outcome", + "error.type", + "lk.log.source", +]; + +/// What the core sends: one line per record, span and checkpoint list, plus the values of every +/// [`ENUM_KEYS`] key. A line changes only when a name or value on the wire does. +const PINNED: &str = "\ +checkpoints lk.connect: answer_sent | engine | join_recv | offer_sent | pc_connected | pc_created \ + | room_connected | signal | ws_open +checkpoints lk.reconnect: attempt 1 quick | attempt 2 full | attempt 3 quick | attempt 4 full \ + | attempt 5 quick | attempt 6 full | attempt 7 quick | attempt 8 full +checkpoints lk.subscribe: first_media | subscribed +record -: code.file.path | code.function.name | code.line.number | lk.log.logger | lk.log.source +record lk.device.app_state.changed: lk.device.app_state +record lk.device.audio.interruption: lk.device.audio.interruption +record lk.device.audio_route.changed: lk.device.audio_route.outputs | lk.device.audio_route.reason +record lk.device.battery.changed: lk.device.battery.charging | lk.device.battery.level +record lk.device.capture.failed: lk.device.capture.device | lk.device.capture.reason +record lk.device.low_power.changed: lk.device.low_power.enabled +record lk.device.memory.changed: lk.device.memory.pressure +record lk.device.network.changed: lk.device.network.constrained | lk.device.network.expensive \ + | network.connection.type +record lk.device.thermal.changed: lk.device.thermal.state +record lk.room.disconnected: lk.disconnect_reason +record lk.rtc.stats.sample: lk.rtc.audio_level.avg | lk.rtc.audio_level.max \ + | lk.rtc.audio_level.min | lk.rtc.bytes | lk.rtc.codec | lk.rtc.concealed_samples \ + | lk.rtc.concealment_events | lk.rtc.fps.avg | lk.rtc.fps.max | lk.rtc.fps.min \ + | lk.rtc.freeze_count | lk.rtc.freezes_duration_ms | lk.rtc.interruption_count \ + | lk.rtc.interruptions_duration_ms | lk.rtc.jitter_buffer_delay_ms \ + | lk.rtc.jitter_buffer_emitted_count | lk.rtc.jitter_ms.avg | lk.rtc.jitter_ms.max \ + | lk.rtc.jitter_ms.min | lk.rtc.packets | lk.rtc.packets_lost | lk.rtc.pause_count \ + | lk.rtc.pauses_duration_ms | lk.rtc.quality_limitation.bandwidth_ms \ + | lk.rtc.quality_limitation.cpu_ms | lk.rtc.quality_limitation.other_ms | lk.rtc.rtt_ms.avg \ + | lk.rtc.rtt_ms.max | lk.rtc.rtt_ms.min | lk.rtc.samples | lk.rtc.silent_concealed_samples \ + | lk.rtc.window_ms | lk.track.direction | lk.track.kind | lk.track_sid +record lk.telemetry.report: lk.telemetry.cache.batches | lk.telemetry.cache.write_errors \ + | lk.telemetry.dropped.cache_error | lk.telemetry.dropped.cache_full \ + | lk.telemetry.dropped.corrupt | lk.telemetry.dropped.expired | lk.telemetry.dropped.invalid \ + | lk.telemetry.dropped.oversized | lk.telemetry.dropped.queue_full \ + | lk.telemetry.dropped.rate_limited | lk.telemetry.dropped.rejected \ + | lk.telemetry.dropped.throttled | lk.telemetry.holds.capped | lk.telemetry.uploads.bytes \ + | lk.telemetry.uploads.failed | lk.telemetry.uploads.sent | lk.telemetry.uploads.timeouts \ + | lk.telemetry.uploads.unauthorized +resource: device.model.identifier | os.name | os.version | service.name | service.version \ + | telemetry.sdk.language | telemetry.sdk.name | telemetry.sdk.version +session: lk.participant_id | lk.pii.participant_identity | lk.pii.room_name | lk.room_sid \ + | otel.event.name | session.id +severity -: ERROR +severity custom.checkout: INFO +severity lk.device.app_state.changed: INFO +severity lk.device.audio.interruption: INFO +severity lk.device.audio_route.changed: INFO +severity lk.device.battery.changed: INFO +severity lk.device.capture.failed: WARN +severity lk.device.low_power.changed: INFO +severity lk.device.memory.changed: INFO +severity lk.device.network.changed: INFO +severity lk.device.thermal.changed: INFO +severity lk.room.disconnected: INFO | WARN +severity lk.rtc.stats.sample: INFO +severity lk.telemetry.report: INFO +span lk.connect SPAN_KIND_CLIENT: lk.connect.attempt | lk.outcome +span lk.publish SPAN_KIND_INTERNAL: lk.outcome | lk.track.kind | lk.track_sid | lk.track_source +span lk.reconnect SPAN_KIND_CLIENT: lk.outcome | lk.reconnect.attempts | lk.reconnect.mode \ + | lk.reconnect.reason +span lk.subscribe SPAN_KIND_INTERNAL: error.type | lk.outcome | lk.pii.remote_participant_identity \ + | lk.track.kind | lk.track_sid | lk.track_source +value error.type: timed_out +value lk.device.app_state: background | foreground +value lk.device.audio.interruption: began | ended +value lk.device.audio_route.outputs: air_play | bluetooth | car_audio | hdmi | other | receiver \ + | speaker | usb | wired_headset +value lk.device.audio_route.reason: category_change | new_device | no_suitable_route \ + | old_device_unavailable | override | route_configuration_change | unknown | wake_from_sleep +value lk.device.capture.device: camera | microphone | screen_share +value lk.device.capture.reason: disconnected | in_use | not_found | other | permission_denied +value lk.device.memory.pressure: critical | normal | warning +value lk.device.thermal.state: critical | fair | nominal | serious +value lk.disconnect_reason: AGENT_ERROR | CLIENT_INITIATED | CONNECTION_TIMEOUT \ + | DUPLICATE_IDENTITY | JOIN_FAILURE | MEDIA_FAILURE | MIGRATION | PARTICIPANT_REMOVED \ + | RECONNECT_FAILED | ROOM_CLOSED | ROOM_DELETED | SERVER_SHUTDOWN | SIGNAL_CLOSE \ + | SIP_TRUNK_FAILURE | STATE_MISMATCH | UNKNOWN_REASON | USER_REJECTED | USER_UNAVAILABLE +value lk.log.source: ffi | sdk | webrtc +value lk.outcome: cancelled | error | ok +value lk.reconnect.mode: full | quick +value lk.reconnect.reason: debug | network_changed | publisher_failed | signal_disconnected \ + | subscriber_failed | switch_candidate | transport_failed | unknown +value lk.track.direction: inbound | outbound +value lk.track.kind: audio | video +value lk.track_source: SOURCE_CAMERA | SOURCE_MICROPHONE | SOURCE_SCREENSHARE \ + | SOURCE_SCREENSHARE_AUDIO | SOURCE_UNKNOWN +value network.connection.type: bluetooth | cell | other | unavailable | unknown | vpn | wifi \ + | wired +value service.name: livekit-client-swift +value telemetry.sdk.language: rust +"; + +#[tokio::test(start_paused = true)] +async fn every_name_and_value_on_the_wire_is_pinned() { + let transport = FakeTransport::scripted([]); + let config = TelemetryConfig { + sdk: Some(TelemetryResource { + sdk: Sdk::Swift, + sdk_version: "2.17.0".into(), + os_name: "iOS".into(), + os_version: "18.5".into(), + device_model: Some("iPhone17,1".into()), + }), + // Every variant below is a discrete event: none may meet the flood guard. + max_events_per_10min: 10_000, + ..test_config() + }; + let telemetry = start_with_collector(config, transport.clone()); + + device_states(&telemetry); + for reason in [ + AudioRouteReason::NewDevice, + AudioRouteReason::OldDeviceUnavailable, + AudioRouteReason::CategoryChange, + AudioRouteReason::Override, + AudioRouteReason::WakeFromSleep, + AudioRouteReason::NoSuitableRoute, + AudioRouteReason::RouteConfigurationChange, + AudioRouteReason::Unknown, + ] { + use AudioOutput::*; + let outputs = + vec![Speaker, Receiver, WiredHeadset, Bluetooth, CarAudio, AirPlay, Hdmi, Usb, Other]; + telemetry.device_event(DeviceEvent::AudioRouteChanged { outputs, reason }); + } + for began in [true, false] { + telemetry.device_event(DeviceEvent::AudioInterruption { began }); + } + for (device, reason) in [ + (CaptureDevice::Camera, CaptureFailure::PermissionDenied), + (CaptureDevice::Microphone, CaptureFailure::NotFound), + (CaptureDevice::ScreenShare, CaptureFailure::InUse), + (CaptureDevice::Camera, CaptureFailure::Disconnected), + (CaptureDevice::Camera, CaptureFailure::Other), + ] { + telemetry.device_event(DeviceEvent::CaptureFailed { device, reason }); + } + for source in [LogSource::Sdk, LogSource::Ffi, LogSource::WebRtc] { + telemetry.log(LogRecord { + severity: Severity::Error, + source, + body: "failed".into(), + logger: Some("Room".into()), + function: Some("connect()".into()), + file: Some("Room.swift".into()), + line: Some(42), + timestamp_ns: None, + span_id: None, + }); + } + let every_counter = Snapshot { + queue_full: 1, + cache_error: 1, + cache_full: 1, + expired: 1, + corrupt: 1, + invalid: 1, + rejected: 1, + oversized: 1, + throttled: 1, + disabled: 1, + rate_limited: 1, + purged: 1, + uploads_sent: 1, + upload_bytes: 1, + upload_failures: 1, + upload_timeouts: 1, + auth_denied: 1, + hold_cap_hits: 1, + cache_write_errors: 1, + }; + telemetry.emit(every_counter.report(1)); + + let room = telemetry.begin_scope(); + room.set_room(RoomIdentity { + sid: Some("RM_1".into()), + name: Some("standup".into()), + participant_sid: Some("PA_1".into()), + participant_identity: Some("alice".into()), + }); + room.emit_custom("checkout", Vec::new()); + spans(&room).await; + stats(&room); + telemetry.flush().await; + for reason in (0..=16).map(DisconnectReason::from).chain([DisconnectReason::ReconnectFailed]) { + telemetry.begin_scope().disconnected(reason); + } + telemetry.shutdown().await; + + assert_eq!(wire(&transport), PINNED); +} + +/// Every value of every device state field, ending at rest so nothing holds the uploads. +fn device_states(telemetry: &crate::Telemetry) { + let mut state = DeviceState { + low_power_mode: Some(true), + battery_level: Some(50), + network_expensive: true, + ..Default::default() + }; + telemetry.set_device_state(state); + for thermal in [ThermalState::Fair, ThermalState::Serious, ThermalState::Critical] { + telemetry.set_device_state(DeviceState { thermal, ..state }); + } + for memory in [MemoryPressure::Warning, MemoryPressure::Critical] { + telemetry.set_device_state(DeviceState { memory, ..state }); + } + for network in [ + NetworkType::Unknown, + NetworkType::Wifi, + NetworkType::Cell, + NetworkType::Wired, + NetworkType::Vpn, + NetworkType::Bluetooth, + NetworkType::Other, + NetworkType::Unavailable, + ] { + telemetry.set_device_state(DeviceState { network, ..state }); + } + state = DeviceState { + thermal: ThermalState::Nominal, + low_power_mode: Some(false), + app_state: AppState::Background, + memory: MemoryPressure::Normal, + network: NetworkType::Wifi, + network_expensive: false, + network_constrained: false, + battery_level: Some(50), + battery_charging: true, + }; + telemetry.set_device_state(state); + telemetry.set_device_state(DeviceState { app_state: AppState::Foreground, ..state }); +} + +/// Every typed span, checkpoint, reconnect reason, track source and outcome. +async fn spans(room: &Scope) { + let connect = room.start(SpanName::Connect, None); + for step in [ + SpanStep::WsOpen, + SpanStep::Signal, + SpanStep::JoinRecv, + SpanStep::PcCreated, + SpanStep::OfferSent, + SpanStep::AnswerSent, + SpanStep::Engine, + SpanStep::PcConnected, + SpanStep::RoomConnected, + ] { + connect.step(step); + } + connect.end(SpanOutcome::Ok, None); + for (n, reason) in [ + ReconnectReason::SignalDisconnected, + ReconnectReason::PublisherFailed, + ReconnectReason::SubscriberFailed, + ReconnectReason::TransportFailed, + ReconnectReason::SwitchCandidate, + ReconnectReason::NetworkChanged, + ReconnectReason::Debug, + ReconnectReason::Unknown, + ] + .into_iter() + .enumerate() + { + let reconnect = room.start(SpanName::Reconnect { reason }, None); + reconnect.step(SpanStep::Attempt { number: n as u32 + 1, full: n % 2 == 1 }); + reconnect.end(SpanOutcome::Ok, None); + } + let track = |sid: &str, kind, source| SpanTrack { + sid: Some(sid.into()), + kind, + source, + remote_identity: Some("bob".into()), + }; + for (sid, source) in [ + ("TR_cam", TrackSource::Camera), + ("TR_mic", TrackSource::Microphone), + ("TR_screen", TrackSource::ScreenShare), + ("TR_screen_audio", TrackSource::ScreenShareAudio), + ("TR_other", TrackSource::Unknown), + ] { + let publish = room.start(SpanName::Publish, None); + publish + .set_track(SpanTrack { remote_identity: None, ..track(sid, TrackKind::Audio, source) }); + publish.end(SpanOutcome::Ok, None); + } + // Ok at first media, cancelled when the track ends first, timed out without media. + room.subscribe_started(track("TR_ok", TrackKind::Video, TrackSource::Camera)); + room.subscribed(track("TR_ok", TrackKind::Video, TrackSource::Camera)); + let mut media = RtcStatsSample::new("TR_ok", TrackKind::Video, StreamDirection::Inbound); + media.bytes = Some(1_500); + room.record_stats(media); + room.subscribe_started(track("TR_gone", TrackKind::Audio, TrackSource::Microphone)); + room.track_ended("TR_gone"); + room.subscribe_started(track("TR_silent", TrackKind::Audio, TrackSource::Microphone)); + tokio::time::sleep(Scope::SUBSCRIBE_TIMEOUT + Duration::from_secs(1)).await; +} + +/// One window per kind and direction, every counter and gauge set. +fn stats(room: &Scope) { + for (sid, kind, direction) in [ + ("TR_av_in", TrackKind::Audio, StreamDirection::Inbound), + ("TR_av_out", TrackKind::Audio, StreamDirection::Outbound), + ("TR_vv_in", TrackKind::Video, StreamDirection::Inbound), + ("TR_vv_out", TrackKind::Video, StreamDirection::Outbound), + ] { + room.record_stats(RtcStatsSample { + codec: Some("opus".into()), + bytes: Some(1), + packets: Some(1), + packets_lost: Some(1), + freeze_count: Some(1), + freezes_duration_ms: Some(1), + concealed_samples: Some(1), + concealment_events: Some(1), + jitter_buffer_delay_ms: Some(1), + jitter_buffer_emitted_count: Some(1), + quality_limitation_bandwidth_ms: Some(1), + quality_limitation_cpu_ms: Some(1), + quality_limitation_other_ms: Some(1), + pause_count: Some(1), + pauses_duration_ms: Some(1), + silent_concealed_samples: Some(1), + interruption_count: Some(1), + interruptions_duration_ms: Some(1), + jitter_ms: Some(1.0), + rtt_ms: Some(1.0), + frames_per_second: Some(1.0), + audio_level: Some(1.0), + ..RtcStatsSample::new(sid, kind, direction) + }); + } +} + +type Lines = BTreeMap>; + +/// What went out, one line per signal: its keys, severities, checkpoints, and the values of +/// [`ENUM_KEYS`]. +fn wire(transport: &FakeTransport) -> String { + let mut lines = Lines::new(); + for request in transport.sent() { + let body = gunzip(&request.body); + if request.url.contains("traces") { + let decoded = ExportTraceServiceRequest::decode(&body[..]).expect("otlp traces"); + for resource_spans in decoded.resource_spans { + let resource = resource_spans.resource.unwrap_or_default(); + add_attributes(&mut lines, "resource", &resource.attributes); + for span in resource_spans.scope_spans.into_iter().flat_map(|s| s.spans) { + let kind = SpanKind::try_from(span.kind).expect("kind").as_str_name(); + let line = format!("span {} {kind}", span.name); + add_attributes(&mut lines, &line, &span.attributes); + for event in &span.events { + add(&mut lines, format!("checkpoints {}", span.name), &event.name); + } + } + } + } else { + let decoded = ExportLogsServiceRequest::decode(&body[..]).expect("otlp logs"); + for resource_logs in decoded.resource_logs { + let resource = resource_logs.resource.unwrap_or_default(); + add_attributes(&mut lines, "resource", &resource.attributes); + for record in resource_logs.scope_logs.into_iter().flat_map(|s| s.log_records) { + let name = Some(record.event_name.as_str()).filter(|n| !n.is_empty()); + let name = name.unwrap_or("-"); + add_attributes(&mut lines, &format!("record {name}"), &record.attributes); + add(&mut lines, format!("severity {name}"), &record.severity_text); + } + } + } + } + lines + .into_iter() + .map(|(line, items)| format!("{line}: {}\n", Vec::from_iter(items).join(" | "))) + .collect() +} + +fn add(lines: &mut Lines, line: String, item: &str) { + lines.entry(line).or_default().insert(item.to_owned()); +} + +/// The keys under `line` (a session's under `session`), and the values of [`ENUM_KEYS`]. +fn add_attributes(lines: &mut Lines, line: &str, attributes: &[KeyValue]) { + for kv in attributes { + let session = SESSION_KEYS.contains(&kv.key.as_str()); + add(lines, if session { "session".into() } else { line.into() }, &kv.key); + let Some(Value::StringValue(value)) = kv.value.as_ref().and_then(|v| v.value.as_ref()) + else { + continue; + }; + if ENUM_KEYS.contains(&kv.key.as_str()) { + // The audio route's outputs are one comma-separated value. + for value in value.split(',') { + add(lines, format!("value {}", kv.key), value); + } + } + } +}