Update the rest of the OTEL deps and use the new APIs

Author: sandhose

Cargo.lock

@@ -5,7 +5,6 @@ resolver = "2"
 
 # Updated in the CI with a `sed` command
 package.version = "0.12.0"
-
 package.license = "AGPL-3.0-only"
 package.authors = ["Element Backend Team"]
 package.edition = "2021"
@@ -337,15 +336,15 @@ version = "0.3.19"
 
 # OpenTelemetry
 [workspace.dependencies.opentelemetry]
-version = "0.24.0"
+version = "0.27.1"
 features = ["trace", "metrics"]
 [workspace.dependencies.opentelemetry-http]
-version = "0.13.0"
+version = "0.27.0"
 features = ["reqwest"]
 [workspace.dependencies.opentelemetry-semantic-conventions]
-version = "0.16.0"
+version = "0.27.0"
 [workspace.dependencies.tracing-opentelemetry]
-version = "0.25.0"
+version = "0.28.0"
 default-features = false
 
 # URL manipulation

Cargo.toml

@@ -25,10 +25,7 @@ use mas_router::UrlBuilder;
 use mas_storage::{BoxClock, BoxRepository, BoxRng, SystemClock};
 use mas_storage_pg::PgRepository;
 use mas_templates::Templates;
-use opentelemetry::{
-    metrics::{Histogram, MetricsError},
-    KeyValue,
-};
+use opentelemetry::{metrics::Histogram, KeyValue};
 use rand::SeedableRng;
 use sqlx::PgPool;
 
@@ -55,50 +52,45 @@ pub struct AppState {
 
 impl AppState {
     /// Init the metrics for the app state.
-    ///
-    /// # Errors
-    ///
-    /// Returns an error if the metrics could not be initialized.
-    pub fn init_metrics(&mut self) -> Result<(), MetricsError> {
+    pub fn init_metrics(&mut self) {
         // XXX: do we want to put that somewhere else?
-        let meter = opentelemetry::global::meter_with_version(
-            env!("CARGO_PKG_NAME"),
-            Some(env!("CARGO_PKG_VERSION")),
-            Some(opentelemetry_semantic_conventions::SCHEMA_URL),
-            None,
-        );
+        let scope = opentelemetry::InstrumentationScope::builder(env!("CARGO_PKG_NAME"))
+            .with_version(env!("CARGO_PKG_VERSION"))
+            .with_schema_url(opentelemetry_semantic_conventions::SCHEMA_URL)
+            .build();
+        let meter = opentelemetry::global::meter_with_scope(scope);
+
         let pool = self.pool.clone();
-        let usage = meter
+        meter
             .i64_observable_up_down_counter("db.connections.usage")
             .with_description("The number of connections that are currently in `state` described by the state attribute.")
             .with_unit("{connection}")
-            .init();
+            .with_callback(move |instrument| {
+                let idle = u32::try_from(pool.num_idle()).unwrap_or(u32::MAX);
+                let used = pool.size() - idle;
+                instrument.observe(i64::from(idle), &[KeyValue::new("state", "idle")]);
+                instrument.observe(i64::from(used), &[KeyValue::new("state", "used")]);
+            })
+            .build();
 
-        let max = meter
+        let pool = self.pool.clone();
+        meter
             .i64_observable_up_down_counter("db.connections.max")
             .with_description("The maximum number of open connections allowed.")
             .with_unit("{connection}")
-            .init();
-
-        // Observe the number of active and idle connections in the pool
-        meter.register_callback(&[usage.as_any(), max.as_any()], move |observer| {
-            let idle = u32::try_from(pool.num_idle()).unwrap_or(u32::MAX);
-            let used = pool.size() - idle;
-            let max_conn = pool.options().get_max_connections();
-            observer.observe_i64(&usage, i64::from(idle), &[KeyValue::new("state", "idle")]);
-            observer.observe_i64(&usage, i64::from(used), &[KeyValue::new("state", "used")]);
-            observer.observe_i64(&max, i64::from(max_conn), &[]);
-        })?;
+            .with_callback(move |instrument| {
+                let max_conn = pool.options().get_max_connections();
+                instrument.observe(i64::from(max_conn), &[]);
+            })
+            .build();
 
         // Track the connection acquisition time
         let histogram = meter
             .u64_histogram("db.client.connections.create_time")
             .with_description("The time it took to create a new connection.")
             .with_unit("ms")
-            .init();
+            .build();
         self.conn_acquisition_histogram = Some(histogram);
-
-        Ok(())
     }
 
     /// Init the metadata cache.

crates/cli/src/app_state.rs

@@ -228,7 +228,7 @@ impl Options {
                 limiter,
                 conn_acquisition_histogram: None,
             };
-            s.init_metrics()?;
+            s.init_metrics();
             // XXX: this might panic
             s.init_metadata_cache().await;
             s

crates/cli/src/commands/server.rs

@@ -248,14 +248,14 @@ pub fn build_router(
     router
         .layer(
             InFlightCounterLayer::new("http.server.active_requests").on_request((
-                name.map(|name| MAS_LISTENER_NAME.string(name.to_owned())),
+                name.map(|name| KeyValue::new(MAS_LISTENER_NAME, name.to_owned())),
                 metrics_attributes_fn(on_http_request_labels),
             )),
         )
         .layer(
             DurationRecorderLayer::new("http.server.duration")
                 .on_request((
-                    name.map(|name| MAS_LISTENER_NAME.string(name.to_owned())),
+                    name.map(|name| KeyValue::new(MAS_LISTENER_NAME, name.to_owned())),
                     metrics_attributes_fn(on_http_request_labels),
                 ))
                 .on_response_fn(on_http_response_labels),

crates/cli/src/server.rs

@@ -18,16 +18,13 @@ use opentelemetry::{
     global,
     propagation::{TextMapCompositePropagator, TextMapPropagator},
     trace::TracerProvider as _,
-    KeyValue,
+    InstrumentationScope, KeyValue,
 };
-use opentelemetry_otlp::MetricsExporterBuilder;
+use opentelemetry_otlp::{WithExportConfig, WithHttpConfig};
 use opentelemetry_prometheus::PrometheusExporter;
 use opentelemetry_sdk::{
     self,
-    metrics::{
-        reader::{DefaultAggregationSelector, DefaultTemporalitySelector},
-        ManualReader, PeriodicReader, SdkMeterProvider,
-    },
+    metrics::{ManualReader, PeriodicReader, SdkMeterProvider},
     propagation::{BaggagePropagator, TraceContextPropagator},
     trace::{Sampler, Tracer, TracerProvider},
     Resource,
@@ -41,16 +38,6 @@ static METER_PROVIDER: OnceCell<SdkMeterProvider> = OnceCell::const_new();
 static PROMETHEUS_REGISTRY: OnceCell<Registry> = OnceCell::const_new();
 
 pub fn setup(config: &TelemetryConfig) -> anyhow::Result<Option<Tracer>> {
-    global::set_error_handler(|e| {
-        // Don't log the propagation errors, else we'll log an error on each request if
-        // the propagation errors aren't there
-        if matches!(e, opentelemetry::global::Error::Propagation(_)) {
-            return;
-        }
-
-        tracing::error!(error = &e as &dyn std::error::Error);
-    })?;
-
     let propagator = propagator(&config.tracing.propagators);
 
     // The CORS filter needs to know what headers it should whitelist for
@@ -96,22 +83,22 @@ fn stdout_tracer_provider() -> TracerProvider {
 }
 
 fn otlp_tracer_provider(endpoint: Option<&Url>) -> anyhow::Result<TracerProvider> {
-    use opentelemetry_otlp::WithExportConfig;
-
-    let mut exporter = opentelemetry_otlp::new_exporter()
-        .http()
+    let mut exporter = opentelemetry_otlp::SpanExporter::builder()
+        .with_http()
         .with_http_client(mas_http::reqwest_client());
     if let Some(endpoint) = endpoint {
         exporter = exporter.with_endpoint(endpoint.to_string());
     }
-
-    let tracer = opentelemetry_otlp::new_pipeline()
-        .tracing()
-        .with_exporter(exporter)
-        .with_trace_config(trace_config())
-        .install_batch(opentelemetry_sdk::runtime::Tokio)
+    let exporter = exporter
+        .build()
         .context("Failed to configure OTLP trace exporter")?;
 
+    let tracer = opentelemetry_sdk::trace::TracerProvider::builder()
+        .with_batch_exporter(exporter, opentelemetry_sdk::runtime::Tokio)
+        .with_resource(resource())
+        .with_sampler(Sampler::AlwaysOn)
+        .build();
+
     Ok(tracer)
 }
 
@@ -122,37 +109,35 @@ fn tracer(config: &TracingConfig) -> anyhow::Result<Option<Tracer>> {
         TracingExporterKind::Otlp => otlp_tracer_provider(config.endpoint.as_ref())?,
     };
 
-    let tracer = tracer_provider
-        .tracer_builder(env!("CARGO_PKG_NAME"))
+    let scope = InstrumentationScope::builder(env!("CARGO_PKG_NAME"))
         .with_version(env!("CARGO_PKG_VERSION"))
         .with_schema_url(semcov::SCHEMA_URL)
         .build();
 
+    let tracer = tracer_provider.tracer_with_scope(scope);
+
     global::set_tracer_provider(tracer_provider);
 
     Ok(Some(tracer))
 }
 
 fn otlp_metric_reader(endpoint: Option<&url::Url>) -> anyhow::Result<PeriodicReader> {
-    use opentelemetry_otlp::WithExportConfig;
-
-    let mut exporter = opentelemetry_otlp::new_exporter()
-        .http()
+    let mut exporter = opentelemetry_otlp::MetricExporter::builder()
+        .with_http()
         .with_http_client(mas_http::reqwest_client());
     if let Some(endpoint) = endpoint {
         exporter = exporter.with_endpoint(endpoint.to_string());
     }
+    let exporter = exporter
+        .build()
+        .context("Failed to configure OTLP metric exporter")?;
 
-    let exporter = MetricsExporterBuilder::from(exporter).build_metrics_exporter(
-        Box::new(DefaultTemporalitySelector::new()),
-        Box::new(DefaultAggregationSelector::new()),
-    )?;
-
-    Ok(PeriodicReader::builder(exporter, opentelemetry_sdk::runtime::Tokio).build())
+    let reader = PeriodicReader::builder(exporter, opentelemetry_sdk::runtime::Tokio).build();
+    Ok(reader)
 }
 
 fn stdout_metric_reader() -> PeriodicReader {
-    let exporter = opentelemetry_stdout::MetricsExporter::default();
+    let exporter = opentelemetry_stdout::MetricExporter::builder().build();
     PeriodicReader::builder(exporter, opentelemetry_sdk::runtime::Tokio).build()
 }
 
@@ -230,12 +215,6 @@ fn init_meter(config: &MetricsConfig) -> anyhow::Result<()> {
     Ok(())
 }
 
-fn trace_config() -> opentelemetry_sdk::trace::Config {
-    opentelemetry_sdk::trace::Config::default()
-        .with_resource(resource())
-        .with_sampler(Sampler::AlwaysOn)
-}
-
 fn resource() -> Resource {
     let resource = Resource::new([
         KeyValue::new(semcov::resource::SERVICE_NAME, env!("CARGO_PKG_NAME")),

crates/cli/src/telemetry.rs

@@ -10,7 +10,7 @@ use chrono::{DateTime, Utc};
 use mas_storage::{user::BrowserSessionRepository, RepositoryAccess};
 use opentelemetry::{
     metrics::{Counter, Histogram},
-    Key,
+    Key, KeyValue,
 };
 use sqlx::PgPool;
 use tokio_util::sync::CancellationToken;
@@ -48,18 +48,17 @@ pub struct Worker {
 
 impl Worker {
     pub(crate) fn new(pool: PgPool) -> Self {
-        let meter = opentelemetry::global::meter_with_version(
-            env!("CARGO_PKG_NAME"),
-            Some(env!("CARGO_PKG_VERSION")),
-            Some(opentelemetry_semantic_conventions::SCHEMA_URL),
-            None,
-        );
+        let scope = opentelemetry::InstrumentationScope::builder(env!("CARGO_PKG_NAME"))
+            .with_version(env!("CARGO_PKG_VERSION"))
+            .with_schema_url(opentelemetry_semantic_conventions::SCHEMA_URL)
+            .build();
+        let meter = opentelemetry::global::meter_with_scope(scope);
 
         let message_counter = meter
             .u64_counter("mas.activity_tracker.messages")
             .with_description("The number of messages received by the activity tracker")
             .with_unit("{messages}")
-            .init();
+            .build();
 
         // Record stuff on the counter so that the metrics are initialized
         for kind in &[
@@ -69,17 +68,20 @@ impl Worker {
         ] {
             message_counter.add(
                 0,
-                &[TYPE.string("record"), SESSION_KIND.string(kind.as_str())],
+                &[
+                    KeyValue::new(TYPE, "record"),
+                    KeyValue::new(SESSION_KIND, kind.as_str()),
+                ],
             );
         }
-        message_counter.add(0, &[TYPE.string("flush")]);
-        message_counter.add(0, &[TYPE.string("shutdown")]);
+        message_counter.add(0, &[KeyValue::new(TYPE, "flush")]);
+        message_counter.add(0, &[KeyValue::new(TYPE, "shutdown")]);
 
         let flush_time_histogram = meter
             .u64_histogram("mas.activity_tracker.flush_time")
             .with_description("The time it took to flush the activity tracker")
             .with_unit("ms")
-            .init();
+            .build();
 
         Self {
             pool,
@@ -137,7 +139,10 @@ impl Worker {
 
                     self.message_counter.add(
                         1,
-                        &[TYPE.string("record"), SESSION_KIND.string(kind.as_str())],
+                        &[
+                            KeyValue::new(TYPE, "record"),
+                            KeyValue::new(SESSION_KIND, kind.as_str()),
+                        ],
                     );
 
                     let record =
@@ -153,7 +158,7 @@ impl Worker {
                 }
 
                 Message::Flush(tx) => {
-                    self.message_counter.add(1, &[TYPE.string("flush")]);
+                    self.message_counter.add(1, &[KeyValue::new(TYPE, "flush")]);
 
                     self.flush().await;
                     let _ = tx.send(());
@@ -182,11 +187,11 @@ impl Worker {
         match res {
             Ok(()) => {
                 self.flush_time_histogram
-                    .record(duration_ms, &[RESULT.string("success")]);
+                    .record(duration_ms, &[KeyValue::new(RESULT, "success")]);
             }
             Err(e) => {
                 self.flush_time_histogram
-                    .record(duration_ms, &[RESULT.string("failure")]);
+                    .record(duration_ms, &[KeyValue::new(RESULT, "failure")]);
                 tracing::error!("Failed to flush activity tracker: {}", e);
             }
         }

crates/handlers/src/activity_tracker/worker.rs

@@ -20,10 +20,10 @@ pub use self::{
 };
 
 static METER: LazyLock<opentelemetry::metrics::Meter> = LazyLock::new(|| {
-    opentelemetry::global::meter_with_version(
-        env!("CARGO_PKG_NAME"),
-        Some(env!("CARGO_PKG_VERSION")),
-        Some(opentelemetry_semantic_conventions::SCHEMA_URL),
-        None,
-    )
+    let scope = opentelemetry::InstrumentationScope::builder(env!("CARGO_PKG_NAME"))
+        .with_version(env!("CARGO_PKG_VERSION"))
+        .with_schema_url(opentelemetry_semantic_conventions::SCHEMA_URL)
+        .build();
+
+    opentelemetry::global::meter_with_scope(scope)
 });