Run events in span scope (#838)

Author: soundofspace

rama-http/src/layer/trace/make_span.rs

@@ -3,6 +3,7 @@ use crate::header::USER_AGENT;
 use crate::opentelemetry::version_as_protocol_version;
 use rama_core::telemetry::tracing::{self, Level, Span};
 use rama_net::http::RequestContext;
+use rama_utils::str::arcstr::{ArcStr, arcstr};
 
 use super::DEFAULT_MESSAGE_LEVEL;
 
@@ -39,6 +40,7 @@ where
 pub struct DefaultMakeSpan {
     level: Level,
     include_headers: bool,
+    otel_name: ArcStr,
 }
 
 impl DefaultMakeSpan {
@@ -48,6 +50,7 @@ impl DefaultMakeSpan {
         Self {
             level: DEFAULT_MESSAGE_LEVEL,
             include_headers: false,
+            otel_name: arcstr!("request"),
         }
     }
 
@@ -67,13 +70,20 @@ impl DefaultMakeSpan {
         /// Include request headers on the [`Span`].
         ///
         /// By default headers are not included.
-        ///
-        /// [`Span`]: tracing::Span
+
         pub fn include_headers(mut self, include_headers: bool) -> Self {
             self.include_headers = include_headers;
             self
         }
     }
+
+    rama_utils::macros::generate_set_and_with! {
+        /// Set the otel.name field of this [`Span`].
+        pub fn name(mut self, otel_name: ArcStr) -> Self {
+            self.otel_name = otel_name;
+            self
+        }
+    }
 }
 
 impl Default for DefaultMakeSpan {
@@ -114,6 +124,7 @@ impl<B> MakeSpan<B> for DefaultMakeSpan {
                     tracing::span!(
                         $level,
                         "request",
+                        otel.name = self.otel_name.as_str(),
                         http.request.method = %request.method(),
                         url.full = %request.uri(),
                         url.domain = found_domain_str,
@@ -130,6 +141,7 @@ impl<B> MakeSpan<B> for DefaultMakeSpan {
                     tracing::span!(
                         $level,
                         "request",
+                        otel.name = self.otel_name.as_str(),
                         http.request.method = %request.method(),
                         url.full = %request.uri(),
                         url.domain = found_domain_str,

rama-http/src/layer/trace/on_eos.rs

@@ -90,7 +90,7 @@ impl DefaultOnEos {
 }
 
 impl OnEos for DefaultOnEos {
-    fn on_eos(self, trailers: Option<&HeaderMap>, stream_duration: Duration, _span: &Span) {
+    fn on_eos(self, trailers: Option<&HeaderMap>, stream_duration: Duration, span: &Span) {
         let stream_duration = Latency {
             unit: self.latency_unit,
             duration: stream_duration,
@@ -108,6 +108,6 @@ impl OnEos for DefaultOnEos {
             }
         });
 
-        event_dynamic_lvl!(self.level, %stream_duration, status, "end of stream");
+        span.in_scope(|| event_dynamic_lvl!(self.level, %stream_duration, status, "end of stream"));
     }
 }

rama-http/src/layer/trace/on_failure.rs

@@ -90,16 +90,18 @@ impl<FailureClass> OnFailure<FailureClass> for DefaultOnFailure
 where
     FailureClass: fmt::Display,
 {
-    fn on_failure(&self, failure_classification: FailureClass, latency: Duration, _: &Span) {
+    fn on_failure(&self, failure_classification: FailureClass, latency: Duration, span: &Span) {
         let latency = Latency {
             unit: self.latency_unit,
             duration: latency,
         };
-        event_dynamic_lvl!(
-            self.level,
-            classification = %failure_classification,
-            %latency,
-            "response failed"
-        );
+        span.in_scope(|| {
+            event_dynamic_lvl!(
+                self.level,
+                classification = %failure_classification,
+                %latency,
+                "response failed"
+            )
+        });
     }
 }

rama-http/src/layer/trace/on_request.rs

@@ -78,7 +78,7 @@ impl DefaultOnRequest {
 }
 
 impl<B> OnRequest<B> for DefaultOnRequest {
-    fn on_request(&self, _: &Request<B>, _: &Span) {
-        event_dynamic_lvl!(self.level, "started processing request");
+    fn on_request(&self, _: &Request<B>, span: &Span) {
+        span.in_scope(|| event_dynamic_lvl!(self.level, "started processing request"));
     }
 }

rama-http/src/layer/trace/on_response.rs

@@ -108,7 +108,7 @@ impl DefaultOnResponse {
 }
 
 impl<B> OnResponse<B> for DefaultOnResponse {
-    fn on_response(self, response: &Response<B>, latency: Duration, _: &Span) {
+    fn on_response(self, response: &Response<B>, latency: Duration, span: &Span) {
         let latency = Latency {
             unit: self.latency_unit,
             duration: latency,
@@ -117,13 +117,15 @@ impl<B> OnResponse<B> for DefaultOnResponse {
             .include_headers
             .then(|| tracing::field::debug(response.headers()));
 
-        event_dynamic_lvl!(
-            self.level,
-            %latency,
-            status = status(response),
-            response_headers,
-            "finished processing request"
-        );
+        span.in_scope(|| {
+            event_dynamic_lvl!(
+                self.level,
+                %latency,
+                status = status(response),
+                response_headers,
+                "finished processing request"
+            );
+        })
     }
 }