Upgrade OTel SDK to 0.30.0

Cargo.toml

@@ -75,3 +75,7 @@ sha1 = "0.10.6"
 tempfile = "3.10.1"
 tokio = { version = "1.36.0", features = ["full"] }
 tower = { version = "0.5.1", features = ["full"] }
+
+
+[patch.crates-io]
+opentelemetry-prometheus = { git = "https://github.com/sandhose/opentelemetry-rust.git", branch = "otel-prometheus-0.30"}

apollo-router/Cargo.toml

@@ -161,20 +161,23 @@ once_cell = "1.19.0"
 # groups `^tracing` and `^opentelemetry*` dependencies together as of
 # https://github.com/apollographql/router/pull/1509.  A comment which exists
 # there (and on `tracing` packages below) should be updated should this change.
-opentelemetry = { version = "0.24.0", features = ["trace", "metrics"] }
-opentelemetry_sdk = { version = "0.24.1", default-features = false, features = [
+opentelemetry = { version = "0.30.0", features = ["trace", "metrics"] }
+opentelemetry_sdk = { version = "0.30.0", default-features = false, features = [
     "rt-tokio",
     "trace",
+    "spec_unstable_metrics_views",
+    "experimental_trace_batch_span_processor_with_async_runtime",
+    "metrics",
+    "experimental_metrics_custom_reader",
+    "experimental_metrics_periodicreader_with_async_runtime",
 ] }
-opentelemetry-aws = "0.12.0"
-# START TEMP DATADOG Temporarily remove until we upgrade otel to the latest version
-# This means including the rmp library
-# opentelemetry-datadog = { version = "0.12.0", features = ["reqwest-client"] }
+opentelemetry-appender-tracing = "0.27.0"
+opentelemetry-aws = "0.18.0"
+opentelemetry-datadog = { version = "0.18.0", features = ["agent-sampling", "reqwest-client"] }
 rmp = "0.8"
-# END TEMP DATADOG
-opentelemetry-http = "0.13.0"
-opentelemetry-jaeger-propagator = "0.3.0"
-opentelemetry-otlp = { version = "0.17.0", default-features = false, features = [
+opentelemetry-http = "0.30.0"
+opentelemetry-jaeger-propagator = "0.30.0"
+opentelemetry-otlp = { version = "0.30.0", default-features = false, features = [
     "grpc-tonic",
     "gzip-tonic",
     "tonic",
@@ -183,16 +186,17 @@ opentelemetry-otlp = { version = "0.17.0", default-features = false, features =
     "metrics",
     "reqwest-client",
     "trace",
+    "tls"
 ] }
-opentelemetry-semantic-conventions = "0.16.0"
-opentelemetry-zipkin = { version = "0.22.0", default-features = false, features = [
+opentelemetry-semantic-conventions = { version = "0.30.0", features = ["semconv_experimental"]}
+opentelemetry-zipkin = { version = "0.30.0", default-features = false, features = [
     "reqwest-client",
     "reqwest-rustls",
 ] }
-opentelemetry-prometheus = "0.17.0"
+opentelemetry-prometheus = "0.29.1"
 paste = "1.0.15"
 pin-project-lite = "0.2.14"
-prometheus = "0.13"
+prometheus = "0.14"
 prost = "0.13.0"
 prost-types = "0.13.0"
 proteus = "0.5.0"
@@ -233,11 +237,10 @@ thiserror = "2.0.0"
 tokio.workspace = true
 tokio-stream = { version = "0.1.15", features = ["sync", "net", "fs"] }
 tokio-util = { version = "0.7.11", features = ["net", "codec", "time", "compat"] }
-tonic = { version = "0.12.3", features = [
+tonic = { version = "0.13.1", features = [
     "transport",
-    "tls",
-    "tls-roots",
     "gzip",
+    "tls-native-roots",
 ] }
 tower.workspace = true
 tower-http = { version = "0.6.2", features = ["full"] }
@@ -319,16 +322,16 @@ memchr = { version = "2.7.4", default-features = false }
 mockall = "0.14.0"
 num-traits = "0.2.19"
 once_cell.workspace = true
-opentelemetry-stdout = { version = "0.5.0", features = ["trace"] }
-opentelemetry = { version = "0.24.0", features = ["testing"] }
-opentelemetry_sdk = { version = "0.24.1", features = ["testing"] }
-opentelemetry-proto = { version = "0.7.0", features = [
+opentelemetry-stdout = { version = "0.30.0", features = ["trace"] }
+opentelemetry = { version = "0.30.0", features = ["testing"] }
+opentelemetry_sdk = { version = "0.30.0", features = ["testing"] }
+opentelemetry-proto = { version = "0.30.0", features = [
     "metrics",
     "trace",
     "gen-tonic-messages",
     "with-serde",
 ] }
-opentelemetry-datadog = { version = "0.12.0", features = ["reqwest-client"] }
+opentelemetry-datadog = { version = "0.18.0", features = ["agent-sampling", "reqwest-client"] }
 p256 = "0.13.2"
 pretty_assertions = "1.4.1"
 reqwest = { version = "0.12.9", default-features = false, features = [
@@ -355,7 +358,7 @@ tracing-subscriber = { version = "0.3.20", default-features = false, features =
     "env-filter",
     "fmt",
 ] }
-tracing-opentelemetry = "0.25.0"
+tracing-opentelemetry = "0.31.0"
 tracing-test = "=0.2.5"
 tracing-mock = "0.1.0-beta.1"
 walkdir = "2.5.0"

apollo-router/src/axum_factory/metrics.rs

@@ -32,7 +32,7 @@ pub(crate) mod jemalloc {
                         tracing::warn!("Failed to read jemalloc {} stats", stringify!($name));
                     }
                 })
-                .init()
+                .build()
         };
     }
 

apollo-router/src/cache/metrics.rs

@@ -206,7 +206,7 @@ impl RedisMetricsCollector {
                     &[KeyValue::new("kind", caller)],
                 );
             })
-            .init()
+            .build()
     }
 
     /// Generic method to create a weighted average gauge
@@ -237,7 +237,7 @@ impl RedisMetricsCollector {
 
                 gauge.observe(average, &[KeyValue::new("kind", caller)]);
             })
-            .init()
+            .build()
     }
 
     fn create_client_count_gauge() -> ObservableGauge<u64> {
@@ -249,7 +249,7 @@ impl RedisMetricsCollector {
             .with_callback(move |gauge| {
                 gauge.observe(ACTIVE_CLIENT_COUNT.load(Ordering::Relaxed), &[]);
             })
-            .init()
+            .build()
     }
 
     /// Spawn the metrics collection task
@@ -521,7 +521,7 @@ mod tests {
             // Verify Redis connection metrics are emitted.
             // Since this metric is based on a global AtomicU64, it's not unique across tests - so
             // we can only reliably check for metric existence, rather than a specific value.
-            crate::metrics::collect_metrics().metric_exists::<u64>(
+            crate::metrics::collect_metrics().metric_exists(
                 "apollo.router.cache.redis.clients",
                 MetricType::Gauge,
                 &[],

apollo-router/src/cache/storage.rs

@@ -128,7 +128,7 @@ where
                     ],
                 )
             })
-            .init()
+            .build()
     }
 
     fn create_cache_estimated_storage_size_gauge(&self) -> ObservableGauge<i64> {
@@ -153,7 +153,7 @@ where
                     )
                 }
             })
-            .init()
+            .build()
     }
 
     /// `init_from_redis` is called with values newly deserialized from Redis cache
@@ -346,6 +346,7 @@ mod test {
             cache.activate();
 
             cache.insert("test".to_string(), Stuff {}).await;
+
             assert_gauge!(
                 "apollo.router.cache.storage.estimated_size",
                 1,

apollo-router/src/compute_job/metrics.rs

@@ -97,36 +97,41 @@ mod tests {
     use crate::compute_job::ComputeJobType;
     use crate::compute_job::metrics::JobWatcher;
     use crate::compute_job::metrics::Outcome;
+    use crate::metrics::FutureMetricsExt;
 
-    #[test]
-    fn test_job_watcher() {
-        let check_histogram_count =
-            |count: u64, job_type: &'static str, job_outcome: &'static str| {
-                assert_histogram_count!(
-                    "apollo.router.compute_jobs.duration",
-                    count,
-                    "job.type" = job_type,
-                    "job.outcome" = job_outcome
-                );
-            };
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_job_watcher() {
+        async {
+            let check_histogram_count =
+                |count: u64, job_type: &'static str, job_outcome: &'static str| {
+                    assert_histogram_count!(
+                        "apollo.router.compute_jobs.duration",
+                        count,
+                        "job.type" = job_type,
+                        "job.outcome" = job_outcome
+                    );
+                };
 
-        {
-            let _job_watcher = JobWatcher::new(ComputeJobType::Introspection);
-        }
-        check_histogram_count(1, "introspection", "abandoned");
-
-        {
-            let mut job_watcher = JobWatcher::new(ComputeJobType::QueryParsing);
-            job_watcher.outcome = Outcome::ExecutedOk;
-        }
-        check_histogram_count(1, "query_parsing", "executed_ok");
+            {
+                let _job_watcher = JobWatcher::new(ComputeJobType::Introspection);
+            }
+            check_histogram_count(1, "introspection", "abandoned");
 
-        for count in 1..5 {
             {
-                let mut job_watcher = JobWatcher::new(ComputeJobType::QueryPlanning);
-                job_watcher.outcome = Outcome::RejectedQueueFull;
+                let mut job_watcher = JobWatcher::new(ComputeJobType::QueryParsing);
+                job_watcher.outcome = Outcome::ExecutedOk;
+            }
+            check_histogram_count(1, "query_parsing", "executed_ok");
+
+            for count in 1..5 {
+                {
+                    let mut job_watcher = JobWatcher::new(ComputeJobType::QueryPlanning);
+                    job_watcher.outcome = Outcome::RejectedQueueFull;
+                }
+                check_histogram_count(count, "query_planning", "rejected_queue_full");
             }
-            check_histogram_count(count, "query_planning", "rejected_queue_full");
         }
+        .with_metrics()
+        .await;
     }
 }

apollo-router/src/compute_job/mod.rs

@@ -314,7 +314,7 @@ pub(crate) fn create_queue_size_gauge() -> ObservableGauge<u64> {
             "Number of computation jobs (parsing, planning, …) waiting to be scheduled",
         )
         .with_callback(move |m| m.observe(queue().queued_count() as u64, &[]))
-        .init()
+        .build()
 }
 
 #[cfg(test)]

apollo-router/src/compute_job/snapshots/apollo_router__compute_job__tests__observability@logs.snap

@@ -1,6 +1,7 @@
 ---
 source: apollo-router/src/compute_job/mod.rs
 expression: yaml
+snapshot_kind: text
 ---
 - fields: {}
   level: INFO

apollo-router/src/configuration/metrics.rs

@@ -591,7 +591,7 @@ impl From<InstrumentData> for Metrics {
                         .with_callback(move |observer| {
                             observer.observe(value, &attributes);
                         })
-                        .init()
+                        .build()
                 })
                 .collect(),
         }
@@ -615,17 +615,24 @@ mod test {
     #[test]
     fn test_metrics() {
         for file_name in Asset::iter() {
-            let source = Asset::get(&file_name).expect("test file must exist");
-            let input = std::str::from_utf8(&source.data)
-                .expect("expected utf8")
-                .to_string();
-            let yaml = &serde_yaml::from_str::<serde_json::Value>(&input)
-                .expect("config must be valid yaml");
-
-            let mut data = InstrumentData::default();
-            data.populate_config_instruments(yaml);
-            let _metrics: Metrics = data.into();
-            assert_non_zero_metrics_snapshot!(file_name);
+            let file_name = file_name.to_string();
+            // Spawn a new thread (and therefore meter provider) per file so metrics don't carry
+            // over to next iteration.
+            std::thread::spawn(move || {
+                let source = Asset::get(&file_name).expect("test file must exist");
+                let input = std::str::from_utf8(&source.data)
+                    .expect("expected utf8")
+                    .to_string();
+                let yaml = &serde_yaml::from_str::<serde_json::Value>(&input)
+                    .expect("config must be valid yaml");
+
+                let mut data = InstrumentData::default();
+                data.populate_config_instruments(yaml);
+                let _metrics: Metrics = data.into();
+                assert_non_zero_metrics_snapshot!(file_name);
+            })
+            .join()
+            .expect("metrics test thread panicked")
         }
     }
 

apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__schema_generation.snap

@@ -1,6 +1,7 @@
 ---
 source: apollo-router/src/configuration/tests.rs
 expression: "&schema"
+snapshot_kind: text
 ---
 {
   "$schema": "http://json-schema.org/draft-07/schema#",

apollo-router/src/executable.rs

@@ -441,7 +441,6 @@ impl Executable {
         if apollo_telemetry_initialized {
             // We should be good to shutdown OpenTelemetry now as the router should have finished everything.
             tokio::task::spawn_blocking(move || {
-                opentelemetry::global::shutdown_tracer_provider();
                 meter_provider_internal().shutdown();
             })
             .await?;

apollo-router/src/logging/mod.rs

@@ -68,7 +68,7 @@ pub(crate) mod test {
             } else {
                 let parsed_log: Vec<Value> = log
                     .lines()
-                    .map(|line| {
+                    .filter_map(|line| {
                         let mut line: serde_json::Value = serde_json::from_str(line).unwrap();
                         // move the message field to the top level
                         let fields = line
@@ -78,11 +78,19 @@ pub(crate) mod test {
                             .unwrap()
                             .as_object_mut()
                             .unwrap();
+                        // Remove noisy OTel logs when meter provider is dropped
+                        if matches!(
+                            fields.get("name").and_then(|v| v.as_str()),
+                            Some("MeterProvider.Drop")
+                        ) {
+                            return None;
+                        }
+
                         let message = fields.remove("message").unwrap_or_default();
                         line.as_object_mut()
                             .unwrap()
                             .insert("message".to_string(), message);
-                        line
+                        Some(line)
                     })
                     .collect();
                 serde_json::json!(parsed_log)