fix(ds): Emit dynamic sampling outcomes for spans contained in transaction (#4569)

Dynamic sampling currently only emits outcomes for the transaction but
not the contained spans. With this PR sampled/filtered outcomes are also
emitted for contained spans.

Author: Dav1dde

CHANGELOG.md

@@ -16,6 +16,7 @@
 **Bug Fixes**:
 
 - Prevent partial trims in indexed and queryable span data. ([#4557](https://github.com/getsentry/relay/pull/4557))
+- Emit filtered/sampled outcomes for spans attached to a dropped transaction. ([#4569](https://github.com/getsentry/relay/pull/4569))
 
 **Internal**:
 

relay-server/src/services/processor.rs

@@ -1784,7 +1784,12 @@ impl EnvelopeProcessorService {
                 spans_extracted,
             )?;
 
-            dynamic_sampling::drop_unsampled_items(managed_envelope, event, outcome);
+            dynamic_sampling::drop_unsampled_items(
+                managed_envelope,
+                event,
+                outcome,
+                spans_extracted,
+            );
 
             // At this point we have:
             //  - An empty envelope.

relay-server/src/services/processor/dynamic_sampling.rs

@@ -7,13 +7,16 @@ use relay_config::Config;
 use relay_dynamic_config::ErrorBoundary;
 use relay_event_schema::protocol::{Contexts, Event, TraceContext};
 use relay_protocol::{Annotated, Empty};
+use relay_quotas::DataCategory;
 use relay_sampling::config::RuleType;
 use relay_sampling::evaluation::{ReservoirEvaluator, SamplingEvaluator};
 use relay_sampling::{DynamicSamplingContext, SamplingConfig};
 
 use crate::envelope::{CountFor, ItemType};
 use crate::services::outcome::Outcome;
-use crate::services::processor::{event_category, EventProcessing, Sampling, TransactionGroup};
+use crate::services::processor::{
+    event_category, EventProcessing, Sampling, SpansExtracted, TransactionGroup,
+};
 use crate::services::projects::project::ProjectInfo;
 use crate::utils::{self, SamplingResult, TypedEnvelope};
 
@@ -112,6 +115,7 @@ pub fn drop_unsampled_items(
     managed_envelope: &mut TypedEnvelope<TransactionGroup>,
     event: Annotated<Event>,
     outcome: Outcome,
+    spans_extracted: SpansExtracted,
 ) {
     // Remove all items from the envelope which need to be dropped due to dynamic sampling.
     let dropped_items = managed_envelope
@@ -136,6 +140,23 @@ pub fn drop_unsampled_items(
         item.set_sampled(false);
     }
 
+    // Another 'hack' to emit outcomes from the container item for the contained items (spans).
+    //
+    // The entire tracking outcomes for contained elements is not handled in a systematic way
+    // and whenever an event/transaction is discarded, contained elements are tracked in a 'best
+    // effort' basis (basically in all the cases where someone figured out this is a problem).
+    //
+    // This is yet another case, when the spans have not yet been separated from the transaction
+    // also emit dynamic sampling outcomes for the contained spans.
+    if !spans_extracted.0 {
+        let spans = event.value().and_then(|e| e.spans.value());
+        let span_count = spans.map_or(0, |s| s.len());
+
+        // Track the amount of contained spans + 1 segment span (the transaction itself which would
+        // be converted to a span).
+        managed_envelope.track_outcome(outcome.clone(), DataCategory::SpanIndexed, span_count + 1);
+    }
+
     // All items have been dropped, now make sure the event is also handled and dropped.
     if let Some(category) = event_category(&event) {
         let category = category.index_category().unwrap_or(category);

tests/integration/test_dynamic_sampling.py

@@ -492,7 +492,9 @@ def test_uses_trace_public_key(mini_sentry, relay):
         mini_sentry.captured_events.get(timeout=1)
 
     # and it should create an outcome
-    outcomes = mini_sentry.captured_outcomes.get(timeout=2)
+    outcomes = mini_sentry.captured_outcomes.get(timeout=2)  # Spans
+    assert outcomes is not None
+    outcomes = mini_sentry.captured_outcomes.get(timeout=2)  # Transactions
     assert outcomes is not None
     with pytest.raises(queue.Empty):
         mini_sentry.captured_outcomes.get(timeout=1)
@@ -586,6 +588,14 @@ def test_client_sample_rate_adjusted(mini_sentry, relay, rule_type, event_factor
     itself to 1.0. The chances of this test passing without the adjustment in
     place are very low (but not 0).
     """
+
+    def assert_client_reports():
+        # Relay is sending a client report, skip over it (1 for transactions, 1 for spans)
+        for _ in range(2):
+            received_envelope = mini_sentry.captured_events.get(timeout=1)
+            assert received_envelope.get_transaction_event() is None
+            assert received_envelope.get_event() is None
+
     project_id = 42
     relay = relay(mini_sentry)
     config = mini_sentry.add_basic_project_config(project_id)
@@ -604,18 +614,12 @@ def test_client_sample_rate_adjusted(mini_sentry, relay, rule_type, event_factor
     )
 
     relay.send_envelope(project_id, envelope)
-
-    received_envelope = mini_sentry.captured_events.get(timeout=1)
-    received_envelope.get_transaction_event()
+    assert_client_reports()
 
     envelope, trace_id, event_id = event_factory(public_key, client_sample_rate=1.0)
 
     relay.send_envelope(project_id, envelope)
-
-    # Relay is sending a client report, skip over it
-    received_envelope = mini_sentry.captured_events.get(timeout=1)
-    assert received_envelope.get_transaction_event() is None
-    assert received_envelope.get_event() is None
+    assert_client_reports()
 
     with pytest.raises(queue.Empty):
         mini_sentry.captured_events.get(timeout=1)

tests/integration/test_metrics.py

@@ -1019,6 +1019,8 @@ def test_transaction_metrics_extraction_external_relays(
     if expect_metrics_extraction:
         metrics_envelope = mini_sentry.captured_events.get(timeout=3)
         assert len(metrics_envelope.items) == 1
+        metrics_envelope = mini_sentry.captured_events.get(timeout=3)
+        assert len(metrics_envelope.items) == 1
 
         payload = json.loads(metrics_envelope.items[0].get_bytes().decode())
         assert len(payload) == 4
@@ -1576,9 +1578,13 @@ def test_generic_metric_extraction(mini_sentry, relay):
     relay = relay(relay(mini_sentry, options=TEST_CONFIG), options=TEST_CONFIG)
     relay.send_transaction(PROJECT_ID, transaction)
 
+    # Skip client reports
     envelope = mini_sentry.captured_events.get(timeout=3)
+    assert envelope.get_event() is None
     envelope = mini_sentry.captured_events.get(timeout=3)
+    assert envelope.get_event() is None
 
+    envelope = mini_sentry.captured_events.get(timeout=3)
     for item in envelope.items:
         # Transaction items should be sampled and not among the envelope items.
         assert item.headers.get("type") != "transaction"
@@ -2008,7 +2014,7 @@ def test_histogram_outliers(mini_sentry, relay):
     relay.send_event(42, event)
 
     tags = {}
-    for _ in range(2):
+    for _ in range(3):
         envelope = mini_sentry.captured_events.get()
         for item in envelope:
             if item.type == "metric_buckets":