Fix span processor exporting unsampled spans (#871)

Currently span processors will send span data to exporters even if the
sampled flag is false. This patch fixes this by explicitly checking the
sampled state in both the batch and simple span processors.

opentelemetry-sdk/src/testing/trace.rs

@@ -9,14 +9,22 @@ use crate::{
 use async_trait::async_trait;
 use futures_util::future::BoxFuture;
 pub use opentelemetry_api::testing::trace::TestSpan;
-use opentelemetry_api::trace::{SpanContext, SpanId, SpanKind, Status};
+use opentelemetry_api::trace::{
+    SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId, TraceState,
+};
 use std::fmt::{Display, Formatter};
 use std::sync::mpsc::{channel, Receiver, Sender};
 
 pub fn new_test_export_span_data() -> SpanData {
     let config = Config::default();
     SpanData {
-        span_context: SpanContext::empty_context(),
+        span_context: SpanContext::new(
+            TraceId::from_u128(1),
+            SpanId::from_u64(1),
+            TraceFlags::SAMPLED,
+            false,
+            TraceState::default(),
+        ),
         parent_span_id: SpanId::INVALID,
         span_kind: SpanKind::Internal,
         name: "opentelemetry".into(),

opentelemetry-sdk/src/trace/span_processor.rs

@@ -141,6 +141,10 @@ impl SpanProcessor for SimpleSpanProcessor {
     }
 
     fn on_end(&self, span: SpanData) {
+        if !span.span_context.is_sampled() {
+            return;
+        }
+
         if let Err(err) = self.sender.send(Some(span)) {
             global::handle_error(TraceError::from(format!("error processing span {:?}", err)));
         }
@@ -242,6 +246,10 @@ impl<R: TraceRuntime> SpanProcessor for BatchSpanProcessor<R> {
     }
 
     fn on_end(&self, span: SpanData) {
+        if !span.span_context.is_sampled() {
+            return;
+        }
+
         let result = self.message_sender.try_send(BatchMessage::ExportSpan(span));
 
         if let Err(err) = result {
@@ -687,8 +695,9 @@ mod tests {
     use crate::testing::trace::{
         new_test_export_span_data, new_test_exporter, new_tokio_test_exporter,
     };
-    use crate::trace::BatchConfig;
+    use crate::trace::{BatchConfig, EvictedHashMap, EvictedQueue};
     use async_trait::async_trait;
+    use opentelemetry_api::trace::{SpanContext, SpanId, SpanKind, Status};
     use std::fmt::Debug;
     use std::future::Future;
     use std::time::Duration;
@@ -702,6 +711,28 @@ mod tests {
         let _result = processor.shutdown();
     }
 
+    #[test]
+    fn simple_span_processor_on_end_skips_export_if_not_sampled() {
+        let (exporter, rx_export, _rx_shutdown) = new_test_exporter();
+        let processor = SimpleSpanProcessor::new(Box::new(exporter));
+        let unsampled = SpanData {
+            span_context: SpanContext::empty_context(),
+            parent_span_id: SpanId::INVALID,
+            span_kind: SpanKind::Internal,
+            name: "opentelemetry".into(),
+            start_time: opentelemetry_api::time::now(),
+            end_time: opentelemetry_api::time::now(),
+            attributes: EvictedHashMap::new(0, 0),
+            events: EvictedQueue::new(0),
+            links: EvictedQueue::new(0),
+            status: Status::Unset,
+            resource: Default::default(),
+            instrumentation_lib: Default::default(),
+        };
+        processor.on_end(unsampled);
+        assert!(rx_export.recv_timeout(Duration::from_millis(100)).is_err());
+    }
+
     #[test]
     fn simple_span_processor_shutdown_calls_shutdown() {
         let (exporter, _rx_export, rx_shutdown) = new_test_exporter();

scripts/patch_dependencies.sh

@@ -1,6 +1,9 @@
-# Dashmap 5.3.4 requires 1.59
+# Dashmap >= 5.3.4 requires 1.59
 latest_version=$(cargo search --limit 1 dashmap | head -1 | cut -d'"' -f2)  &&
 cargo update -p dashmap:$latest_version --precise 5.1.0 &&
 # We have time 0.1 and 0.3
 latest_version=$(cargo search --limit 1 time | head -1 | cut -d'"' -f2) &&
-cargo update -p time:$latest_version --precise 0.3.9
+cargo update -p time:$latest_version --precise 0.3.9 &&
+# async-global-executor >= 2.3.0 requires 1.59
+latest_version=$(cargo search --limit 1 async-global-executor | head -1 | cut -d'"' -f2)  &&
+cargo update -p async-global-executor:$latest_version --precise 2.2.0