Upgrade opentelemetry libs to 0.29

Author: svix-jplatte

server/Cargo.lock

@@ -58,11 +58,11 @@ lapin = "3.7.2"
 num_enum = "0.7.2"
 omniqueue = { git = "https://github.com/svix/omniqueue-rs", rev = "4b1f3a056a0391c2d18bb85cd16a75ee9eddd640", default-features = false, features = ["in_memory", "rabbitmq-with-message-ids", "redis_cluster", "redis_sentinel", "beta"] }
 openssl = "0.10.72"
-opentelemetry = { version = "0.26.0", features = ["metrics"] }
-opentelemetry-http = "0.26.0"
-opentelemetry-otlp = { version = "0.26.0", features = ["metrics"] }
-opentelemetry-semantic-conventions = "0.26.0"
-opentelemetry_sdk = { version = "0.26.0", features = ["rt-tokio"] }
+opentelemetry = { version = "0.29.0", features = ["metrics"] }
+opentelemetry-http = "0.29.0"
+opentelemetry-otlp = { version = "0.29.0", default-features = false, features = ["grpc-tonic", "internal-logs", "metrics", "trace"] }
+opentelemetry-semantic-conventions = "0.29.0"
+opentelemetry_sdk = { version = "0.29.0", features = ["rt-tokio", "experimental_metrics_periodicreader_with_async_runtime", "experimental_trace_batch_span_processor_with_async_runtime"] }
 rand = "0.8.5"
 redis = { version = "0.32.5", features = ["tokio-comp", "tokio-native-tls-comp", "streams", "cluster-async", "tcp_nodelay", "connection-manager", "sentinel"] }
 regex = "1.5.5"
@@ -84,7 +84,7 @@ tokio-util = "0.7"
 tower = "0.5.1"
 tower-http = { version = "0.6.1", features = ["trace", "cors", "normalize-path", "request-id"] }
 tracing = "0.1.35"
-tracing-opentelemetry = "0.27.0"
+tracing-opentelemetry = "0.30.0"
 tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
 url = { version = "2.2.2", features = ["serde"] }
 urlencoding = "2.1.2"

server/svix-server/Cargo.toml

@@ -8,9 +8,13 @@ use std::{borrow::Cow, sync::LazyLock, time::Duration};
 
 use aide::axum::ApiRouter;
 use cfg::ConfigurationInner;
-use opentelemetry::trace::TracerProvider as _;
-use opentelemetry_otlp::WithExportConfig;
-use opentelemetry_sdk::{metrics::SdkMeterProvider, runtime::Tokio};
+use opentelemetry::{InstrumentationScope, trace::TracerProvider as _};
+use opentelemetry_otlp::WithExportConfig as _;
+use opentelemetry_sdk::{
+    metrics::{SdkMeterProvider, periodic_reader_with_async_runtime::PeriodicReader},
+    runtime,
+    trace::{SdkTracerProvider, span_processor_with_async_runtime::BatchSpanProcessor},
+};
 use queue::TaskQueueProducer;
 use redis::RedisManager;
 use sea_orm::DatabaseConnection;
@@ -100,7 +104,7 @@ async fn graceful_shutdown_handler() {
 }
 
 pub async fn run(cfg: Configuration) {
-    let _metrics = setup_metrics(&cfg);
+    setup_metrics(&cfg);
     run_with_prefix(cfg.queue_prefix.clone(), cfg, None).await
 }
 
@@ -246,7 +250,11 @@ pub async fn run_with_prefix(
 pub fn setup_tracing(
     cfg: &ConfigurationInner,
     for_test: bool,
-) -> (tracing::Dispatch, sentry::ClientInitGuard) {
+) -> (
+    tracing::Dispatch,
+    sentry::ClientInitGuard,
+    Option<SdkTracerProvider>,
+) {
     let filter_directives = std::env::var("RUST_LOG").unwrap_or_else(|e| {
         if let std::env::VarError::NotUnicode(_) = e {
             eprintln!("RUST_LOG environment variable has non-utf8 contents, ignoring!");
@@ -265,45 +273,47 @@ pub fn setup_tracing(
         var.join(",")
     });
 
-    let otel_layer = cfg.opentelemetry_address.as_ref().map(|addr| {
+    let mapped = cfg.opentelemetry_address.as_ref().map(|addr| {
         // Configure the OpenTelemetry tracing layer
         opentelemetry::global::set_text_map_propagator(
             opentelemetry_sdk::propagation::TraceContextPropagator::new(),
         );
 
-        let exporter = opentelemetry_otlp::new_exporter()
-            .tonic()
-            .with_endpoint(addr);
-
-        let provider = opentelemetry_otlp::new_pipeline()
-            .tracing()
-            .with_exporter(exporter)
-            .with_trace_config(
-                opentelemetry_sdk::trace::Config::default()
-                    .with_sampler(
-                        cfg.opentelemetry_sample_ratio
-                            .map(opentelemetry_sdk::trace::Sampler::TraceIdRatioBased)
-                            .unwrap_or(opentelemetry_sdk::trace::Sampler::AlwaysOn),
-                    )
-                    .with_resource(opentelemetry_sdk::Resource::new(vec![
-                        opentelemetry::KeyValue::new(
-                            "service.name",
-                            cfg.opentelemetry_service_name.clone(),
-                        ),
-                    ])),
-            )
-            .install_batch(Tokio)
+        let exporter = opentelemetry_otlp::SpanExporter::builder()
+            .with_tonic()
+            .with_endpoint(addr)
+            .build()
             .unwrap();
 
-        // Based on the private `build_batch_with_exporter` method from opentelemetry-otlp
-        let tracer = provider
-            .tracer_builder("opentelemetry-otlp")
-            .with_schema_url(opentelemetry_semantic_conventions::SCHEMA_URL)
+        let provider = SdkTracerProvider::builder()
+            .with_span_processor(BatchSpanProcessor::builder(exporter, runtime::Tokio).build())
+            .with_sampler(
+                cfg.opentelemetry_sample_ratio
+                    .map(opentelemetry_sdk::trace::Sampler::TraceIdRatioBased)
+                    .unwrap_or(opentelemetry_sdk::trace::Sampler::AlwaysOn),
+            )
+            .with_resource(
+                opentelemetry_sdk::Resource::builder()
+                    .with_service_name(cfg.opentelemetry_service_name.clone())
+                    .build(),
+            )
             .build();
 
-        tracing_opentelemetry::layer().with_tracer(tracer)
+        // Based on the private `build_batch_with_exporter` method from opentelemetry-otlp
+        let layer = tracing_opentelemetry::layer().with_tracer(
+            provider.tracer_with_scope(
+                InstrumentationScope::builder("opentelemetry-otlp")
+                    .with_schema_url(opentelemetry_semantic_conventions::SCHEMA_URL)
+                    .build(),
+            ),
+        );
+
+        _ = opentelemetry::global::set_tracer_provider(provider.clone());
+        (layer, provider)
     });
 
+    let (otel_layer, otel_tracer_provider) = mapped.unzip();
+
     let sentry_guard = sentry::init(sentry::ClientOptions {
         dsn: cfg.sentry_dsn.clone(),
         environment: Some(Cow::Owned(cfg.environment.to_string())),
@@ -336,40 +346,46 @@ pub fn setup_tracing(
         }
     };
 
-    let registry = tracing_subscriber::Registry::default()
+    let dispatch = tracing_subscriber::Registry::default()
         .with(otel_layer)
         .with(sentry_layer)
         .with(stdout_layer)
         .with(tracing_subscriber::EnvFilter::new(filter_directives))
         .into();
 
-    (registry, sentry_guard)
+    (dispatch, sentry_guard, otel_tracer_provider)
 }
 
-pub fn setup_metrics(cfg: &ConfigurationInner) -> Option<SdkMeterProvider> {
-    cfg.opentelemetry_address.as_ref().map(|addr| {
-        let exporter = opentelemetry_otlp::new_exporter()
-            .tonic()
-            .with_endpoint(addr);
-
-        opentelemetry_otlp::new_pipeline()
-            .metrics(Tokio)
-            .with_delta_temporality()
-            .with_exporter(exporter)
-            .with_resource(opentelemetry_sdk::Resource::new(vec![
-                opentelemetry::KeyValue::new(
-                    "service.name",
-                    cfg.opentelemetry_service_name.clone(),
-                ),
-                opentelemetry::KeyValue::new("instance_id", INSTANCE_ID.to_owned()),
-                opentelemetry::KeyValue::new(
-                    "service.version",
-                    option_env!("GITHUB_SHA").unwrap_or("unknown"),
-                ),
-            ]))
+pub fn setup_metrics(cfg: &ConfigurationInner) {
+    if let Some(addr) = &cfg.opentelemetry_address {
+        let exporter = opentelemetry_otlp::MetricExporter::builder()
+            .with_tonic()
+            .with_endpoint(addr)
+            .with_temporality(opentelemetry_sdk::metrics::Temporality::Delta)
             .build()
-            .unwrap()
-    })
+            .unwrap();
+
+        let reader = PeriodicReader::builder(exporter, runtime::Tokio).build();
+
+        let provider = SdkMeterProvider::builder()
+            .with_reader(reader)
+            .with_resource(
+                opentelemetry_sdk::Resource::builder()
+                    .with_service_name(cfg.opentelemetry_service_name.clone())
+                    .with_attribute(opentelemetry::KeyValue::new(
+                        "instance_id",
+                        INSTANCE_ID.as_str(),
+                    ))
+                    .with_attribute(opentelemetry::KeyValue::new(
+                        "service.version",
+                        option_env!("GITHUB_SHA").unwrap_or("unknown"),
+                    ))
+                    .build(),
+            )
+            .build();
+
+        opentelemetry::global::set_meter_provider(provider);
+    }
 }
 
 pub fn setup_tracing_for_tests() {

server/svix-server/src/lib.rs

@@ -141,7 +141,8 @@ async fn main() -> anyhow::Result<()> {
 
     let cfg = cfg::load()?;
 
-    let (tracing_subscriber, _guard) = setup_tracing(&cfg, /* for_test = */ false);
+    let (tracing_subscriber, _guard, otel_tracer_provider) =
+        setup_tracing(&cfg, /* for_test = */ false);
     tracing_subscriber.init();
 
     if let Some(wait_for_seconds) = args.wait_for {
@@ -232,6 +233,12 @@ async fn main() -> anyhow::Result<()> {
         }
     };
 
-    opentelemetry::global::shutdown_tracer_provider();
+    if let Some(provider) = otel_tracer_provider {
+        _ = tokio::task::spawn_blocking(move || {
+            _ = provider.shutdown();
+        })
+        .await;
+    }
+
     Ok(())
 }

server/svix-server/src/main.rs

@@ -1,13 +1,3 @@
 mod redis;
 
-pub fn init_metric<T, E: std::fmt::Debug>(result: Result<T, E>) -> Option<T> {
-    match result {
-        Ok(t) => Some(t),
-        Err(e) => {
-            tracing::error!(error = ?e, "Failed to initialize metric");
-            None
-        }
-    }
-}
-
 pub use self::redis::{RedisQueueMetrics, RedisQueueType};