XXX add basic otel tracing

Author: nshalman

dropshot/Cargo.toml

@@ -48,6 +48,11 @@ tokio-rustls = "0.25.0"
 toml = "0.8.19"
 waitgroup = "0.1.2"
 
+opentelemetry = { version = "0.27", optional = true }
+opentelemetry-http = { version = "0.27", features = ["hyper"], optional = true }
+opentelemetry-semantic-conventions = { version = "0.27", optional = true }
+tracing = { version = "0.1", optional = true }
+
 [dependencies.chrono]
 version = "0.4.38"
 features = [ "serde", "std", "clock" ]
@@ -136,6 +141,8 @@ version_check = "0.9.5"
 [features]
 usdt-probes = ["usdt/asm"]
 internal-docs = ["simple-mermaid"]
+otel-tracing = ["opentelemetry", "opentelemetry-http", "opentelemetry-semantic-conventions"]
+tokio-tracing = ["tracing"]
 
 [package.metadata.docs.rs]
 features = ["internal-docs"]

dropshot/src/handler.rs

@@ -86,6 +86,8 @@ pub struct RequestContext<Context: ServerContext> {
     pub log: Logger,
     /// basic request information (method, URI, etc.)
     pub request: RequestInfo,
+    #[cfg(feature = "otel-tracing")]
+    pub span_context: opentelemetry::trace::SpanContext,
 }
 
 // This is deliberately as close to compatible with `hyper::Request` as
@@ -377,6 +379,19 @@ where
     }
 }
 
+impl std::fmt::Display for HandlerError {
+    fn fmt(&self, f: &mut Formatter) -> FmtResult {
+        write!(
+            f,
+            "{}",
+            match self {
+                Self::Handler { ref message, .. } => message,
+                Self::Dropshot(ref e) => &e.external_message,
+            }
+        )
+    }
+}
+
 /// An error type which can be converted into an HTTP response.
 ///
 /// The error types returned by handlers must implement this trait, so that a

dropshot/src/lib.rs

@@ -834,6 +834,8 @@ mod from_map;
 mod handler;
 mod http_util;
 mod logging;
+#[cfg(feature = "otel-tracing")]
+mod otel;
 mod pagination;
 mod router;
 mod schema_util;

dropshot/src/otel.rs

@@ -0,0 +1,104 @@
+// Copyright 2024 Oxide Computer Company
+//! Opentelemetry tracing support
+//!
+//! Fields that we want to produce to provide comparable
+//! functionality to reqwest-tracing[1]:
+//!
+//! - http.request.method
+//! - url.scheme
+//! - server.address
+//! - server.port
+//! - otel.kind
+//! - otel.name
+//! - otel.status_code
+//! - user_agent.original
+//! - http.response.status_code
+//! - error.message
+//! - error.cause_chain
+//!
+//! [1] <https://docs.rs/reqwest-tracing/0.5.4/reqwest_tracing/macro.reqwest_otel_span.html>
+
+use opentelemetry::{global, trace::Span, trace::TracerProvider, trace::Tracer};
+use opentelemetry_http::HeaderExtractor;
+use opentelemetry_semantic_conventions::trace;
+
+// - http.request.method
+// - url.scheme
+// - server.address
+// - server.port
+// - otel.kind
+// - otel.name
+// - otel.status_code
+// - user_agent.original
+#[derive(Debug, Clone, serde::Serialize)]
+pub(crate) struct RequestInfo {
+    pub id: String,
+    pub local_addr: std::net::SocketAddr,
+    pub remote_addr: std::net::SocketAddr,
+    pub method: String,
+    pub path: String,
+    pub query: Option<String>,
+}
+
+#[derive(Debug, Clone, serde::Serialize)]
+pub(crate) struct ResponseInfo {
+    pub id: String,
+    pub local_addr: std::net::SocketAddr,
+    pub remote_addr: std::net::SocketAddr,
+    pub status_code: u16,
+    pub message: String,
+}
+
+fn extract_context_from_request(
+    request: &hyper::Request<hyper::body::Incoming>,
+) -> opentelemetry::Context {
+    global::get_text_map_propagator(|propagator| {
+        propagator.extract(&HeaderExtractor(request.headers()))
+    })
+}
+
+pub fn create_request_span(
+    request: &hyper::Request<hyper::body::Incoming>,
+) -> opentelemetry::global::BoxedSpan {
+    let tracer_provider = global::tracer_provider();
+    let scope = opentelemetry::InstrumentationScope::builder("dropshot_tracing")
+        .with_version(env!("CARGO_PKG_VERSION"))
+        .with_schema_url("https://opentelemetry.io/schemas/1.17.0")
+        .build();
+    let tracer = tracer_provider.tracer_with_scope(scope);
+    let parent_cx = extract_context_from_request(&request);
+    tracer
+        .span_builder("dropshot_request") //XXX Fixme
+        .with_kind(opentelemetry::trace::SpanKind::Server)
+        .start_with_context(&tracer, &parent_cx)
+}
+
+pub trait TraceDropshot {
+    fn trace_request(&mut self, request: RequestInfo);
+    fn trace_response(&mut self, response: ResponseInfo);
+}
+
+impl TraceDropshot for opentelemetry::global::BoxedSpan {
+    fn trace_request(&mut self, request: RequestInfo) {
+        self.set_attributes(vec![
+            opentelemetry::KeyValue::new("http.id".to_string(), request.id),
+            opentelemetry::KeyValue::new(
+                "http.method".to_string(),
+                request.method,
+            ),
+            opentelemetry::KeyValue::new("http.path".to_string(), request.path),
+        ]);
+    }
+    fn trace_response(&mut self, response: ResponseInfo) {
+        self.set_attributes(vec![
+            opentelemetry::KeyValue::new(
+                trace::HTTP_RESPONSE_STATUS_CODE,
+                i64::from(response.status_code),
+            ),
+            opentelemetry::KeyValue::new(
+                "http.message".to_string(),
+                response.message,
+            ),
+        ]);
+    }
+}

dropshot/src/server.rs

@@ -13,6 +13,11 @@ use super::router::HttpRouter;
 use super::versioning::VersionPolicy;
 use super::ProbeRegistration;
 
+#[cfg(feature = "otel-tracing")]
+use crate::{otel, otel::TraceDropshot};
+#[cfg(feature = "otel-tracing")]
+use opentelemetry::trace::Span;
+
 use async_stream::stream;
 use debug_ignore::DebugIgnore;
 use futures::future::{
@@ -730,6 +735,10 @@ impl<C: ServerContext> FusedFuture for HttpServer<C> {
 /// invoked by Hyper when a new request is received.  This function returns a
 /// Result that either represents a valid HTTP response or an error (which will
 /// also get turned into an HTTP response).
+#[cfg_attr(feature = "tokio-tracing", tracing::instrument(
+    err,
+    skip_all,
+))]
 async fn http_request_handle_wrap<C: ServerContext>(
     server: Arc<DropshotState<C>>,
     remote_addr: SocketAddr,
@@ -774,6 +783,18 @@ async fn http_request_handle_wrap<C: ServerContext>(
         }
     }
 
+    #[cfg(feature = "otel-tracing")]
+    let mut span = otel::create_request_span(&request);
+    #[cfg(feature = "otel-tracing")]
+    span.trace_request(crate::otel::RequestInfo {
+        id: request_id.clone(),
+        local_addr: server.local_addr,
+        remote_addr,
+        method: request.method().to_string(),
+        path: request.uri().path().to_string(),
+        query: request.uri().query().map(|x| x.to_string()),
+    });
+
     trace!(request_log, "incoming request");
     #[cfg(feature = "usdt-probes")]
     probes::request__start!(|| {
@@ -790,9 +811,12 @@ async fn http_request_handle_wrap<C: ServerContext>(
 
     // Copy local address to report later during the finish probe, as the
     // server is passed by value to the request handler function.
-    #[cfg(feature = "usdt-probes")]
+    #[cfg(any(feature = "usdt-probes", feature = "otel-tracing"))]
     let local_addr = server.local_addr;
 
+    #[cfg(feature = "otel-tracing")]
+    let span_context = span.span_context().clone();
+
     // In the case the client disconnects early, the scopeguard allows us
     // to perform extra housekeeping before this task is dropped.
     let on_disconnect = guard((), |_| {
@@ -802,6 +826,18 @@ async fn http_request_handle_wrap<C: ServerContext>(
             "latency_us" => latency_us,
         );
 
+        #[cfg(feature = "otel-tracing")]
+        span.trace_response(crate::otel::ResponseInfo {
+            id: request_id.clone(),
+            local_addr,
+            remote_addr,
+            // 499 is a non-standard code popularized by nginx to mean "client disconnected".
+            status_code: 499,
+            message: String::from(
+                "client disconnected before response returned",
+            ),
+        });
+
         #[cfg(feature = "usdt-probes")]
         probes::request__done!(|| {
             crate::dtrace::ResponseInfo {
@@ -823,6 +859,8 @@ async fn http_request_handle_wrap<C: ServerContext>(
         &request_id,
         request_log.new(o!()),
         remote_addr,
+        #[cfg(feature = "otel-tracing")]
+        span_context,
     )
     .await;
 
@@ -838,6 +876,17 @@ async fn http_request_handle_wrap<C: ServerContext>(
                 let message_external = error.external_message();
                 let message_internal = error.internal_message();
 
+                #[cfg(feature = "otel-tracing")]
+                span.trace_response(crate::otel::ResponseInfo {
+                    id: request_id.clone(),
+                    local_addr,
+                    remote_addr,
+                    status_code: status.as_u16(),
+                    message: message_external
+                        .cloned()
+                        .unwrap_or_else(|| message_internal.clone()),
+                });
+
                 #[cfg(feature = "usdt-probes")]
                 probes::request__done!(|| {
                     crate::dtrace::ResponseInfo {
@@ -869,6 +918,15 @@ async fn http_request_handle_wrap<C: ServerContext>(
                 "latency_us" => latency_us,
             );
 
+            #[cfg(feature = "otel-tracing")]
+            span.trace_response(crate::otel::ResponseInfo {
+                id: request_id.parse().unwrap(),
+                local_addr,
+                remote_addr,
+                status_code: response.status().as_u16(),
+                message: "".to_string(),
+            });
+
             #[cfg(feature = "usdt-probes")]
             probes::request__done!(|| {
                 crate::dtrace::ResponseInfo {
@@ -887,12 +945,26 @@ async fn http_request_handle_wrap<C: ServerContext>(
     Ok(response)
 }
 
+#[cfg_attr(feature = "tokio-tracing", tracing::instrument(
+    err,
+    skip_all,
+    fields(
+        http.method = request.method().as_str().to_string(),
+        http.uri = request.uri().to_string(),
+        http.version = format!("{:#?}",request.version()),
+        http.headers.accept = format!("{:#?}", request.headers()["accept"]),
+        http.headers.host = format!("{:#?}", request.headers()["host"]),
+        //http.headers.user_agent = format!("{:#?}", request.headers()["user-agent"]),
+    ),
+))]
 async fn http_request_handle<C: ServerContext>(
     server: Arc<DropshotState<C>>,
     request: Request<hyper::body::Incoming>,
     request_id: &str,
     request_log: Logger,
     remote_addr: std::net::SocketAddr,
+    #[cfg(feature = "otel-tracing")]
+    span_context: opentelemetry::trace::SpanContext,
 ) -> Result<Response<Body>, HandlerError> {
     // TODO-hardening: is it correct to (and do we correctly) read the entire
     // request body even if we decide it's too large and are going to send a 400
@@ -916,6 +988,8 @@ async fn http_request_handle<C: ServerContext>(
         endpoint: lookup_result.endpoint,
         request_id: request_id.to_string(),
         log: request_log,
+        #[cfg(feature = "otel-tracing")]
+        span_context: span_context,
     };
     let handler = lookup_result.handler;