feat: add support for Opentelemetry Metrics (#249)

* create metrics feature & meter provider init implementation

* add env vars for meter initialization and test in axum otlp example

* delete superfluous old file metrics.rs

* rename environment variables for metrics

* apply PR suggestions and update README.md

Author: Lur1an

examples/axum-otlp/Cargo.toml

@@ -13,6 +13,7 @@ axum-tracing-opentelemetry = { path = "../../axum-tracing-opentelemetry" }
 init-tracing-opentelemetry = { path = "../../init-tracing-opentelemetry", features = [
   "otlp",
   "tracing_subscriber_ext",
+  "metrics"
 ] }
 opentelemetry = { workspace = true }
 opentelemetry-otlp = { workspace = true, default-features = false, features = [

examples/axum-otlp/src/main.rs

@@ -45,6 +45,7 @@ async fn health() -> impl IntoResponse {
 
 #[tracing::instrument]
 async fn index() -> impl IntoResponse {
+    tracing::info!(monotonic_counter.index = 1);
     let trace_id = find_current_trace_id();
     dbg!(&trace_id);
     //std::thread::sleep(std::time::Duration::from_secs(1));

init-tracing-opentelemetry/Cargo.toml

@@ -20,9 +20,7 @@ opentelemetry-otlp = { workspace = true, optional = true, features = [
   "trace",
 ] }
 # opentelemetry-resource-detectors = { workspace = true } //FIXME enable when available for opentelemetry >= 0.25
-opentelemetry-stdout = { workspace = true, features = [
-  "trace",
-], optional = true }
+opentelemetry-stdout = { workspace = true, features = ["trace"], optional = true }
 opentelemetry-semantic-conventions = { workspace = true, optional = true }
 opentelemetry-zipkin = { workspace = true, features = [], optional = true }
 opentelemetry_sdk = { workspace = true }
@@ -74,3 +72,4 @@ tls = ["opentelemetry-otlp/tls", "tonic"]
 tls-roots = ["opentelemetry-otlp/tls-roots"]
 tls-webpki-roots = ["opentelemetry-otlp/tls-webpki-roots"]
 logfmt = ["dep:tracing-logfmt"]
+metrics = ["opentelemetry-otlp/metrics", "tracing-opentelemetry/metrics", "opentelemetry-stdout/metrics"]

init-tracing-opentelemetry/README.md

@@ -17,9 +17,9 @@ async fn main() -> Result<(), axum::BoxError> {
 }
 ```
 
-The `init_subscribers` function returns a `TracingGuard` instance. Following the guard pattern, this struct provides no functions but, when dropped, ensures that any pending traces are sent before it exits. The syntax `let _guard` is suggested to ensure that Rust does not drop the struct until the application exits.
+The `init_subscribers` function returns a `OtelGuard` instance. Following the guard pattern, this struct provides no functions but, when dropped, ensures that any pending traces/metrics are sent before it exits. The syntax `let _guard` is suggested to ensure that Rust does not drop the struct until the application exits.
 
-To configure opentelemetry tracer & tracing, you can use the functions from `init_tracing_opentelemetry::tracing_subscriber_ext`, but they are very opinionated (and WIP to make them more customizable and friendly), so we recommend making your composition, but look at the code (to avoid some issue) and share your feedback.
+To configure opentelemetry tracer & tracing (& metrics), you can use the functions from `init_tracing_opentelemetry::tracing_subscriber_ext`, but they are very opinionated (and WIP to make them more customizable and friendly), so we recommend making your composition, but look at the code (to avoid some issue) and share your feedback.
 
 ```txt
 pub fn build_loglevel_filter_layer() -> tracing_subscriber::filter::EnvFilter {
@@ -161,6 +161,20 @@ spec:
 - check the environment variables of opentelemetry `OTEL_EXPORTER...` and `OTEL_TRACES_SAMPLER` (values are logged on target `otel::setup` )
 - check that log target `otel::tracing` enable log level `trace` (or `info` if you use `tracing_level_info` feature) to generate span to send to opentelemetry collector.
 
+## Metrics
+
+To configure opentelemetry metrics, enable the `metrics` feature, this will initialize a `SdkMeterProvider`, set it globally and add a a [`MetricsLayer`](https://docs.rs/tracing-opentelemetry/latest/tracing_opentelemetry/struct.MetricsLayer.html) to allow using `tracing` events to produce metrics.
+
+The `opentelemetry_sdk` can still be used to produce metrics as well, since we configured the `SdkMeterProvider` globally, so any Axum/Tonic middleware that does not use `tracing` but directly [opentelemetry::metrics](https://docs.rs/opentelemetry/latest/opentelemetry/metrics/struct.Meter.html) will work.
+
+Configure the following set of environment variables to configure the metrics exporter (on top of those configured above):
+
+- `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` override to `OTEL_EXPORTER_OTLP_ENDPOINT` for the url of the exporter / collector
+- `OTEL_EXPORTER_OTLP_METRICS_PROTOCOL` override to `OTEL_EXPORTER_OTLP_PROTOCOL`, fallback to auto-detection based on ENDPOINT port
+- `OTEL_EXPORTER_OTLP_METRICS_TIMEOUT` to set the timeout for the connection to the exporter
+- `OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE` to set the temporality preference for the exporter
+- `OTEL_METRIC_EXPORT_INTERVAL` to set frequence of metrics export in __*milliseconds*__, defaults to 60s
+
 ## Changelog - History
 
 [CHANGELOG.md](https://github.com/davidB/tracing-opentelemetry-instrumentation-sdk/blob/main/CHANGELOG.md)

init-tracing-opentelemetry/src/otlp/metrics.rs

@@ -0,0 +1,122 @@
+use super::infer_protocol;
+use crate::resource::DetectResource;
+use crate::Error;
+use opentelemetry::global;
+use opentelemetry_otlp::{ExporterBuildError, MetricExporter, WithExportConfig};
+use opentelemetry_sdk::metrics::{
+    MeterProviderBuilder, PeriodicReader, SdkMeterProvider, Temporality,
+};
+use opentelemetry_sdk::Resource;
+use std::env;
+use std::time::Duration;
+use tracing::Subscriber;
+use tracing_opentelemetry::MetricsLayer;
+use tracing_subscriber::registry::LookupSpan;
+#[cfg(feature = "tls")]
+use {opentelemetry_otlp::WithTonicConfig, tonic::transport::ClientTlsConfig};
+
+pub fn build_metrics_layer<S>() -> Result<(MetricsLayer<S>, SdkMeterProvider), Error>
+where
+    S: Subscriber + for<'a> LookupSpan<'a>,
+{
+    let otel_rsrc = DetectResource::default().build();
+    let meter_provider = init_meterprovider(otel_rsrc, identity)?;
+    global::set_meter_provider(meter_provider.clone());
+    let layer = tracing_opentelemetry::MetricsLayer::new(meter_provider.clone());
+    Ok((layer, meter_provider))
+}
+
+#[must_use]
+pub fn identity(v: MeterProviderBuilder) -> MeterProviderBuilder {
+    v
+}
+
+pub fn init_meterprovider<F>(
+    resource: Resource,
+    transform: F,
+) -> Result<SdkMeterProvider, ExporterBuildError>
+where
+    F: FnOnce(MeterProviderBuilder) -> MeterProviderBuilder,
+{
+    let (maybe_protocol, maybe_endpoint) = read_protocol_and_endpoint_from_env();
+    let protocol = infer_protocol(maybe_protocol.as_deref(), maybe_endpoint.as_deref());
+    let timeout = env::var("OTEL_EXPORTER_OTLP_METRICS_TIMEOUT")
+        .ok()
+        .and_then(|var| var.parse::<u64>().ok())
+        .map_or(Duration::from_secs(10), Duration::from_secs);
+    let temporality = env::var("OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE")
+        .ok()
+        .and_then(|var| match var.to_lowercase().as_str() {
+            "delta" => Some(Temporality::Delta),
+            "cumulative" => Some(Temporality::Cumulative),
+            unknown => {
+                tracing::warn!("unknown '{unknown}' env var set for OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY; defaulting to cumulative");
+                None
+            },
+        })
+        .unwrap_or_default();
+    let export_interval = env::var("OTEL_METRIC_EXPORT_INTERVAL")
+        .ok()
+        .and_then(|var| var.parse::<u64>().ok())
+        .map_or(Duration::from_secs(60), Duration::from_millis);
+
+    let exporter = match protocol.as_deref() {
+        Some("http/protobuf") => Some(
+            MetricExporter::builder()
+                .with_http()
+                .with_temporality(temporality)
+                .with_timeout(timeout)
+                .build()?,
+        ),
+        #[cfg(feature = "tls")]
+        Some("grpc/tls") => Some(
+            MetricExporter::builder()
+                .with_tonic()
+                .with_tls_config(ClientTlsConfig::new().with_enabled_roots())
+                .with_temporality(temporality)
+                .with_timeout(timeout)
+                .build()?,
+        ),
+        Some("grpc") => Some(
+            MetricExporter::builder()
+                .with_tonic()
+                .with_temporality(temporality)
+                .with_timeout(timeout)
+                .build()?,
+        ),
+        Some(x) => {
+            tracing::warn!("unknown '{x}' env var set or infered for OTEL_EXPORTER_OTLP_METRICS_PROTOCOL or OTEL_EXPORTER_OTLP_PROTOCOL; no metric exporter will be created");
+            None
+        }
+        None => {
+            tracing::warn!("no env var set or infered for OTEL_EXPORTER_OTLP_METRICS_PROTOCOL or OTEL_EXPORTER_OTLP_PROTOCOL; no metric exporter will be created");
+            None
+        }
+    };
+    let mut meter_provider = SdkMeterProvider::builder().with_resource(resource);
+    if let Some(exporter) = exporter {
+        let reader = PeriodicReader::builder(exporter)
+            .with_interval(export_interval)
+            .build();
+        meter_provider = meter_provider.with_reader(reader);
+    }
+    meter_provider = transform(meter_provider);
+    Ok(meter_provider.build())
+}
+
+fn read_protocol_and_endpoint_from_env() -> (Option<String>, Option<String>) {
+    let maybe_protocol = std::env::var("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL")
+        .or_else(|_| std::env::var("OTEL_EXPORTER_OTLP_PROTOCOL"))
+        .ok();
+    let maybe_endpoint = std::env::var("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT")
+        .or_else(|_| {
+            std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").map(|endpoint| match &maybe_protocol {
+                Some(protocol) if protocol.contains("http") => {
+                    format!("{endpoint}/v1/metrics")
+                }
+                _ => endpoint,
+            })
+        })
+        .ok();
+    (maybe_protocol, maybe_endpoint)
+}

init-tracing-opentelemetry/src/otlp/mod.rs

@@ -0,0 +1,116 @@
+#[cfg(feature = "metrics")]
+pub mod metrics;
+pub mod traces;
+
+#[cfg(feature = "metrics")]
+use opentelemetry::metrics::MeterProvider;
+#[cfg(feature = "metrics")]
+use opentelemetry_sdk::metrics::SdkMeterProvider;
+
+use opentelemetry::trace::TracerProvider;
+use opentelemetry_sdk::trace::SdkTracerProvider;
+
+#[must_use = "Recommend holding with 'let _guard = ' pattern to ensure final traces/metrics are sent to the server"]
+/// On Drop of the `OtelGuard` instance,
+/// the wrapped Tracer/Meter Provider is force to flush and to shutdown (ignoring error).
+pub struct OtelGuard {
+    #[cfg(feature = "metrics")]
+    pub(crate) meter_provider: SdkMeterProvider,
+    pub(crate) tracer_provider: SdkTracerProvider,
+}
+
+impl OtelGuard {
+    #[must_use]
+    pub fn tracer_provider(&self) -> &impl TracerProvider {
+        &self.tracer_provider
+    }
+
+    #[cfg(feature = "metrics")]
+    #[must_use]
+    pub fn meter_provider(&self) -> &impl MeterProvider {
+        &self.meter_provider
+    }
+}
+
+impl Drop for OtelGuard {
+    #[allow(unused_must_use)]
+    fn drop(&mut self) {
+        let _ = self.tracer_provider.force_flush();
+        let _ = self.tracer_provider.shutdown();
+        #[cfg(feature = "metrics")]
+        {
+            let _ = self.meter_provider.force_flush();
+            let _ = self.meter_provider.shutdown();
+        }
+    }
+}
+
+#[allow(unused_mut)]
+pub(crate) fn infer_protocol(
+    maybe_protocol: Option<&str>,
+    maybe_endpoint: Option<&str>,
+) -> Option<String> {
+    let mut maybe_protocol = match (maybe_protocol, maybe_endpoint) {
+        (Some(protocol), _) => Some(protocol.to_string()),
+        (None, Some(endpoint)) => {
+            if endpoint.contains(":4317") {
+                Some("grpc".to_string())
+            } else {
+                Some("http/protobuf".to_string())
+            }
+        }
+        _ => None,
+    };
+    #[cfg(feature = "tls")]
+    if maybe_protocol.as_deref() == Some("grpc")
+        && maybe_endpoint.is_some_and(|e| e.starts_with("https"))
+    {
+        maybe_protocol = Some("grpc/tls".to_string());
+    }
+
+    maybe_protocol
+}
+
+#[cfg(test)]
+mod tests {
+    use assert2::assert;
+    use rstest::rstest;
+
+    use super::*;
+
+    #[rstest]
+    #[case(None, None, None)] //Devskim: ignore DS137138
+    #[case(Some("http/protobuf"), None, Some("http/protobuf"))] //Devskim: ignore DS137138
+    #[case(Some("grpc"), None, Some("grpc"))] //Devskim: ignore DS137138
+    #[case(None, Some("http://localhost:4317"), Some("grpc"))] //Devskim: ignore DS137138
+    #[cfg_attr(
+        feature = "tls",
+        case(None, Some("https://localhost:4317"), Some("grpc/tls"))
+    )]
+    #[cfg_attr(
+        feature = "tls",
+        case(Some("grpc/tls"), Some("https://localhost:4317"), Some("grpc/tls"))
+    )]
+    #[case(
+        Some("http/protobuf"),
+        Some("http://localhost:4318/v1/traces"), //Devskim: ignore DS137138
+        Some("http/protobuf"),
+    )]
+    #[case(
+        Some("http/protobuf"),
+        Some("https://examples.com:4318/v1/traces"),
+        Some("http/protobuf")
+    )]
+    #[case(
+        Some("http/protobuf"),
+        Some("https://examples.com:4317"),
+        Some("http/protobuf")
+    )]
+    fn test_infer_protocol(
+        #[case] traces_protocol: Option<&str>,
+        #[case] traces_endpoint: Option<&str>,
+        #[case] expected_protocol: Option<&str>,
+    ) {
+        assert!(infer_protocol(traces_protocol, traces_endpoint).as_deref() == expected_protocol);
+    }
+}

init-tracing-opentelemetry/src/otlp.rs …tracing-opentelemetry/src/otlp/traces.rsinit-tracing-opentelemetry/src/otlp.rs renamed to init-tracing-opentelemetry/src/otlp/traces.rs init-tracing-opentelemetry/src/otlp.rs renamed to init-tracing-opentelemetry/src/otlp/traces.rs

@@ -1,3 +1,4 @@
+use super::infer_protocol;
 use opentelemetry_otlp::{ExporterBuildError, SpanExporter};
 use opentelemetry_sdk::{trace::SdkTracerProvider, trace::TracerProviderBuilder, Resource};
 #[cfg(feature = "tls")]
@@ -70,70 +71,3 @@ fn read_protocol_and_endpoint_from_env() -> (Option<String>, Option<String>) {
         .ok();
     (maybe_protocol, maybe_endpoint)
 }
-
-#[allow(unused_mut)]
-fn infer_protocol(maybe_protocol: Option<&str>, maybe_endpoint: Option<&str>) -> Option<String> {
-    let mut maybe_protocol = match (maybe_protocol, maybe_endpoint) {
-        (Some(protocol), _) => Some(protocol.to_string()),
-        (None, Some(endpoint)) => {
-            if endpoint.contains(":4317") {
-                Some("grpc".to_string())
-            } else {
-                Some("http/protobuf".to_string())
-            }
-        }
-        _ => None,
-    };
-    #[cfg(feature = "tls")]
-    if maybe_protocol.as_deref() == Some("grpc")
-        && maybe_endpoint.is_some_and(|e| e.starts_with("https"))
-    {
-        maybe_protocol = Some("grpc/tls".to_string());
-    }
-
-    maybe_protocol
-}
-
-#[cfg(test)]
-mod tests {
-    use assert2::assert;
-    use rstest::rstest;
-
-    use super::*;
-
-    #[rstest]
-    #[case(None, None, None)] //Devskim: ignore DS137138
-    #[case(Some("http/protobuf"), None, Some("http/protobuf"))] //Devskim: ignore DS137138
-    #[case(Some("grpc"), None, Some("grpc"))] //Devskim: ignore DS137138
-    #[case(None, Some("http://localhost:4317"), Some("grpc"))] //Devskim: ignore DS137138
-    #[cfg_attr(
-        feature = "tls",
-        case(None, Some("https://localhost:4317"), Some("grpc/tls"))
-    )]
-    #[cfg_attr(
-        feature = "tls",
-        case(Some("grpc/tls"), Some("https://localhost:4317"), Some("grpc/tls"))
-    )]
-    #[case(
-        Some("http/protobuf"),
-        Some("http://localhost:4318/v1/traces"), //Devskim: ignore DS137138
-        Some("http/protobuf"),
-    )]
-    #[case(
-        Some("http/protobuf"),
-        Some("https://examples.com:4318/v1/traces"),
-        Some("http/protobuf")
-    )]
-    #[case(
-        Some("http/protobuf"),
-        Some("https://examples.com:4317"),
-        Some("http/protobuf")
-    )]
-    fn test_infer_protocol(
-        #[case] traces_protocol: Option<&str>,
-        #[case] traces_endpoint: Option<&str>,
-        #[case] expected_protocol: Option<&str>,
-    ) {
-        assert!(infer_protocol(traces_protocol, traces_endpoint).as_deref() == expected_protocol);
-    }
-}