fix(transactions): Defer dynamic sampling until metrics config is valid

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+**Bug Fixes**:
+
+- Defer dynamic sampling until metrics config is valid. ([#6246](https://github.com/getsentry/relay/pull/6246))
+
 ## 26.7.1
 
 **Features**:
@@ -41,6 +47,7 @@
 - Set sentry.trace.status on segment spans. ([#6140](https://github.com/getsentry/relay/pull/6140))
 - Don't modify segment information for V2 web vital spans. ([#6160](https://github.com/getsentry/relay/pull/6160))
 
+
 - Support compressed minidumps when the `relay-minidump-uploads` feature is enabled. ([#6151](https://github.com/getsentry/relay/pull/6151))
 - Make `--log-level` and `--log-format` take effect again and accept them on all subcommands. ([#6198](https://github.com/getsentry/relay/pull/6198))
 - Parse two-component versions in iOS and iPadOS `raw_description` into `version` instead of `kernel_version`. ([#6197](https://github.com/getsentry/relay/pull/6197))

relay-dynamic-config/src/metrics.rs

@@ -80,6 +80,12 @@ pub struct CombinedMetricExtractionConfig<'a> {
 }
 
 impl<'a> CombinedMetricExtractionConfig<'a> {
+    /// Empty config, used in tests and as a fallback.
+    pub const EMPTY: Self = Self {
+        global: MetricExtractionGroups::EMPTY,
+        project: &MetricExtractionConfig::empty(),
+    };
+
     /// Creates a new combined view from two references.
     pub fn new(global: &'a MetricExtractionGroups, project: &'a MetricExtractionConfig) -> Self {
         for key in project.global_groups.keys() {
@@ -230,12 +236,12 @@ impl MetricExtractionConfig {
     /// Returns an empty `MetricExtractionConfig` with the latest version.
     ///
     /// As opposed to `default()`, this will be enabled once populated with specs.
-    pub fn empty() -> Self {
+    pub const fn empty() -> Self {
         Self {
             version: Self::MAX_SUPPORTED_VERSION,
             global_groups: BTreeMap::new(),
-            metrics: Default::default(),
-            tags: Default::default(),
+            metrics: Vec::new(),
+            tags: Vec::new(),
             _conditional_tags_extended: false,
             _span_metrics_extended: false,
         }

relay-server/src/processing/transactions/extraction.rs

@@ -1,5 +1,5 @@
 use relay_base_schema::project::ProjectId;
-use relay_dynamic_config::{CombinedMetricExtractionConfig, ErrorBoundary, MetricExtractionGroups};
+use relay_dynamic_config::CombinedMetricExtractionConfig;
 use relay_event_normalization::span::tag_extraction;
 use relay_event_schema::protocol::{Event, Span};
 use relay_metrics::MetricNamespace;
@@ -28,6 +28,7 @@ pub fn extract_segment_span(
 
 /// Input arguments for [`extract_metrics`].
 pub struct ExtractMetricsContext<'a> {
+    pub config: CombinedMetricExtractionConfig<'a>,
     pub dsc: Option<&'a DynamicSamplingContext>,
     pub project_id: ProjectId,
     pub ctx: Context<'a>,
@@ -40,54 +41,24 @@ pub fn extract_metrics(
     event: &mut Annotated<Event>,
     extracted_metrics: &mut ProcessingExtractedMetrics,
     ctx: ExtractMetricsContext,
-) -> bool {
+) {
     let ExtractMetricsContext {
+        config,
         dsc,
         project_id,
         ctx,
         sampling_decision,
         extract_span_metrics,
     } = ctx;
-    // TODO(follow-up): this function should always extract metrics. Dynamic sampling should validate
-    // the full metrics extraction config and skip sampling if it is incomplete.
-
     let Some(event) = event.value_mut() else {
-        // Nothing to extract, but metrics extraction was called.
-        return true;
-    };
-
-    // NOTE: This function requires a `metric_extraction` in the project config. Legacy configs
-    // will upsert this configuration from transaction and conditional tagging fields, even if
-    // it is not present in the actual project config payload.
-    let combined_config = {
-        let config = match &ctx.project_info.config.metric_extraction {
-            ErrorBoundary::Ok(config) if config.is_supported() => config,
-            _ => return false,
-        };
-        let global_config = match &ctx.global_config.metric_extraction {
-            ErrorBoundary::Ok(global_config) => global_config,
-            #[allow(unused_variables)]
-            ErrorBoundary::Err(e) => {
-                if cfg!(feature = "processing") && ctx.config.processing_enabled() {
-                    // Config is invalid, but we will try to extract what we can with just the
-                    // project config.
-                    relay_log::error!("Failed to parse global extraction config {e}");
-                    MetricExtractionGroups::EMPTY
-                } else {
-                    // If there's an error with global metrics extraction, it is safe to assume that this
-                    // Relay instance is not up-to-date, and we should skip extraction.
-                    relay_log::debug!("Failed to parse global extraction config: {e}");
-                    return false;
-                }
-            }
-        };
-        CombinedMetricExtractionConfig::new(global_config, config)
+        // Nothing to extract.
+        return;
     };
 
     let metrics = crate::metrics_extraction::event::extract_metrics(
         event,
         crate::metrics_extraction::event::ExtractMetricsConfig {
-            config: combined_config,
+            config,
             sampling_decision,
             target_project_id: project_id,
             max_tag_value_size: ctx
@@ -99,6 +70,4 @@ pub fn extract_metrics(
         },
     );
     extracted_metrics.extend(metrics, Some(sampling_decision));
-
-    true
 }

relay-server/src/processing/transactions/process.rs

@@ -1,4 +1,5 @@
 use relay_base_schema::events::EventType;
+use relay_dynamic_config::{CombinedMetricExtractionConfig, ErrorBoundary, MetricExtractionGroups};
 use relay_event_normalization::GeoIpLookup;
 use relay_event_schema::protocol::Event;
 use relay_profiling::{ProfileError, ProfileType};
@@ -203,19 +204,33 @@ pub enum SamplingOutput {
         payload: Managed<Box<ExpandedTransaction>>,
         sample_rate: Option<f64>,
     },
-    /// The decision was discard keep only extracted metrics and an optional profile.
+    /// The decision was discard, keep only extracted metrics and an optional profile.
     Drop {
         metrics: Managed<ExtractedMetrics>,
         profile: Option<Managed<Box<StandaloneProfile>>>,
     },
 }
 
 /// Computes the sampling decision for a transaction and associated items.
+///
+/// Returns the sampling output as well as the validated metrics config, if possible / needed.
 pub fn run_dynamic_sampling(
     payload: Managed<Box<ExpandedTransaction>>,
     ctx: Context<'_>,
     filters_status: FiltersStatus,
 ) -> SamplingOutput {
+    let conf = match get_metrics_config(ctx) {
+        Ok(conf) => conf,
+        Err(_) if ctx.is_processing() => CombinedMetricExtractionConfig::EMPTY,
+        Err(_) => {
+            // Defer dynamic sampling until the next relay.
+            return SamplingOutput::Keep {
+                payload,
+                sample_rate: None,
+            };
+        }
+    };
+
     let sampling_result = make_dynamic_sampling_decision(&payload, ctx, filters_status);
 
     let sampling_match = match sampling_result {
@@ -229,7 +244,7 @@ pub fn run_dynamic_sampling(
     };
 
     // At this point the decision is to drop the payload.
-    let (payload, metrics) = split_indexed_and_total(payload, ctx, SamplingDecision::Drop);
+    let (payload, metrics) = split_indexed_and_total(payload, ctx, SamplingDecision::Drop, conf);
 
     let (payload, profile) = payload.split_once(|mut payload, _| {
         let profile = payload.profile.take().map(|profile| StandaloneProfile {
@@ -251,6 +266,32 @@ pub fn run_dynamic_sampling(
     }
 }
 
+/// Compiles a valid metrics config from a [`Context`].
+pub fn get_metrics_config<'a>(ctx: Context<'a>) -> Result<CombinedMetricExtractionConfig<'a>, ()> {
+    let config = match &ctx.project_info.config.metric_extraction {
+        ErrorBoundary::Ok(config) if config.is_supported() => config,
+        _ => return Err(()),
+    };
+    let global_config = match &ctx.global_config.metric_extraction {
+        ErrorBoundary::Ok(global_config) => global_config,
+        ErrorBoundary::Err(e) => {
+            if ctx.is_processing() {
+                // Config is invalid, but we will try to extract what we can with just the
+                // project config.
+                relay_log::error!("Failed to parse global extraction config {e}");
+                MetricExtractionGroups::EMPTY
+            } else {
+                // If there's an error with global metrics extraction, it is safe to assume that this
+                // Relay instance is not up-to-date, and we should skip extraction.
+                relay_log::debug!("Failed to parse global extraction config: {e}");
+                return Err(());
+            }
+        }
+    };
+
+    Ok(CombinedMetricExtractionConfig::new(global_config, config))
+}
+
 /// Computes the dynamic sampling decision for the unit of work, but does not perform action on data.
 fn make_dynamic_sampling_decision(
     work: &Managed<Box<ExpandedTransaction>>,
@@ -298,30 +339,28 @@ pub fn split_indexed_and_total_with_extracted_spans(
 ) -> IndexedTransactionAndSpanAndMetrics {
     let scoping = transaction.scoping();
 
-    let mut metrics_extracted = false;
+    debug_assert!(ctx.is_processing());
+
     let (transaction, metrics) = transaction.split_once(|mut tx, r| {
         r.lenient(DataCategory::MetricBucket);
 
         let mut metrics = ProcessingExtractedMetrics::new();
-
-        metrics_extracted = extraction::extract_metrics(
+        extraction::extract_metrics(
             &mut tx.event,
             &mut metrics,
             ExtractMetricsContext {
                 dsc: tx.headers.dsc(),
                 project_id: scoping.project_id,
+                // We can fall back to the default, we're in a processing Relay in which the config
+                // must always be valid, worst case we fall back to a default empty config, because
+                // we can't really do anything else.
+                config: get_metrics_config(ctx).unwrap_or(CombinedMetricExtractionConfig::EMPTY),
                 ctx,
                 sampling_decision: SamplingDecision::Keep,
                 extract_span_metrics: spans.is_some(),
             },
         );
 
-        if !metrics_extracted {
-            // Invalid config or invalid original transaction
-            r.lenient(DataCategory::Transaction);
-            r.lenient(DataCategory::Span);
-        }
-
         // This really is a bug, we ignore here.
         //
         // Transactions are counted using a span metric, as transaction payloads should
@@ -365,7 +404,7 @@ pub fn split_indexed_and_total_with_extracted_spans(
                 // "Insurance" that metrics extracted from the transaction spans match the extracted
                 // spans.
                 r.modify_by(*c, -(*q as isize));
-            } else if metrics_extracted {
+            } else {
                 // Metrics were extracted but do not contain a span quantity,
                 // be lenient about the span counts instead of failing bookkeeping.
                 r.lenient(DataCategory::Span);
@@ -385,35 +424,33 @@ type IndexedAndMetrics = (
 
 /// Splits transaction into indexed payload and metrics representing the total counts.
 pub fn split_indexed_and_total(
-    mut work: Managed<Box<ExpandedTransaction>>,
+    work: Managed<Box<ExpandedTransaction>>,
     ctx: Context<'_>,
     sampling_decision: SamplingDecision,
+    config: CombinedMetricExtractionConfig<'_>,
 ) -> IndexedAndMetrics {
     let scoping = work.scoping();
 
-    let mut metrics = ProcessingExtractedMetrics::new();
-    let mut metrics_extracted = false;
-    work.modify(|work, _| {
-        metrics_extracted = extraction::extract_metrics(
+    work.split_once(|mut work, r| {
+        r.lenient(DataCategory::MetricBucket);
+
+        let mut metrics = ProcessingExtractedMetrics::new();
+
+        extraction::extract_metrics(
             &mut work.event,
             &mut metrics,
             ExtractMetricsContext {
+                config,
                 dsc: work.headers.dsc(),
                 project_id: scoping.project_id,
                 ctx,
                 sampling_decision,
                 extract_span_metrics: true,
             },
         );
-    });
 
-    work.split_once(|work, r| {
-        r.lenient(DataCategory::MetricBucket);
-        if !metrics_extracted {
-            // Invalid config or invalid original transaction
-            r.lenient(DataCategory::Transaction);
-            r.lenient(DataCategory::Span);
-        }
+        r.lenient(DataCategory::Transaction);
+        r.lenient(DataCategory::Span);
 
         (Box::new(work.into_indexed()), metrics.into_inner())
     })

relay-server/src/processing/transactions/types/expanded.rs

@@ -1,4 +1,5 @@
 use either::Either;
+use relay_dynamic_config::CombinedMetricExtractionConfig;
 use relay_event_schema::protocol::{Event, SpanV2};
 use relay_profiling::{ProfileMetadata, ProfileType};
 use relay_protocol::Annotated;
@@ -12,7 +13,7 @@ use crate::envelope::{ContentType, EnvelopeHeaders, Item, ItemType, Items};
 use crate::managed::{Counted, Managed, Quantities, Rejected};
 use crate::metrics_extraction::ExtractedMetrics;
 use crate::processing::transactions::Error;
-use crate::processing::transactions::process::split_indexed_and_total;
+use crate::processing::transactions::process::{get_metrics_config, split_indexed_and_total};
 use crate::processing::utils::types::{Indexed, TotalAndIndexed, TotalCategory};
 use crate::processing::{Context, CountRateLimited, RateLimited, RateLimiter};
 use crate::statsd::RelayTimers;
@@ -191,7 +192,11 @@ impl RateLimited for Managed<Box<ExpandedTransaction<TotalAndIndexed>>> {
             .await;
         if !limits.is_empty() {
             let error = Error::from(limits);
-            let (indexed, metrics) = split_indexed_and_total(self, ctx, SamplingDecision::Keep);
+
+            // Everything will be dropped, might as well fallback to an empty config.
+            let config = get_metrics_config(ctx).unwrap_or(CombinedMetricExtractionConfig::EMPTY);
+            let (indexed, metrics) =
+                split_indexed_and_total(self, ctx, SamplingDecision::Keep, config);
             let _ = indexed.reject_err(error);
 
             return Ok(metrics.map(|metrics, _| Either::Right(metrics)));

tests/integration/test_dynamic_sampling.py

@@ -936,6 +936,28 @@ def test_invalid_global_generic_filters_skip_dynamic_sampling(mini_sentry, relay
     assert mini_sentry.get_captured_envelope()
 
 
+def test_invalid_metric_extraction_config_skips_dynamic_sampling(mini_sentry, relay):
+    relay = relay(mini_sentry, _outcomes_enabled_config())
+
+    project_id = 42
+    config = mini_sentry.add_basic_project_config(project_id)
+    public_key = config["publicKeys"][0]["publicKey"]
+
+    # Unsupported metric extraction config, so no metrics can be extracted
+    config["config"]["metricExtraction"] = {
+        "version": 666,  # this version is too new for this relay
+        "metrics": [],
+    }
+
+    # Reject all transactions with dynamic sampling
+    add_sampling_config(config, sample_rate=0, rule_type="transaction")
+
+    envelope, _, _ = _create_transaction_envelope(public_key)
+
+    relay.send_envelope(project_id, envelope)
+    assert mini_sentry.get_captured_envelope()
+
+
 def get_transaction_envelope(
     trace_id: str,
     segment_id: str,