fix: otel types

examples/otel/metrics/src/main.rs

@@ -2,10 +2,7 @@ use std::net::SocketAddr;
 use tokio::net::TcpListener;
 
 use opentelemetry::{global, KeyValue};
-use opentelemetry_sdk::{
-    metrics::{self, Aggregation, Instrument, MeterProviderBuilder, Stream},
-    Resource,
-};
+use opentelemetry_sdk::{metrics::MeterProviderBuilder, Resource};
 
 use viz::{
     handlers::prometheus::{ExporterBuilder, Prometheus, Registry},
@@ -24,29 +21,13 @@ async fn main() -> Result<()> {
     println!("listening on http://{addr}");
 
     let registry = Registry::new();
-    let (exporter, controller) = {
-        (
-            ExporterBuilder::default()
-                .with_registry(registry.clone())
-                .build()
-                .map_err(Error::boxed)?,
-            metrics::new_view(
-                Instrument::new().name("http.server.duration"),
-                Stream::new().aggregation(Aggregation::ExplicitBucketHistogram {
-                    boundaries: vec![
-                        0.0, 0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1.0, 2.5, 5.0,
-                        7.5, 10.0,
-                    ],
-                    record_min_max: true,
-                }),
-            )
-            .unwrap(),
-        )
-    };
+    let exporter = ExporterBuilder::default()
+        .with_registry(registry.clone())
+        .build()
+        .map_err(Error::boxed)?;
     let provider = MeterProviderBuilder::default()
         .with_reader(exporter)
         .with_resource(Resource::new([KeyValue::new("service.name", "viz")]))
-        .with_view(controller)
         .build();
 
     global::set_meter_provider(provider.clone());
@@ -61,9 +42,9 @@ async fn main() -> Result<()> {
         println!("{e}");
     }
 
-    Ok(())
-
     // Ensure all spans have been reported
-    // global::shutdown_tracer_provider();
-    // provider.shutdown();
+    global::shutdown_tracer_provider();
+    provider.shutdown().map_err(Error::boxed)?;
+
+    Ok(())
 }

examples/otel/tracing/src/main.rs

@@ -1,21 +1,27 @@
 #![allow(clippy::unused_async)]
 
 use opentelemetry::global;
+use opentelemetry_otlp::{SpanExporter, WithExportConfig};
 use opentelemetry_sdk::{
     runtime::TokioCurrentThread,
     {propagation::TraceContextPropagator, trace::TracerProvider},
 };
 use std::net::SocketAddr;
 use tokio::net::TcpListener;
-use viz::{middleware::otel, serve, Request, Result, Router};
+use viz::{middleware::otel, serve, Error, Request, Result, Router};
 
 fn init_tracer_provider() -> TracerProvider {
     global::set_text_map_propagator(TraceContextPropagator::new());
-    opentelemetry_otlp::new_pipeline()
-        .tracing()
-        .with_exporter(opentelemetry_otlp::new_exporter().http())
-        .install_batch(TokioCurrentThread)
-        .unwrap()
+
+    let exporter = SpanExporter::builder()
+        .with_http()
+        .with_protocol(opentelemetry_otlp::Protocol::HttpBinary)
+        .build()
+        .unwrap();
+
+    TracerProvider::builder()
+        .with_batch_exporter(exporter, TokioCurrentThread)
+        .build()
 }
 
 async fn index(_: Request) -> Result<&'static str> {
@@ -33,11 +39,15 @@ async fn main() -> Result<()> {
     let app = Router::new()
         .get("/", index)
         .get("/:username", index)
-        .with(otel::tracing::Config::new(tracer_provider, None));
+        .with(otel::tracing::Config::new(tracer_provider.clone(), None));
 
     if let Err(e) = serve(listener, app).await {
         println!("{e}");
     }
 
+    // Ensure all spans have been reported
+    global::shutdown_tracer_provider();
+    tracer_provider.shutdown().map_err(Error::boxed)?;
+
     Ok(())
 }

examples/websocket-chat/src/main.rs

@@ -28,7 +28,7 @@ async fn ws(mut req: Request) -> Result<impl IntoResponse> {
 
         tokio::task::spawn(async move {
             while let Ok(msg) = rx.recv().await {
-                if ws_tx.send(Message::Text(msg)).await.is_err() {
+                if ws_tx.send(Message::Text(msg.into())).await.is_err() {
                     break;
                 }
             }

viz-core/src/middleware/otel/metrics.rs

@@ -40,25 +40,25 @@ impl Config {
                 "Measures the number of concurrent HTTP requests that are currently in-flight.",
             )
             .with_unit("{request}")
-            .init();
+            .build();
 
         let duration = meter
             .f64_histogram(HTTP_SERVER_DURATION)
             .with_description("Measures the duration of inbound HTTP requests.")
             .with_unit("s")
-            .init();
+            .build();
 
         let request_size = meter
             .u64_histogram(HTTP_SERVER_REQUEST_SIZE)
             .with_description("Measures the size of HTTP request messages (compressed).")
             .with_unit("By")
-            .init();
+            .build();
 
         let response_size = meter
             .u64_histogram(HTTP_SERVER_RESPONSE_SIZE)
             .with_description("Measures the size of HTTP request messages (compressed).")
             .with_unit("By")
-            .init();
+            .build();
 
         Self {
             active_requests,

viz-core/src/middleware/otel/tracing.rs

@@ -9,7 +9,7 @@ use opentelemetry::{
     trace::{
         FutureExt as OtelFutureExt, Span, SpanKind, Status, TraceContextExt, Tracer, TracerProvider,
     },
-    Context, KeyValue,
+    Context, InstrumentationScope, KeyValue,
 };
 use opentelemetry_semantic_conventions::trace::{
     CLIENT_ADDRESS, EXCEPTION_MESSAGE, HTTP_REQUEST_METHOD, HTTP_RESPONSE_STATUS_CODE, HTTP_ROUTE,
@@ -81,12 +81,10 @@ where
 
         let http_route = &req.route_info().pattern;
         let attributes = build_attributes(&req, http_route.as_str());
-
-        let tracer = self
-            .tracer
-            .tracer_builder(self.name.clone())
+        let scope = InstrumentationScope::builder(self.name.clone())
             .with_attributes(attributes)
             .build();
+        let tracer = self.tracer.tracer_with_scope(scope);
         let mut span = tracer.build_with_context(
             tracer
                 .span_builder(format!("{} {}", req.method(), http_route))

viz-handlers/Cargo.toml

@@ -30,7 +30,7 @@ embed = ["dep:hex", "dep:mime_guess", "dep:http-body-util", "dep:rust-embed"]
 
 prometheus = [
   "dep:http-body-util",
-  "opentelemetry/metrics",
+  "opentelemetry/internal-logs",
   "dep:opentelemetry-prometheus",
   "dep:prometheus"
 ]

viz-handlers/src/prometheus.rs

@@ -3,7 +3,7 @@
 //! [OTEL]: https://docs.rs/opentelemetry-prometheus
 
 use http_body_util::Full;
-use opentelemetry::{global::handle_error, metrics::MetricsError};
+use opentelemetry::otel_error;
 use prometheus::{Encoder, TextEncoder};
 
 use viz_core::{
@@ -40,9 +40,10 @@ impl Handler<Request> for Prometheus {
         let mut body = Vec::new();
 
         if let Err(err) = encoder.encode(&metric_families, &mut body) {
+            let error = StatusCode::INTERNAL_SERVER_ERROR;
             let text = err.to_string();
-            handle_error(MetricsError::Other(text.clone()));
-            Err((StatusCode::INTERNAL_SERVER_ERROR, text).into_error())?;
+            otel_error!(name: "prometheus_encode_failure", error_code = error.as_u16(), error = text.clone());
+            Err((error, text).into_error())?;
         }
 
         let mut res = Response::new(Full::from(body).into());