support tracing metrics layer (#76)

Author: davidhewitt

src/bridges/tracing.rs

@@ -19,20 +19,30 @@ use crate::{__macros_impl::LogfireValue, LogfireTracer};
 /// this layer.
 pub struct LogfireTracingLayer<S> {
     tracer: LogfireTracer,
+    /// This odd structure with two inner layers is deliberate; we don't want to send any events
+    /// to the `otel_layer` and we only send (some) events to the `metrics_layer`.
     otel_layer: tracing_opentelemetry::OpenTelemetryLayer<S, opentelemetry_sdk::trace::Tracer>,
+    metrics_layer: Option<tracing_opentelemetry::MetricsLayer<S>>,
 }
 
 impl<S> LogfireTracingLayer<S>
 where
     S: Subscriber + for<'span> LookupSpan<'span>,
 {
     /// Create a new `LogfireTracingLayer` with the given tracer.
-    pub(crate) fn new(tracer: LogfireTracer) -> Self {
+    pub(crate) fn new(tracer: LogfireTracer, enable_tracing_metrics: bool) -> Self {
         let otel_layer = tracing_opentelemetry::layer()
             .with_error_records_to_exceptions(true)
             .with_tracer(tracer.inner.clone());
 
-        LogfireTracingLayer { tracer, otel_layer }
+        let metrics_layer = enable_tracing_metrics
+            .then(|| tracing_opentelemetry::MetricsLayer::new(tracer.meter_provider.clone()));
+
+        LogfireTracingLayer {
+            tracer,
+            otel_layer,
+            metrics_layer,
+        }
     }
 }
 
@@ -49,6 +59,9 @@ where
         // Delegate to OpenTelemetry layer first
         self.otel_layer.on_new_span(attrs, id, ctx.clone());
 
+        // Delegate to MetricsLayer as well
+        self.metrics_layer.on_new_span(attrs, id, ctx.clone());
+
         // Add Logfire-specific attributes
         let span = ctx.span(id).expect("span not found");
         let mut extensions = span.extensions_mut();
@@ -95,7 +108,23 @@ where
     ///
     /// Instead we need to handle them here and write them to the logfire writer.
     fn on_event(&self, event: &tracing::Event<'_>, ctx: tracing_subscriber::layer::Context<'_, S>) {
-        // Don't delegate events to OpenTelemetry layer, we emit them as log spans instead.
+        let is_metrics_event = self.metrics_layer.is_some()
+            && event.fields().any(|field| {
+                let name = field.name();
+
+                name.starts_with("counter.")
+                    || name.starts_with("monotonic_counter.")
+                    || name.starts_with("histogram.")
+                    || name.starts_with("monotonic_histogram.")
+            });
+
+        // Allow the metrics layer to see all events, so it can record metrics as needed.
+        if is_metrics_event {
+            self.metrics_layer.on_event(event, ctx.clone());
+        }
+
+        // However we don't want to allow the opentelemetry layer to see events, it will record them
+        // as span events. Instead we handle them here and emit them as log spans.
         let event_span = ctx.event_span(event).and_then(|span| ctx.span(&span.id()));
         let mut event_span_extensions = event_span.as_ref().map(|s| s.extensions_mut());
 
@@ -161,6 +190,43 @@ where
             unsafe { self.otel_layer.downcast_raw(id) }
         }
     }
+
+    fn on_register_dispatch(&self, subscriber: &tracing::Dispatch) {
+        self.otel_layer.on_register_dispatch(subscriber);
+        self.metrics_layer.on_register_dispatch(subscriber);
+    }
+
+    fn on_layer(&mut self, subscriber: &mut S) {
+        self.otel_layer.on_layer(subscriber);
+        self.metrics_layer.on_layer(subscriber);
+    }
+
+    fn enabled(
+        &self,
+        metadata: &tracing::Metadata<'_>,
+        ctx: tracing_subscriber::layer::Context<'_, S>,
+    ) -> bool {
+        self.otel_layer.enabled(metadata, ctx.clone()) || self.metrics_layer.enabled(metadata, ctx)
+    }
+
+    fn event_enabled(
+        &self,
+        event: &tracing::Event<'_>,
+        ctx: tracing_subscriber::layer::Context<'_, S>,
+    ) -> bool {
+        self.otel_layer.event_enabled(event, ctx.clone())
+            || self.metrics_layer.event_enabled(event, ctx)
+    }
+
+    fn on_id_change(
+        &self,
+        old: &tracing::span::Id,
+        new: &tracing::span::Id,
+        ctx: tracing_subscriber::layer::Context<'_, S>,
+    ) {
+        self.otel_layer.on_id_change(old, new, ctx.clone());
+        self.metrics_layer.on_id_change(old, new, ctx);
+    }
 }
 
 /// Dummy struct to mark that we've already entered this span.
@@ -1971,4 +2037,233 @@ mod tests {
         1970-01-01T00:00:00.000009Z DEBUG opentelemetry_sdk::logs::logger_provider  name=LoggerProvider.ShutdownInvokedByUser
         ");
     }
+
+    #[tokio::test]
+    async fn test_tracing_metrics_layer() {
+        use crate::test_utils::make_deterministic_resource_metrics;
+        use opentelemetry_sdk::metrics::{
+            InMemoryMetricExporterBuilder, ManualReader, data::ResourceMetrics,
+            exporter::PushMetricExporter, reader::MetricReader,
+        };
+        use std::sync::Arc;
+
+        #[derive(Clone, Debug)]
+        struct SharedManualReader {
+            reader: Arc<ManualReader>,
+        }
+
+        impl SharedManualReader {
+            fn new(reader: ManualReader) -> Self {
+                Self {
+                    reader: Arc::new(reader),
+                }
+            }
+
+            async fn export<E: PushMetricExporter>(&self, exporter: &E) {
+                let mut metrics = ResourceMetrics::default();
+                self.reader.collect(&mut metrics).unwrap();
+                exporter.export(&mut metrics).await.unwrap();
+            }
+        }
+
+        impl MetricReader for SharedManualReader {
+            fn register_pipeline(
+                &self,
+                pipeline: std::sync::Weak<opentelemetry_sdk::metrics::Pipeline>,
+            ) {
+                self.reader.register_pipeline(pipeline);
+            }
+
+            fn collect(
+                &self,
+                rm: &mut opentelemetry_sdk::metrics::data::ResourceMetrics,
+            ) -> opentelemetry_sdk::error::OTelSdkResult {
+                self.reader.collect(rm)
+            }
+
+            fn force_flush(&self) -> opentelemetry_sdk::error::OTelSdkResult {
+                self.reader.force_flush()
+            }
+
+            fn shutdown(&self) -> opentelemetry_sdk::error::OTelSdkResult {
+                self.reader.shutdown()
+            }
+
+            fn shutdown_with_timeout(
+                &self,
+                timeout: std::time::Duration,
+            ) -> opentelemetry_sdk::error::OTelSdkResult {
+                self.reader.shutdown_with_timeout(timeout)
+            }
+
+            fn temporality(
+                &self,
+                kind: opentelemetry_sdk::metrics::InstrumentKind,
+            ) -> opentelemetry_sdk::metrics::Temporality {
+                self.reader.temporality(kind)
+            }
+        }
+
+        let mut exporter = InMemoryMetricExporterBuilder::new().build();
+
+        let reader = SharedManualReader::new(
+            ManualReader::builder()
+                .with_temporality(opentelemetry_sdk::metrics::Temporality::Delta)
+                .build(),
+        );
+
+        let handler = crate::configure()
+            .local()
+            .send_to_logfire(false)
+            .with_metrics(Some(
+                crate::config::MetricsOptions::default().with_additional_reader(reader.clone()),
+            ))
+            .install_panic_handler()
+            .with_default_level_filter(LevelFilter::TRACE)
+            .with_advanced_options(
+                AdvancedOptions::default()
+                    .with_resource(
+                        opentelemetry_sdk::Resource::builder_empty()
+                            .with_service_name("test")
+                            .build(),
+                    )
+                    .with_id_generator(DeterministicIdGenerator::new())
+                    .with_tracing_metrics(true),
+            )
+            .finish()
+            .unwrap();
+
+        let _guard = set_local_logfire(handler.clone());
+
+        tracing::info!(counter.test_counter = 1, "test counter event");
+        tracing::info!(histogram.test_histogram = 2.5, "test histogram event");
+        tracing::info!(
+            monotonic_counter.test_monotonic = 3,
+            "test monotonic counter event"
+        );
+
+        tracing::info!(counter.test_counter = 2, "test counter event");
+        tracing::info!(histogram.test_histogram = 3.5, "test histogram event");
+        tracing::info!(
+            monotonic_counter.test_monotonic = 4,
+            "test monotonic counter event"
+        );
+
+        reader.export(&mut exporter).await;
+
+        tracing::info!(counter.test_counter = 3, "test counter event");
+        tracing::info!(histogram.test_histogram = 4.5, "test histogram event");
+        tracing::info!(
+            monotonic_counter.test_monotonic = 5,
+            "test monotonic counter event"
+        );
+
+        reader.export(&mut exporter).await;
+
+        handler.shutdown().unwrap();
+
+        let metrics = exporter.get_finished_metrics().unwrap();
+        let metrics = make_deterministic_resource_metrics(metrics);
+
+        assert_debug_snapshot!(metrics, @r#"
+        [
+            DeterministicResourceMetrics {
+                resource: Resource {
+                    inner: ResourceInner {
+                        attrs: {
+                            Static(
+                                "service.name",
+                            ): String(
+                                Static(
+                                    "test",
+                                ),
+                            ),
+                        },
+                        schema_url: None,
+                    },
+                },
+                scope_metrics: [
+                    DeterministicScopeMetrics {
+                        scope: InstrumentationScope {
+                            name: "tracing/tracing-opentelemetry",
+                            version: Some(
+                                "0.31.0",
+                            ),
+                            schema_url: None,
+                            attributes: [],
+                        },
+                        metrics: [
+                            DeterministicMetric {
+                                name: "test_counter",
+                                values: [
+                                    3,
+                                ],
+                            },
+                            DeterministicMetric {
+                                name: "test_histogram",
+                                values: [
+                                    2,
+                                ],
+                            },
+                            DeterministicMetric {
+                                name: "test_monotonic",
+                                values: [
+                                    7,
+                                ],
+                            },
+                        ],
+                    },
+                ],
+            },
+            DeterministicResourceMetrics {
+                resource: Resource {
+                    inner: ResourceInner {
+                        attrs: {
+                            Static(
+                                "service.name",
+                            ): String(
+                                Static(
+                                    "test",
+                                ),
+                            ),
+                        },
+                        schema_url: None,
+                    },
+                },
+                scope_metrics: [
+                    DeterministicScopeMetrics {
+                        scope: InstrumentationScope {
+                            name: "tracing/tracing-opentelemetry",
+                            version: Some(
+                                "0.31.0",
+                            ),
+                            schema_url: None,
+                            attributes: [],
+                        },
+                        metrics: [
+                            DeterministicMetric {
+                                name: "test_counter",
+                                values: [
+                                    6,
+                                ],
+                            },
+                            DeterministicMetric {
+                                name: "test_histogram",
+                                values: [
+                                    1,
+                                ],
+                            },
+                            DeterministicMetric {
+                                name: "test_monotonic",
+                                values: [
+                                    5,
+                                ],
+                            },
+                        ],
+                    },
+                ],
+            },
+        ]
+        "#);
+    }
 }

src/config.rs

@@ -188,15 +188,11 @@ impl std::fmt::Debug for Target {
 /// Options primarily used for testing by Logfire developers.
 #[derive(Default)]
 pub struct AdvancedOptions {
-    /// Root URL for the Logfire API.
     pub(crate) base_url: Option<String>,
-    /// Generator for trace and span IDs.
     pub(crate) id_generator: Option<BoxedIdGenerator>,
-    /// Resource to override default resource detection.
     pub(crate) resource: Option<opentelemetry_sdk::Resource>,
-
-    // Configuration for OpenTelemetry logging. This is experimental and may be removed.
     pub(crate) log_record_processors: Vec<BoxedLogProcessor>,
+    pub(crate) enable_tracing_metrics: bool,
     //
     //
     // TODO: arguments below supported by Python
@@ -230,7 +226,7 @@ impl AdvancedOptions {
         self
     }
 
-    /// Add a log processor to the list of log processors.
+    /// Add a log processor to the list of log processors. This is experimental and may be removed.
     #[must_use]
     pub fn with_log_processor<T: LogProcessor + Send + Sync + 'static>(
         mut self,
@@ -240,6 +236,13 @@ impl AdvancedOptions {
             .push(BoxedLogProcessor::new(Box::new(processor)));
         self
     }
+
+    /// Support capturing tracing events as metrics as per [`tracing_opentelemetry::MetricsLayer`](https://docs.rs/tracing-opentelemetry/latest/tracing_opentelemetry/struct.MetricsLayer.html).
+    #[must_use]
+    pub fn with_tracing_metrics(mut self, enable: bool) -> Self {
+        self.enable_tracing_metrics = enable;
+        self
+    }
 }
 
 struct RegionData {