diff --git a/crates/proton-sdk/src/telemetry.rs b/crates/proton-sdk/src/telemetry.rs new file mode 100644 index 0000000..a354a7b --- /dev/null +++ b/crates/proton-sdk/src/telemetry.rs @@ -0,0 +1,267 @@ +//! Pluggable telemetry / structured-observability hooks. +//! +//! The SDK records a [`TelemetryEvent`] for each instrumented operation +//! (transfers, navigation, API calls) and hands it to a consumer-supplied +//! [`Telemetry`] observer. This mirrors the C# `Proton.Sdk` telemetry layer's +//! `ITelemetry`-style sink: the SDK measures, the host decides what to do with +//! the measurements (export to metrics, log, drop). +//! +//! Observers are **fire-and-forget and must not block**: [`Telemetry::record`] +//! is called on the SDK's own task, synchronously, often inside a hot path. Do +//! cheap work (increment a counter, enqueue) and return; offload anything +//! heavy. +//! +//! Two built-ins are provided: [`NoopTelemetry`] (the default — records +//! nothing) and [`TracingTelemetry`] (forwards every event to the `tracing` +//! crate, tying telemetry into the SDK's existing structured logging). + +use std::sync::Arc; +use std::time::Duration; + +/// The outcome of an instrumented operation. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Outcome { + /// The operation completed successfully. + Success, + /// The operation returned an error (or was dropped before being marked + /// successful — see [`OpTimer`]). + Failure, +} + +impl Outcome { + /// `"success"` / `"failure"` — the label form used by [`TracingTelemetry`] + /// and convenient for metric dimensions. + pub fn as_str(self) -> &'static str { + match self { + Outcome::Success => "success", + Outcome::Failure => "failure", + } + } +} + +/// A single measurement emitted by the SDK for one instrumented operation. +/// +/// `operation` is a stable, low-cardinality identifier (e.g. `"download_file"`, +/// `"upload_file"`, `"enumerate_folder_children"`) safe to use as a metric name +/// or dimension. `attributes` carry extra low-cardinality context (e.g. +/// `("block_count", "12")`); avoid per-request unique values (ids, names) that +/// would explode a metrics backend's cardinality. +#[derive(Debug, Clone)] +pub struct TelemetryEvent { + /// Stable operation identifier. + pub operation: &'static str, + /// Whether the operation succeeded. + pub outcome: Outcome, + /// Wall-clock time the operation took. + pub duration: Duration, + /// Extra low-cardinality key/value context. + pub attributes: Vec<(&'static str, String)>, +} + +/// A consumer-supplied sink for [`TelemetryEvent`]s. +/// +/// Mirrors the C# telemetry sink. Implementations must be cheap to share across +/// tasks (hence [`Send`] + [`Sync`]); the SDK holds them as +/// `Arc`. [`record`](Telemetry::record) must not block. +pub trait Telemetry: Send + Sync { + /// Receive one measurement. Called synchronously on the SDK's task. + fn record(&self, event: &TelemetryEvent); +} + +/// The default observer: records nothing. +#[derive(Debug, Clone, Copy, Default)] +pub struct NoopTelemetry; + +impl Telemetry for NoopTelemetry { + fn record(&self, _event: &TelemetryEvent) {} +} + +impl NoopTelemetry { + /// A shared no-op observer, for use as the default telemetry sink. + pub fn shared() -> Arc { + Arc::new(NoopTelemetry) + } +} + +/// An observer that forwards every event to the `tracing` crate. +/// +/// Successful operations emit at `debug`, failures at `warn`, both on the +/// `proton_sdk::telemetry` target with structured `operation`, `outcome`, +/// `duration_ms` and the event's attributes flattened into the message. This +/// bridges telemetry into the SDK's existing `tracing`-based logging without a +/// separate metrics backend. +#[derive(Debug, Clone, Copy, Default)] +pub struct TracingTelemetry; + +impl TracingTelemetry { + /// A shared `tracing`-forwarding observer. + pub fn shared() -> Arc { + Arc::new(TracingTelemetry) + } +} + +impl Telemetry for TracingTelemetry { + fn record(&self, event: &TelemetryEvent) { + let duration_ms = event.duration.as_secs_f64() * 1000.0; + // Attributes are flattened into a single string so the log line stays + // readable regardless of how many there are. + let attrs = event + .attributes + .iter() + .map(|(k, v)| format!("{k}={v}")) + .collect::>() + .join(" "); + match event.outcome { + Outcome::Success => tracing::debug!( + target: "proton_sdk::telemetry", + operation = event.operation, + outcome = event.outcome.as_str(), + duration_ms, + attributes = %attrs, + "operation completed", + ), + Outcome::Failure => tracing::warn!( + target: "proton_sdk::telemetry", + operation = event.operation, + outcome = event.outcome.as_str(), + duration_ms, + attributes = %attrs, + "operation failed", + ), + } + } +} + +/// A scoped timer that records a [`TelemetryEvent`] when dropped. +/// +/// Construct one at the start of an operation via [`Telemetry`]'s +/// [`start`](TelemetryExt::start) helper. It defaults to [`Outcome::Failure`] +/// so an early `?`-return is recorded as a failure automatically; call +/// [`success`](OpTimer::success) on the happy path before returning. Attach +/// extra context with [`attr`](OpTimer::attr). +/// +/// ```ignore +/// let mut timer = telemetry.start("download_file"); +/// let blocks = fetch_blocks().await?; // early return => Failure +/// timer.attr("block_count", blocks.len()); +/// timer.success(); // mark before Ok +/// Ok(data) +/// ``` +pub struct OpTimer { + telemetry: Arc, + operation: &'static str, + start: std::time::Instant, + outcome: Outcome, + attributes: Vec<(&'static str, String)>, +} + +impl OpTimer { + /// Mark the operation successful. Without this, drop records a failure. + pub fn success(&mut self) { + self.outcome = Outcome::Success; + } + + /// Explicitly set the outcome (rarely needed; prefer [`success`](Self::success)). + pub fn set_outcome(&mut self, outcome: Outcome) { + self.outcome = outcome; + } + + /// Attach a low-cardinality attribute to the event recorded on drop. + pub fn attr(&mut self, key: &'static str, value: impl ToString) { + self.attributes.push((key, value.to_string())); + } +} + +impl Drop for OpTimer { + fn drop(&mut self) { + let event = TelemetryEvent { + operation: self.operation, + outcome: self.outcome, + duration: self.start.elapsed(), + // Take the attributes out so we don't clone; `self` is being dropped. + attributes: std::mem::take(&mut self.attributes), + }; + self.telemetry.record(&event); + } +} + +/// Ergonomic entry point for instrumenting an operation against a shared +/// telemetry sink. +pub trait TelemetryExt { + /// Begin timing `operation`, returning an [`OpTimer`] that records on drop. + fn start(&self, operation: &'static str) -> OpTimer; +} + +impl TelemetryExt for Arc { + fn start(&self, operation: &'static str) -> OpTimer { + OpTimer { + telemetry: Arc::clone(self), + operation, + start: std::time::Instant::now(), + outcome: Outcome::Failure, + attributes: Vec::new(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Mutex; + + /// A test sink that captures every recorded event. + #[derive(Default)] + struct Capture(Mutex>); + + impl Telemetry for Capture { + fn record(&self, event: &TelemetryEvent) { + self.0.lock().unwrap().push(event.clone()); + } + } + + fn shared_capture() -> (Arc, Arc) { + let capture = Arc::new(Capture::default()); + (capture.clone() as Arc, capture) + } + + #[test] + fn timer_records_failure_by_default() { + let (sink, capture) = shared_capture(); + { + let _timer = sink.start("op_a"); + // dropped without success() + } + let events = capture.0.lock().unwrap(); + assert_eq!(events.len(), 1); + assert_eq!(events[0].operation, "op_a"); + assert_eq!(events[0].outcome, Outcome::Failure); + } + + #[test] + fn timer_records_success_and_attributes() { + let (sink, capture) = shared_capture(); + { + let mut timer = sink.start("op_b"); + timer.attr("block_count", 3usize); + timer.success(); + } + let events = capture.0.lock().unwrap(); + assert_eq!(events.len(), 1); + assert_eq!(events[0].outcome, Outcome::Success); + assert_eq!(events[0].attributes, vec![("block_count", "3".to_string())]); + } + + #[test] + fn noop_records_nothing() { + let sink = NoopTelemetry::shared(); + let mut timer = sink.start("op_c"); + timer.success(); + // No panic, nothing to assert — the point is it compiles and runs. + } + + #[test] + fn outcome_label_form() { + assert_eq!(Outcome::Success.as_str(), "success"); + assert_eq!(Outcome::Failure.as_str(), "failure"); + } +}