feat: add an experimental on_ending api

This API allows to mutations of the span when it is ending.
It's marked as on development in the spec, but it is useful for
span obfuscation for example, which needs to done after
attributes can added to the span anymore.

Author: paullegranddc

examples/tracing-http-propagator/Cargo.toml

@@ -24,7 +24,7 @@ hyper = { workspace = true, features = ["full"] }
 hyper-util = { workspace = true, features = ["full"] }
 tokio = { workspace = true, features = ["full"] }
 opentelemetry = { path = "../../opentelemetry" }
-opentelemetry_sdk = { path = "../../opentelemetry-sdk" }
+opentelemetry_sdk = { path = "../../opentelemetry-sdk", features = ["experimental_span_processor_on_ending"]}
 opentelemetry-http = { path = "../../opentelemetry-http" }
 opentelemetry-stdout = { workspace = true, features = ["trace", "logs"] }
 opentelemetry-semantic-conventions = { path = "../../opentelemetry-semantic-conventions" }

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

@@ -166,6 +166,44 @@ impl SpanProcessor for RouteConcurrencyCounterSpanProcessor {
     }
 }
 
+fn obfuscate_http_auth_url(s: &str) -> Option<String> {
+    let uri = hyper::http::Uri::from_maybe_shared(s.to_owned()).ok()?;
+    let authority = uri.authority()?;
+    let (_, url) = authority.as_str().split_once('@')?;
+    let new_auth = format!("REDACTED_USERNAME:REDACTED_PASSWORD@{url}");
+    let mut parts = uri.into_parts();
+    parts.authority = Some(hyper::http::uri::Authority::from_maybe_shared(new_auth).ok()?);
+    Some(hyper::Uri::from_parts(parts).ok()?.to_string())
+}
+
+#[derive(Debug)]
+/// A custom span processor that uses on_ending to obfuscate sensitive information in span attributes.
+///
+/// Currently this only overrides http auth information in the URI.
+struct SpanOnbfuscationProcessor;
+
+impl SpanProcessor for SpanOnbfuscationProcessor {
+    fn force_flush(&self) -> OTelSdkResult {
+        Ok(())
+    }
+
+    fn shutdown(&self) -> OTelSdkResult {
+        Ok(())
+    }
+    fn on_start(&self, _span: &mut opentelemetry_sdk::trace::Span, _cx: &Context) {}
+
+    fn on_ending(&self, span: &mut opentelemetry_sdk::trace::Span) {
+        let mut obfuscated_attributes = Vec::new();
+        for KeyValue { key, value, .. } in span.attributes() {
+            if let Some(redacted_uri) = obfuscate_http_auth_url(value.as_str().as_ref()) {
+                obfuscated_attributes.push((key.clone(), KeyValue::new(key.clone(), redacted_uri)));
+            }
+        }
+    }
+
+    fn on_end(&self, _span: &mut FinishedSpan) {}
+}
+
 /// A custom log processor that enriches LogRecords with baggage attributes.
 /// Baggage information is not added automatically without this processor.
 #[derive(Debug)]
@@ -225,6 +263,7 @@ fn init_tracer() -> SdkTracerProvider {
     let provider = SdkTracerProvider::builder()
         .with_span_processor(RouteConcurrencyCounterSpanProcessor::default())
         .with_span_processor(EnrichWithBaggageSpanProcessor)
+        .with_span_processor(SpanOnbfuscationProcessor)
         .with_simple_exporter(SpanExporter::default())
         .build();
 

opentelemetry-sdk/Cargo.toml

@@ -58,6 +58,7 @@ experimental_logs_batch_log_processor_with_async_runtime = ["logs", "experimenta
 experimental_logs_concurrent_log_processor = ["logs"]
 experimental_trace_batch_span_processor_with_async_runtime = ["trace", "experimental_async_runtime"]
 experimental_metrics_disable_name_validation = ["metrics"]
+experimental_span_processor_on_ending = ["trace"]
 
 [[bench]]
 name = "context"

opentelemetry-sdk/src/trace/span.rs

@@ -198,33 +198,47 @@ impl opentelemetry::trace::Span for Span {
 
 impl Span {
     fn ensure_ended_and_exported(&mut self, timestamp: Option<SystemTime>) {
+        match self.data {
+            Some(ref mut data) => {
+                // ensure end time is set via explicit end or implicitly on drop
+                if let Some(timestamp) = timestamp {
+                    data.end_time = timestamp;
+                } else if data.end_time == data.start_time {
+                    data.end_time = opentelemetry::time::now();
+                }
+            }
+            None => {
+                return;
+            }
+        };
+
+        if self.tracer.provider().is_shutdown() {
+            return;
+        }
+        let provider = self.tracer.provider().clone();
+
+        #[cfg(feature = "experimental_span_processor_on_ending")]
+        {
+            for processor in provider.span_processors() {
+                processor.on_ending( self);
+            }
+        }
+
         let Span {
             data,
             tracer,
             span_context,
             ..
         } = self;
+    
         // skip if data has already been exported
-        let mut data = match data.take() {
+        let data = match data.take() {
             Some(data) => data,
             None => return,
         };
         let span_context: SpanContext =
             std::mem::replace(span_context, SpanContext::empty_context());
 
-        let provider = tracer.provider();
-        // skip if provider has been shut down
-        if provider.is_shutdown() {
-            return;
-        }
-
-        // ensure end time is set via explicit end or implicitly on drop
-        if let Some(timestamp) = timestamp {
-            data.end_time = timestamp;
-        } else if data.end_time == data.start_time {
-            data.end_time = opentelemetry::time::now();
-        }
-
         let mut finished_span = FinishedSpan {
             span: Some(build_export_data(data, span_context, tracer)),
             is_last_processor: false,

opentelemetry-sdk/src/trace/span_processor.rs

@@ -79,10 +79,23 @@ pub trait SpanProcessor: Send + Sync + std::fmt::Debug {
     /// synchronously on the thread that started the span, therefore it should
     /// not block or throw exceptions.
     fn on_start(&self, span: &mut Span, cx: &Context);
+    #[cfg(feature = "experimental_span_processor_on_ending")]
+    /// `on_ending` is called when a `Span` is ending. The en timestampe has already
+    /// been computed.
+    /// Mutations done to the span in this method will be reflected in the span passed
+    /// to other span processors.
+    /// This method is called synchronously within the `Span::end` API, therefore it
+    /// should not block or throw an exception.
+    /// If multiple SpanProcessors are registered, their on_ending methods are invoked
+    /// in the order they have been registered.
+    fn on_ending(&self, _span: &mut Span) {
+        // Default implementation is a no-op so existing processor implementations
+        // don't break if this feature in enabled transitively.
+    }
+
     /// `on_end` is called after a `Span` is ended (i.e., the end timestamp is
     /// already set). This method is called synchronously within the `Span::end`
     /// API, therefore it should not block or throw an exception.
-    /// TODO - This method should take reference to `SpanData`
     fn on_end(&self, span: &mut FinishedSpan);
     /// Force the spans lying in the cache to be exported.
     fn force_flush(&self) -> OTelSdkResult;
@@ -870,8 +883,8 @@ mod tests {
         OTEL_BSP_EXPORT_TIMEOUT_DEFAULT, OTEL_BSP_MAX_CONCURRENT_EXPORTS,
         OTEL_BSP_MAX_CONCURRENT_EXPORTS_DEFAULT, OTEL_BSP_MAX_EXPORT_BATCH_SIZE_DEFAULT,
     };
-    use crate::trace::{FinishedSpan, InMemorySpanExporterBuilder};
     use crate::trace::{BatchConfig, BatchConfigBuilder, SpanEvents, SpanLinks};
+    use crate::trace::{FinishedSpan, InMemorySpanExporterBuilder};
     use crate::trace::{SpanData, SpanExporter};
     use opentelemetry::trace::{SpanContext, SpanId, SpanKind, Status};
     use std::fmt::Debug;