diff --git a/.sampo/changesets/ancient-iceseeker-ahti.md b/.sampo/changesets/ancient-iceseeker-ahti.md new file mode 100644 index 00000000..797ddaf1 --- /dev/null +++ b/.sampo/changesets/ancient-iceseeker-ahti.md @@ -0,0 +1,5 @@ +--- +cargo/posthog-rs: minor +--- + +Add tracing-subscriber integration for capturing tracing events as PostHog events diff --git a/Cargo.toml b/Cargo.toml index 2456050b..45412960 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -27,6 +27,7 @@ os_info = "3.14" sha1 = "0.11" regex = "1.10" tracing = "0.1" +tracing-subscriber = { version = "0.3", optional = true, default-features = false, features = ["registry", "std"] } tokio = { version = "1", features = [ "rt", "sync", @@ -54,6 +55,7 @@ async-client = ["tokio"] capture-v1 = ["brotli", "zstd"] test-harness = [] error-tracking = ["dep:backtrace", "dep:findshlibs"] +tracing-subscriber = ["dep:tracing-subscriber"] [workspace] members = [".", "compliance/adapter"] diff --git a/examples/README.md b/examples/README.md index 16add4a3..a0a38515 100644 --- a/examples/README.md +++ b/examples/README.md @@ -18,6 +18,7 @@ cargo run --example feature_flags --features async-client ``` Shows: + - Boolean feature flag checks - A/B test variants (multivariate flags) - Property-based targeting @@ -35,6 +36,7 @@ cargo run --example local_evaluation --features async-client ``` Shows: + - Performance comparison (API vs local evaluation) - Setting up local evaluation with polling - Batch flag evaluation @@ -50,6 +52,7 @@ cargo run --example advanced_config --features async-client ``` Shows: + - Basic client setup (US region) - EU region configuration (GDPR compliance) - Self-hosted instance configuration @@ -67,6 +70,7 @@ cargo run --example error_tracking ``` Shows: + - Capturing a Rust error as a PostHog Error Tracking event - Attaching a distinct ID - Adding custom exception properties @@ -123,11 +127,30 @@ Custom HTTP clients are separate from TLS crypto-provider selection. The `tls-no-provider` feature proposed in #245 is on `v1`, not this `main` branch; changing the provider does not change the reqwest client types. +### 6. Tracing + +Captures selected `tracing` events as PostHog events: + +```bash +cargo run --example tracing --features tracing-subscriber +``` + +Shows: + +- Capturing tracing events as PostHog events +- Filtering events with `tracing-subscriber` +- Configuring event names, identity, and shared properties +- Flushing events before shutdown + +The `tracing-subscriber` feature is optional. The integration supports both +async and blocking clients. + ## Key Concepts ### Feature Flag Types 1. **Boolean Flags**: Simple on/off toggles + ```rust FlagValue::Boolean(true) // enabled FlagValue::Boolean(false) // disabled diff --git a/src/lib.rs b/src/lib.rs index 8e898a42..fdcafbe4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -100,6 +100,8 @@ mod feature_flag_evaluations; mod feature_flags; mod global; mod local_evaluation; +#[cfg(feature = "tracing-subscriber")] +mod tracing_layer; // Public interface - any change to this is breaking! // Client @@ -112,6 +114,8 @@ pub use client::ClientOptions; pub use client::ClientOptionsBuilder; pub use client::ClientOptionsBuilderError; pub use client::{CaptureFailure, FlagsFailure, LocalEvaluationFailure, PostHogError}; +#[cfg(feature = "tracing-subscriber")] +pub use tracing_layer::{EventNamer, PostHogLayer}; // Endpoints pub use endpoints::{ diff --git a/src/tracing_layer.rs b/src/tracing_layer.rs new file mode 100644 index 00000000..ad8cedaf --- /dev/null +++ b/src/tracing_layer.rs @@ -0,0 +1,346 @@ +use crate::{Client, Event}; +use serde_json::{Map, Value}; +use std::{cell::Cell, fmt::Debug, sync::Arc}; +use tracing::{ + field::{Field, Visit}, + Event as TracingEvent, Metadata, Subscriber, +}; +use tracing_subscriber::layer::{Context, Layer}; + +thread_local! { + static IS_CAPTURING: Cell = const { Cell::new(false) }; +} + +struct CapturingGuard; + +impl CapturingGuard { + #[inline] + fn try_acquire() -> Option { + IS_CAPTURING.with(|capturing| { + if capturing.get() { + None + } else { + capturing.set(true); + Some(Self) + } + }) + } +} + +impl Drop for CapturingGuard { + #[inline] + fn drop(&mut self) { + IS_CAPTURING.with(|capturing| capturing.set(false)); + } +} + +#[derive(Default)] +struct Visitor { + fields: Map, +} + +impl Visit for Visitor { + #[inline] + fn record_i64(&mut self, field: &Field, value: i64) { + self.fields.insert(field.name().to_string(), value.into()); + } + + #[inline] + fn record_u64(&mut self, field: &Field, value: u64) { + self.fields.insert(field.name().to_string(), value.into()); + } + + #[inline] + fn record_i128(&mut self, field: &Field, value: i128) { + let val = serde_json::Number::from_f64(value as f64) + .map(Value::Number) + .unwrap_or_else(|| Value::String(value.to_string())); + self.fields.insert(field.name().to_string(), val); + } + + #[inline] + fn record_u128(&mut self, field: &Field, value: u128) { + let val = serde_json::Number::from_f64(value as f64) + .map(Value::Number) + .unwrap_or_else(|| Value::String(value.to_string())); + self.fields.insert(field.name().to_string(), val); + } + + #[inline] + fn record_bool(&mut self, field: &Field, value: bool) { + self.fields.insert(field.name().to_string(), value.into()); + } + + #[inline] + fn record_f64(&mut self, field: &Field, value: f64) { + if let Some(num) = serde_json::Number::from_f64(value) { + self.fields + .insert(field.name().to_string(), Value::Number(num)); + } else { + self.fields.insert(field.name().to_string(), Value::Null); + } + } + + #[inline] + fn record_str(&mut self, field: &Field, value: &str) { + self.fields + .insert(field.name().to_string(), Value::String(value.to_string())); + } + + #[inline] + fn record_bytes(&mut self, field: &Field, value: &[u8]) { + self.fields.insert( + field.name().to_string(), + Value::String(String::from_utf8_lossy(value).into_owned()), + ); + } + + #[inline] + fn record_error(&mut self, field: &Field, value: &(dyn std::error::Error + 'static)) { + self.fields + .insert(field.name().to_string(), Value::String(value.to_string())); + } + + fn record_debug(&mut self, field: &Field, value: &dyn Debug) { + self.fields.insert( + field.name().to_string(), + Value::String(format!("{value:?}")), + ); + } +} + +/// Controls how tracing metadata is turned into a PostHog event name. +pub enum EventNamer { + /// Use the tracing target as the event name. + Target, + /// Use the tracing callsite name as the event name. + Name, + /// Combine the tracing target and callsite name. + TargetAndName, + /// Derive the event name from the tracing metadata. + Custom(Arc) -> String + Send + Sync>), +} + +impl Default for EventNamer { + fn default() -> Self { + Self::Target + } +} + +impl EventNamer { + fn name(&self, metadata: &Metadata<'_>) -> String { + match self { + Self::Target => metadata.target().to_string(), + Self::Name => metadata.name().to_string(), + Self::TargetAndName => format!("{}.{}", metadata.target(), metadata.name()), + Self::Custom(namer) => namer(metadata), + } + } +} + +pub enum DistinctIdSource { + Field, + Static(String), + Provider(Arc Option + Send + Sync>), + Anonymous, +} + +impl Default for DistinctIdSource { + fn default() -> Self { + Self::Field + } +} + +impl DistinctIdSource { + fn resolve(&self, field: Option) -> Option { + match self { + Self::Field => field.and_then(|value| match value { + Value::String(s) if !s.is_empty() => Some(s), + Value::Null => None, + value => Some(value.to_string()), + }), + Self::Static(value) => Some(value.clone()), + Self::Provider(provider) => provider(), + Self::Anonymous => None, + } + } +} + +/// Captures selected `tracing` events as PostHog events. +/// +/// Tracing fields are stored as PostHog properties, while the existing +/// PostHog client handles batching and delivery. +/// +/// Use `tracing_subscriber` filters to control which events are captured. +/// +/// # Example +/// +/// ```no_run +/// use std::sync::Arc; +/// use tracing_subscriber::layer::SubscriberExt; +/// +/// let layer = posthog_rs::PostHogLayer::new(Arc::new(client)); +/// +/// tracing_subscriber::registry() +/// .with(layer) +/// .init(); +/// +/// tracing::info!( +/// target: "posthog", +/// distinct_id = "user-123", +/// feature = "checkout", +/// "checkout started", +/// ); +/// ``` +/// +/// SDK internal `posthog_rs` targets are ignored automatically. +/// +/// For short lived progrms, flush the client before exiting. +/// +/// ```no_run +/// client.flush(); +/// ``` +/// +/// With the async client: +/// +/// ```no_run +/// client.flush().await; +/// ``` +pub struct PostHogLayer { + client: Arc, + event_namer: EventNamer, + distinct_id: DistinctIdSource, + properties: Map, +} + +impl PostHogLayer { + /// Creates a layer backed by the given PostHog client. + /// + /// By default, the tracing target is used as the PostHog event name and + /// `distinct_id` is read from the tracing fields. + pub fn new(client: Arc) -> Self { + Self { + client, + event_namer: EventNamer::Target, + distinct_id: DistinctIdSource::Field, + properties: Map::new(), + } + } + + /// Sets how tracing events are named in PostHog. + /// + /// # Example + /// + /// ```no_run + /// let layer = PostHogLayer::new(client) + /// .with_event_namer(EventNamer::TargetAndName); + /// ``` + pub fn with_event_namer(mut self, event_namer: EventNamer) -> Self { + self.event_namer = event_namer; + self + } + + /// Sets a fixed distinct ID for all captured events. + /// + /// # Example + /// + /// ```no_run + /// let layer = PostHogLayer::new(client) + /// .with_distinct_id("user-123"); + /// ``` + pub fn with_distinct_id(mut self, distinct_id: impl Into) -> Self { + self.distinct_id = DistinctIdSource::Static(distinct_id.into()); + self + } + + /// Gets the distinct ID for each captured event from a callback. + /// + /// Return `None` to create an anonymous PostHog event. + /// + /// # Example + /// + /// ```no_run + /// let layer = PostHogLayer::new(client) + /// .with_distinct_id_provider(|| { + /// Some("user-123".to_owned()) + /// }); + /// ``` + pub fn with_distinct_id_provider(mut self, provider: F) -> Self + where + F: Fn() -> Option + Send + Sync + 'static, + { + self.distinct_id = DistinctIdSource::Provider(Arc::new(provider)); + self + } + + /// Captures events without a caller-provided distinct ID. + pub fn anonymous(mut self) -> Self { + self.distinct_id = DistinctIdSource::Anonymous; + self + } + + /// Adds a property to every captured PostHog event. + /// + /// Event fields with the same name override shared properties. + /// + /// # Example + /// + /// ```no_run + /// let layer = PostHogLayer::new(client) + /// .with_property("service", "checkout"); + /// + /// tracing::info!(service = "api", "request handled"); + /// // The event contains `service = "api"`. + /// ``` + pub fn with_property(mut self, key: impl Into, value: impl Into) -> Self { + self.properties.insert(key.into(), value.into()); + self + } +} + +#[inline] +fn is_sdk_target(target: &str) -> bool { + target == "posthog_rs" || target.starts_with("posthog_rs::") +} + +impl Layer for PostHogLayer +where + S: Subscriber, +{ + fn on_event(&self, event: &TracingEvent<'_>, _ctx: Context<'_, S>) { + let metadata = event.metadata(); + + if is_sdk_target(metadata.target()) { + return; + } + + // for OOMs + let _guard = match CapturingGuard::try_acquire() { + Some(guard) => guard, + None => return, + }; + + let mut visitor = Visitor::default(); + event.record(&mut visitor); + + let distinct_id = self + .distinct_id + .resolve(visitor.fields.remove("distinct_id")); + + let event_name = self.event_namer.name(metadata); + + let mut posthog_event = match distinct_id { + Some(id) => Event::new(event_name, id), + None => Event::new_anon(event_name), + }; + + let mut fields = self.properties.clone(); + fields.extend(visitor.fields); + + for (key, value) in fields { + _ = posthog_event.insert_prop(key, value); + } + + self.client.capture(posthog_event); + } +} diff --git a/tests/test_tracing.rs b/tests/test_tracing.rs new file mode 100644 index 00000000..cce18ccc --- /dev/null +++ b/tests/test_tracing.rs @@ -0,0 +1,351 @@ +#![cfg(feature = "tracing-subscriber")] + +use std::{ + fmt::{self, Debug, Display}, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, Mutex, + }, +}; + +use httpmock::prelude::*; +use posthog_rs::{Client, ClientOptions, ClientOptionsBuilder, Event, EventNamer, PostHogLayer}; +use serde_json::Value; +use tracing::{Metadata, Subscriber}; +use tracing_subscriber::{ + filter::{LevelFilter, Targets}, + layer::{Context, Layer, SubscriberExt}, +}; + +async fn create_client(options: ClientOptions) -> Client { + #[cfg(feature = "async-client")] + { + posthog_rs::client(options).await + } + + #[cfg(not(feature = "async-client"))] + { + posthog_rs::client(options) + } +} + +async fn flush(client: &Client) { + #[cfg(feature = "async-client")] + { + client.flush().await; + } + + #[cfg(not(feature = "async-client"))] + { + client.flush(); + } +} + +async fn client_with_hook(host: &str, hook: F) -> Client +where + F: FnMut(Event) -> Option + Send + 'static, +{ + let mut builder = ClientOptionsBuilder::default(); + + builder + .api_key("phc_test".to_string()) + .host(host.to_string()) + .before_send(hook); + + create_client(builder.build().unwrap()).await +} + +async fn client_without_hook(host: &str) -> Client { + let mut builder = ClientOptionsBuilder::default(); + + builder + .api_key("phc_test".to_string()) + .host(host.to_string()); + + create_client(builder.build().unwrap()).await +} + +fn capture_events() -> Arc>> { + Arc::new(Mutex::new(Vec::new())) +} + +fn capture_hook(events: Arc>>) -> impl FnMut(Event) -> Option + Send { + move |event| { + events + .lock() + .unwrap() + .push(serde_json::to_value(event).unwrap()); + None + } +} + +struct DisplayValue; + +impl Display for DisplayValue { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("display-value") + } +} + +struct DebugValue; + +impl Debug for DebugValue { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("debug-value") + } +} + +#[tokio::test] +async fn captures_event_with_field_and_context_fidelity() { + let events = capture_events(); + let client = + Arc::new(client_with_hook("http://127.0.0.1:1", capture_hook(Arc::clone(&events))).await); + + let layer = PostHogLayer::new(Arc::clone(&client)) + .with_event_namer(EventNamer::Custom(Arc::new(|_| "trace_event".to_owned()))) + .with_property("service", "static"); + + let subscriber = tracing_subscriber::registry().with(layer); + + let display_value = DisplayValue; + let debug_value = DebugValue; + + tracing::subscriber::with_default(subscriber, || { + tracing::trace!( + src = "posthog", + field1 = "one", + field2 = "two", + distinct_id = "user-123", + service = "dynamic", + signed = i64::MIN, + unsigned = u64::MAX, + amount = 49.95_f64, + successful = true, + display_field = %display_value, + debug_field = ?debug_value, + "payment completed" + ); + }); + + flush(&client).await; + + let events = events.lock().unwrap(); + assert_eq!(events.len(), 1); + + let event = &events[0]; + + assert_eq!(event["event"], "trace_event"); + assert_eq!(event["distinct_id"], "user-123"); + + let properties = &event["properties"]; + assert_eq!(properties["src"], "posthog"); + assert_eq!(properties["field1"], "one"); + assert_eq!(properties["field2"], "two"); + assert_eq!(properties["service"], "dynamic"); + assert_eq!(properties["signed"], i64::MIN); + assert_eq!(properties["unsigned"], u64::MAX); + assert_eq!(properties["amount"], 49.95); + assert_eq!(properties["successful"], true); + assert_eq!(properties["display_field"], "display-value"); + assert_eq!(properties["debug_field"], "debug-value"); + assert_eq!(properties["message"], "payment completed"); + + assert!(properties.get("distinct_id").is_none()); +} + +struct CountingLayer { + count: Arc, +} + +impl Layer for CountingLayer +where + S: Subscriber, +{ + fn on_event(&self, _event: &tracing::Event<'_>, _ctx: Context<'_, S>) { + self.count.fetch_add(1, Ordering::Relaxed); + } +} + +#[tokio::test] +async fn filters_events_per_layer_and_excludes_sdk_targets() { + let events = capture_events(); + let client = + Arc::new(client_with_hook("http://127.0.0.1:1", capture_hook(Arc::clone(&events))).await); + + let seen_by_other_layer = Arc::new(AtomicUsize::new(0)); + + let posthog_layer = PostHogLayer::new(Arc::clone(&client)).with_filter( + Targets::new() + .with_target("posthog", tracing::Level::INFO) + .with_target("posthog_rs", tracing::Level::INFO) + .with_default(LevelFilter::OFF), + ); + + let subscriber = tracing_subscriber::registry() + .with(CountingLayer { + count: Arc::clone(&seen_by_other_layer), + }) + .with(posthog_layer); + + tracing::subscriber::with_default(subscriber, || { + tracing::info!( + target: "posthog", + marker = "captured", + "capture this" + ); + + tracing::debug!( + target: "posthog", + marker = "debug", + "filter this" + ); + + tracing::info!( + target: "some_other_target", + marker = "other", + "filter this too" + ); + + tracing::info!( + target: "posthog_rs::client", + marker = "sdk", + "never capture this" + ); + + assert_eq!(seen_by_other_layer.load(Ordering::Relaxed), 4); + }); + + flush(&client).await; + + let events = events.lock().unwrap(); + + assert_eq!(events.len(), 1); + assert_eq!(events[0]["properties"]["marker"], "captured"); +} + +#[tokio::test] +async fn prevents_reentrant_capture_and_resets_guard() { + let events = capture_events(); + let client = + Arc::new(client_with_hook("http://127.0.0.1:1", capture_hook(Arc::clone(&events))).await); + + let provider_calls = Arc::new(AtomicUsize::new(0)); + + let provider = { + let provider_calls = Arc::clone(&provider_calls); + + move || { + let call = provider_calls.fetch_add(1, Ordering::SeqCst); + + if call == 0 { + tracing::info!( + target: "posthog", + marker = "nested", + "must not recurse" + ); + + Some("user-123".to_owned()) + } else { + None + } + } + }; + + let layer = PostHogLayer::new(Arc::clone(&client)) + .with_distinct_id_provider(provider) + .with_filter( + Targets::new() + .with_target("posthog", LevelFilter::INFO) + .with_target("posthog_rs", LevelFilter::INFO) + .with_default(LevelFilter::OFF), + ); + + let subscriber = tracing_subscriber::registry().with(layer); + + tracing::subscriber::with_default(subscriber, || { + tracing::info!( + target: "posthog", + marker = "first", + "first event" + ); + + tracing::info!( + target: "posthog", + marker = "second", + "second event" + ); + + assert_eq!(provider_calls.load(Ordering::SeqCst), 2); + }); + + flush(&client).await; + + let events = events.lock().unwrap(); + + assert_eq!(events.len(), 2); + + assert_eq!(events[0]["distinct_id"], "user-123"); + assert_eq!(events[0]["properties"]["marker"], "first"); + + assert_eq!(events[1]["properties"]["marker"], "second"); + assert_eq!(events[1]["properties"]["$process_person_profile"], false); + + let second_distinct_id = events[1]["distinct_id"] + .as_str() + .expect("anonymous event should have a distinct_id"); + + assert!(!second_distinct_id.is_empty()); + assert_ne!(second_distinct_id, "user-123"); +} + +#[tokio::test] +async fn captures_tracing_event_through_client_transport() { + let server = MockServer::start(); + + #[cfg(feature = "capture-v1")] + let capture_mock = server.mock(|when, then| { + when.method(POST) + .path("/i/v1/analytics/events") + .body_includes(r#""event":"posthog""#) + .body_includes(r#""distinct_id":"transport-user""#) + .body_includes(r#""request_id":"req-123""#) + .body_includes(r#""message":"transport smoke test""#); + + then.status(200); + }); + + #[cfg(not(feature = "capture-v1"))] + let capture_mock = server.mock(|when, then| { + when.method(POST) + .path("/batch/") + .body_includes(r#""event":"posthog""#) + .body_includes(r#""distinct_id":"transport-user""#) + .body_includes(r#""request_id":"req-123""#) + .body_includes(r#""message":"transport smoke test""#); + + then.status(200); + }); + + let client = Arc::new(client_without_hook(&server.base_url()).await); + + let layer = PostHogLayer::new(Arc::clone(&client)).with_filter( + Targets::new() + .with_target("posthog", tracing::Level::INFO) + .with_default(LevelFilter::OFF), + ); + + let subscriber = tracing_subscriber::registry().with(layer); + + tracing::subscriber::with_default(subscriber, || { + tracing::info!( + target: "posthog", + distinct_id = "transport-user", + request_id = "req-123", + "transport smoke test" + ); + }); + + flush(&client).await; + + capture_mock.assert_calls(1); +}