Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 70 additions & 3 deletions examples/tracing-http-propagator/src/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -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");
Expand All @@ -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)]
Expand Down Expand Up @@ -146,7 +212,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 {
Expand All @@ -162,6 +228,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();
Expand Down
14 changes: 14 additions & 0 deletions opentelemetry-sdk/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,20 @@

## vNext

- **Breaking** `SpanProcessor::on_end` now takes `&mut FinishedSpan` instead of `SpanData`.
Processors that only need to read span data can use the `ReadableSpan` trait without cloning.
Processors that need ownership call `FinishedSpan::consume()` — the last processor receives
ownership via move (zero-copy), earlier processors receive a clone. Both `Span` (live) and
`FinishedSpan` (ended) implement `ReadableSpan`, enabling processors to inspect span data
in both `on_start` and `on_end` hooks. New `FinishedSpan::is_consumed()` method allows
checking whether span data has been consumed.
Supersedes [#2962](https://github.com/open-telemetry/opentelemetry-rust/pull/2962).
Relates to [#2940](https://github.com/open-telemetry/opentelemetry-rust/issues/2940),
[#2726](https://github.com/open-telemetry/opentelemetry-rust/issues/2726),
[#2939](https://github.com/open-telemetry/opentelemetry-rust/issues/2939).
- Fix `Span::end_with_timestamp` preserving explicit end times even when equal to start time,
instead of silently overwriting with the current time.

## 0.32.1

Released 2026-May-23
Expand Down
5 changes: 5 additions & 0 deletions opentelemetry-sdk/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,11 @@ name = "bound_instruments"
harness = false
required-features = ["metrics", "experimental_metrics_custom_reader", "experimental_metrics_bound_instruments", "spec_unstable_metrics_views"]

[[bench]]
name = "span_processor_api"
harness = false
required-features = ["testing"]

[lib]
bench = false

Expand Down
5 changes: 3 additions & 2 deletions opentelemetry-sdk/benches/batch_span_processor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -63,7 +63,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;
}
}));
Expand Down
88 changes: 88 additions & 0 deletions opentelemetry-sdk/benches/span_processor_api.rs
Original file line number Diff line number Diff line change
@@ -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 M4 Max
Total Number of Cores: 16 (12 performance and 4 efficiency)

SpanProcessorApi/0_processors
time: [190.51 ns 190.82 ns 191.13 ns]
SpanProcessorApi/1_processors
time: [191.60 ns 192.53 ns 193.42 ns]
SpanProcessorApi/2_processors
time: [191.24 ns 191.67 ns 192.15 ns]
SpanProcessorApi/4_processors
time: [192.94 ns 193.48 ns 194.12 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);
6 changes: 3 additions & 3 deletions opentelemetry-sdk/src/trace/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@ pub use id_generator::{IdGenerator, RandomIdGenerator};
pub use links::SpanLinks;
pub use provider::{SdkTracerProvider, TracerProviderBuilder};
pub use sampler::{Sampler, SamplingDecision, SamplingResult, ShouldSample};
pub use span::Span;
pub use span::{FinishedSpan, ReadableSpan, Span};
pub use span_limit::SpanLimits;
pub use span_processor::{
BatchConfig, BatchConfigBuilder, BatchSpanProcessor, BatchSpanProcessorBuilder,
Expand Down Expand Up @@ -150,7 +150,7 @@ mod tests {
}
}

fn on_end(&self, span: SpanData) {
fn on_end(&self, span: &mut FinishedSpan) {
// Fixed: Context::current() no longer panics from Drop
// See https://github.com/open-telemetry/opentelemetry-rust/issues/2871
Context::current();
Expand All @@ -161,7 +161,7 @@ mod tests {

// Verify: on_start stored the baggage as an attribute
assert!(
span.attributes
span.attributes()
.iter()
.any(|kv| kv.key.as_str() == "bag-key"),
"Baggage should have been stored as span attribute in on_start"
Expand Down
8 changes: 4 additions & 4 deletions opentelemetry-sdk/src/trace/provider.rs
Original file line number Diff line number Diff line change
Expand Up @@ -510,8 +510,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};
Expand Down Expand Up @@ -565,7 +565,7 @@ mod tests {
.fetch_add(1, Ordering::SeqCst);
}

fn on_end(&self, _span: SpanData) {
fn on_end(&self, _span: &mut FinishedSpan) {
// ignore
}

Expand Down Expand Up @@ -858,7 +858,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
}

Expand Down
Loading
Loading