feat: span processor api refactor

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

@@ -15,12 +15,17 @@ 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::time::Duration;
-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};
@@ -84,6 +89,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");
@@ -105,6 +111,66 @@ async fn router(
     response
 }
 
+#[derive(Debug, Default)]
+/// A custom span processor that counts concurrent requests for each route (indentified 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_with_timeout(&self, _timeout: Duration) -> crate::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 Ok(mut counts) = self.0.lock() else {
+            return;
+        };
+        let count = counts.entry(route.key.clone()).or_default();
+        *count += 1;
+        span.set_attribute(KeyValue::new(
+            "http.route.concurrent_requests",
+            *count as i64,
+        ));
+    }
+
+    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 Ok(mut counts) = self.0.lock() else {
+            return;
+        };
+        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)]
@@ -142,7 +208,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 {
@@ -158,6 +224,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/Cargo.toml

@@ -117,6 +117,12 @@ name = "log"
 harness = false
 required-features = ["logs"]
 
+[[bench]]
+name = "span_processor_api"
+harness = false
+required-features = ["testing"]
+
+
 [lib]
 bench = false
 

opentelemetry-sdk/benches/batch_span_processor.rs

@@ -4,10 +4,10 @@ use opentelemetry::trace::{
     SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId, TraceState,
 };
 use opentelemetry_sdk::testing::trace::NoopSpanExporter;
-use opentelemetry_sdk::trace::SpanData;
 use opentelemetry_sdk::trace::{
     BatchConfigBuilder, BatchSpanProcessor, SpanEvents, SpanLinks, SpanProcessor,
 };
+use opentelemetry_sdk::trace::{FinishedSpan, SpanData};
 use std::sync::Arc;
 use tokio::runtime::Runtime;
 
@@ -62,7 +62,8 @@ fn criterion_benchmark(c: &mut Criterion) {
                             let spans = get_span_data();
                             handles.push(tokio::spawn(async move {
                                 for span in spans {
-                                    span_processor.on_end(span);
+                                    let mut span = FinishedSpan::new(span);
+                                    span_processor.on_end(&mut span);
                                     tokio::task::yield_now().await;
                                 }
                             }));

opentelemetry-sdk/benches/span_processor_api.rs

@@ -0,0 +1,88 @@
+use std::time::Duration;
+
+use criterion::{black_box, criterion_group, criterion_main, Criterion};
+use opentelemetry::{
+    trace::{Span, Tracer, TracerProvider},
+    Context, KeyValue,
+};
+use opentelemetry_sdk::trace as sdktrace;
+
+#[cfg(not(target_os = "windows"))]
+use pprof::criterion::{Output, PProfProfiler};
+
+/*
+Adding results in comments for a quick reference.
+  Chip:	Apple M1 Max
+  Total Number of Cores:	10 (8 performance and 2 efficiency)
+
+SpanProcessorApi/0_processors
+    time:   [385.15 ns 386.14 ns 387.25 ns]
+SpanProcessorApi/1_processors
+    time:   [385.73 ns 387.17 ns 388.85 ns]
+SpanProcessorApi/2_processors
+    time:   [384.84 ns 385.66 ns 386.50 ns]
+SpanProcessorApi/4_processors
+    time:   [386.78 ns 388.17 ns 389.58 ns]
+*/
+
+#[derive(Debug)]
+struct NoopSpanProcessor;
+
+impl sdktrace::SpanProcessor for NoopSpanProcessor {
+    fn on_start(&self, _span: &mut sdktrace::Span, _parent_cx: &Context) {}
+    fn on_end(&self, _span: &mut sdktrace::FinishedSpan) {}
+    fn force_flush(&self) -> opentelemetry_sdk::error::OTelSdkResult {
+        Ok(())
+    }
+    fn shutdown_with_timeout(&self, _timeout: Duration) -> opentelemetry_sdk::error::OTelSdkResult {
+        Ok(())
+    }
+}
+
+fn create_tracer(span_processors_count: usize) -> sdktrace::SdkTracer {
+    let mut builder = sdktrace::SdkTracerProvider::builder();
+    for _ in 0..span_processors_count {
+        builder = builder.with_span_processor(NoopSpanProcessor);
+    }
+    builder.build().tracer("tracer")
+}
+
+fn create_span(tracer: &sdktrace::Tracer) -> sdktrace::Span {
+    let mut span = tracer.start("foo");
+    span.set_attribute(KeyValue::new("key1", false));
+    span.set_attribute(KeyValue::new("key2", "hello"));
+    span.set_attribute(KeyValue::new("key4", 123.456));
+    span.add_event("my_event", vec![KeyValue::new("key1", "value1")]);
+    span
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+    let mut group = c.benchmark_group("SpanProcessorApi");
+    for i in [0, 1, 2, 4] {
+        group.bench_function(format!("{}_processors", i), |b| {
+            let tracer = create_tracer(i);
+            b.iter(|| {
+                black_box(create_span(&tracer));
+            });
+        });
+    }
+}
+
+#[cfg(not(target_os = "windows"))]
+criterion_group! {
+    name = benches;
+    config = Criterion::default().with_profiler(PProfProfiler::new(100, Output::Flamegraph(None)))
+                               .warm_up_time(std::time::Duration::from_secs(1))
+                               .measurement_time(std::time::Duration::from_secs(2));
+    targets = criterion_benchmark
+}
+
+#[cfg(target_os = "windows")]
+criterion_group! {
+    name = benches;
+    config = Criterion::default().warm_up_time(std::time::Duration::from_secs(1))
+                               .measurement_time(std::time::Duration::from_secs(2));
+    targets = criterion_benchmark
+}
+
+criterion_main!(benches);

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::{FinishedSpan, ReadableSpan, Span};
 pub use span_limit::SpanLimits;
 pub use span_processor::{
     BatchConfig, BatchConfigBuilder, BatchSpanProcessor, BatchSpanProcessorBuilder,
@@ -137,7 +137,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

@@ -471,8 +471,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};
@@ -526,7 +526,7 @@ mod tests {
                 .fetch_add(1, Ordering::SeqCst);
         }
 
-        fn on_end(&self, _span: SpanData) {
+        fn on_end(&self, _span: &mut FinishedSpan) {
             // ignore
         }
 
@@ -789,7 +789,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
         }