feat: add a readable span interface

Currently there is no way to read the span data without cloning it.
This causes performance issues when the span for span
processors that need to read the span data.

This commit introduces a new trait `ReadableSpan` in the sdk
implemented by SDK Spans that allows to read the span data without
cloning it.

Author: paullegranddc

examples/tracing-http-propagator/src/server.rs

@@ -15,11 +15,16 @@ use opentelemetry_sdk::{
     error::OTelSdkResult,
     logs::{LogProcessor, SdkLogRecord, SdkLoggerProvider},
     propagation::{BaggagePropagator, TraceContextPropagator},
-    trace::{SdkTracerProvider, SpanProcessor},
+    trace::{FinishedSpan, ReadableSpan, SdkTracerProvider, SpanProcessor},
 };
 use opentelemetry_semantic_conventions::trace;
 use opentelemetry_stdout::{LogExporter, SpanExporter};
-use std::{convert::Infallible, net::SocketAddr, sync::OnceLock};
+use std::{
+    collections::HashMap,
+    convert::Infallible,
+    net::SocketAddr,
+    sync::{Mutex, OnceLock},
+};
 use tokio::net::TcpListener;
 use tracing::info;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
@@ -83,6 +88,7 @@ async fn router(
         let span = tracer
             .span_builder("router")
             .with_kind(SpanKind::Server)
+            .with_attributes([KeyValue::new("http.route", req.uri().path().to_string())])
             .start_with_context(tracer, &parent_cx);
 
         info!(name = "router", message = "Dispatching request");
@@ -104,6 +110,62 @@ async fn router(
     response
 }
 
+#[derive(Debug, Default)]
+/// A custom span processor that counts concurrent requests for each route (indetified by the http.route
+/// attribute) and adds that information to the span attributes.
+struct RouteConcurrencyCounterSpanProcessor(Mutex<HashMap<opentelemetry::Key, usize>>);
+
+impl SpanProcessor for RouteConcurrencyCounterSpanProcessor {
+    fn force_flush(&self) -> OTelSdkResult {
+        Ok(())
+    }
+
+    fn shutdown(&self) -> OTelSdkResult {
+        Ok(())
+    }
+
+    fn on_start(&self, span: &mut opentelemetry_sdk::trace::Span, _cx: &Context) {
+        if !matches!(span.span_kind(), SpanKind::Server) {
+            return;
+        }
+        let Some(route) = span
+            .attributes()
+            .iter()
+            .find(|kv| kv.key.as_str() == "http.route")
+        else {
+            return;
+        };
+        let mut counts = self.0.lock().unwrap();
+        let count = counts.entry(route.key.clone()).or_default();
+        *count += 1;
+        span.set_attribute(KeyValue::new(
+            "http.route.concurrent_requests",
+            count.to_string(),
+        ));
+    }
+
+    fn on_end(&self, span: &mut FinishedSpan) {
+        if !matches!(span.span_kind(), SpanKind::Server) {
+            return;
+        }
+        let Some(route) = span
+            .attributes()
+            .iter()
+            .find(|kv| kv.key.as_str() == "http.route")
+        else {
+            return;
+        };
+        let mut counts = self.0.lock().unwrap();
+        let Some(count) = counts.get_mut(&route.key) else {
+            return;
+        };
+        *count -= 1;
+        if *count == 0 {
+            counts.remove(&route.key);
+        }
+    }
+}
+
 /// A custom log processor that enriches LogRecords with baggage attributes.
 /// Baggage information is not added automatically without this processor.
 #[derive(Debug)]
@@ -145,7 +207,7 @@ impl SpanProcessor for EnrichWithBaggageSpanProcessor {
         }
     }
 
-    fn on_end(&self, _span: opentelemetry_sdk::trace::SpanData) {}
+    fn on_end(&self, _span: &mut opentelemetry_sdk::trace::FinishedSpan) {}
 }
 
 fn init_tracer() -> SdkTracerProvider {
@@ -161,6 +223,7 @@ fn init_tracer() -> SdkTracerProvider {
     // Setup tracerprovider with stdout exporter
     // that prints the spans to stdout.
     let provider = SdkTracerProvider::builder()
+        .with_span_processor(RouteConcurrencyCounterSpanProcessor::default())
         .with_span_processor(EnrichWithBaggageSpanProcessor)
         .with_simple_exporter(SpanExporter::default())
         .build();

opentelemetry-sdk/src/trace/mod.rs

@@ -39,7 +39,7 @@ pub use id_generator::{IdGenerator, RandomIdGenerator};
 pub use links::SpanLinks;
 pub use provider::{SdkTracerProvider, TracerProviderBuilder};
 pub use sampler::{Sampler, ShouldSample};
-pub use span::Span;
+pub use span::{Span, ReadableSpan, FinishedSpan};
 pub use span_limit::SpanLimits;
 pub use span_processor::{
     BatchConfig, BatchConfigBuilder, BatchSpanProcessor, BatchSpanProcessorBuilder,
@@ -136,7 +136,7 @@ mod tests {
             }
         }
 
-        fn on_end(&self, _span: SpanData) {
+        fn on_end(&self, _span: &mut FinishedSpan) {
             // TODO: Accessing Context::current() will panic today and hence commented out.
             // See https://github.com/open-telemetry/opentelemetry-rust/issues/2871
             // let _c = Context::current();

opentelemetry-sdk/src/trace/provider.rs

@@ -462,8 +462,8 @@ mod tests {
         SERVICE_NAME, TELEMETRY_SDK_LANGUAGE, TELEMETRY_SDK_NAME, TELEMETRY_SDK_VERSION,
     };
     use crate::trace::provider::TracerProviderInner;
-    use crate::trace::{Config, Span, SpanProcessor};
-    use crate::trace::{SdkTracerProvider, SpanData};
+    use crate::trace::SdkTracerProvider;
+    use crate::trace::{Config, FinishedSpan, Span, SpanProcessor};
     use crate::Resource;
     use opentelemetry::trace::{Tracer, TracerProvider};
     use opentelemetry::{Context, Key, KeyValue, Value};
@@ -516,7 +516,7 @@ mod tests {
                 .fetch_add(1, Ordering::SeqCst);
         }
 
-        fn on_end(&self, _span: SpanData) {
+        fn on_end(&self, _span: &mut FinishedSpan) {
             // ignore
         }
 
@@ -779,7 +779,7 @@ mod tests {
             // No operation needed for this processor
         }
 
-        fn on_end(&self, _span: SpanData) {
+        fn on_end(&self, _span: &mut FinishedSpan) {
             // No operation needed for this processor
         }