fix(metrics): Introduce revisions for metrics extraction (#1251)

* fix(metrics): Introduce revisions for metrics extraction * meta: Changelog
getsentry · Apr 29, 2022 · 2ca65bc · 2ca65bc
1 parent 9595e7e
commit 2ca65bc
Show file tree

Hide file tree

Showing 5 changed files with 84 additions and 25 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -11,6 +11,7 @@
 
 - fix(metrics): Enforce metric name length limit. ([#1238](https://github.com/getsentry/relay/pull/1238))
 - Accept and forward unknown Envelope items. In processing mode, drop items individually rather than rejecting the entire request. This allows SDKs to send new data in combined Envelopes in the future. ([#1246](https://github.com/getsentry/relay/pull/1246))
+- Stop extracting metrics with outdated names from sessions. ([#1251](https://github.com/getsentry/relay/pull/1251))
 
 **Internal**:
 

diff --git a/relay-server/src/actors/envelopes.rs b/relay-server/src/actors/envelopes.rs
@@ -51,7 +51,7 @@ use crate::actors::upstream::{SendRequest, UpstreamRelay, UpstreamRequest, Upstr
 use crate::envelope::{self, AttachmentType, ContentType, Envelope, EnvelopeError, Item, ItemType};
 use crate::extractors::{PartialDsn, RequestMeta};
 use crate::http::{HttpError, Request, RequestBuilder, Response};
-use crate::metrics_extraction::sessions::extract_session_metrics;
+use crate::metrics_extraction::sessions::{extract_session_metrics, SessionMetricsConfig};
 use crate::service::ServerError;
 use crate::statsd::{RelayCounters, RelayHistograms, RelaySets, RelayTimers};
 use crate::utils::{
@@ -368,8 +368,8 @@ struct ProcessEnvelopeState {
 
     /// Metrics extracted from items in the envelope.
     ///
-    /// This is controlled by [`Feature::MetricsExtraction`]. Relay extracts metrics for sessions
-    /// and transactions.
+    /// Relay can extract metrics for sessions and transactions, which is controlled by
+    /// configuration objects in the project config.
     extracted_metrics: Vec<Metric>,
 
     /// The state of the project that this envelope belongs to.
@@ -603,7 +603,7 @@ impl EnvelopeProcessor {
         received: DateTime<Utc>,
         client: Option<&str>,
         client_addr: Option<net::IpAddr>,
-        metrics_extraction_enabled: bool,
+        metrics_config: SessionMetricsConfig,
         clock_drift_processor: &ClockDriftProcessor,
         extracted_metrics: &mut Vec<Metric>,
     ) -> bool {
@@ -660,12 +660,17 @@ impl EnvelopeProcessor {
             }
         }
 
-        // Extract metrics
-        if metrics_extraction_enabled && !item.metrics_extracted() {
+        // Extract metrics if they haven't been extracted by a prior Relay
+        if metrics_config.is_enabled() && !item.metrics_extracted() {
             extract_session_metrics(&session.attributes, &session, client, extracted_metrics);
             item.set_metrics_extracted(true);
         }
 
+        // Drop the session if metrics have been extracted in this or a prior Relay
+        if metrics_config.should_drop() && item.metrics_extracted() {
+            return false;
+        }
+
         if changed {
             let json_string = match serde_json::to_string(&session) {
                 Ok(json) => json,
@@ -688,7 +693,7 @@ impl EnvelopeProcessor {
         received: DateTime<Utc>,
         client: Option<&str>,
         client_addr: Option<net::IpAddr>,
-        metrics_extraction_enabled: bool,
+        metrics_config: SessionMetricsConfig,
         clock_drift_processor: &ClockDriftProcessor,
         extracted_metrics: &mut Vec<Metric>,
     ) -> bool {
@@ -729,14 +734,19 @@ impl EnvelopeProcessor {
             }
         }
 
-        // Extract metrics
-        if metrics_extraction_enabled && !item.metrics_extracted() {
+        // Extract metrics if they haven't been extracted by a prior Relay
+        if metrics_config.is_enabled() && !item.metrics_extracted() {
             for aggregate in &session.aggregates {
                 extract_session_metrics(&session.attributes, aggregate, client, extracted_metrics);
                 item.set_metrics_extracted(true);
             }
         }
 
+        // Drop the aggregate if metrics have been extracted in this or a prior Relay
+        if metrics_config.should_drop() && item.metrics_extracted() {
+            return false;
+        }
+
         if changed {
             let json_string = match serde_json::to_string(&session) {
                 Ok(json) => json,
@@ -759,8 +769,7 @@ impl EnvelopeProcessor {
     fn process_sessions(&self, state: &mut ProcessEnvelopeState) {
         let received = state.envelope_context.received_at;
         let extracted_metrics = &mut state.extracted_metrics;
-        let metrics_extraction_enabled =
-            state.project_state.has_feature(Feature::MetricsExtraction);
+        let metrics_config = state.project_state.config().session_metrics;
         let envelope = &mut state.envelope;
         let client = envelope.meta().client().map(|x| x.to_owned());
         let client_addr = envelope.meta().client_addr();
@@ -775,7 +784,7 @@ impl EnvelopeProcessor {
                     received,
                     client.as_deref(),
                     client_addr,
-                    metrics_extraction_enabled,
+                    metrics_config,
                     &clock_drift_processor,
                     extracted_metrics,
                 ),
@@ -784,7 +793,7 @@ impl EnvelopeProcessor {
                     received,
                     client.as_deref(),
                     client_addr,
-                    metrics_extraction_enabled,
+                    metrics_config,
                     &clock_drift_processor,
                     extracted_metrics,
                 ),

diff --git a/relay-server/src/actors/project.rs b/relay-server/src/actors/project.rs
@@ -29,6 +29,7 @@ use crate::actors::project_cache::{
 };
 use crate::envelope::Envelope;
 use crate::extractors::RequestMeta;
+use crate::metrics_extraction::sessions::SessionMetricsConfig;
 use crate::metrics_extraction::transactions::TransactionMetricsConfig;
 use crate::metrics_extraction::TaggingRule;
 use crate::statsd::RelayCounters;
@@ -47,16 +48,21 @@ pub enum Expiry {
     Expired,
 }
 
-/// Features exposed by project config
+/// Features exposed by project config.
 #[derive(Clone, Copy, Debug, Eq, PartialEq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
 pub enum Feature {
-    #[serde(rename = "organizations:metrics-extraction")]
-    MetricsExtraction,
-
+    /// Enables ingestion and normalization of profiles.
     #[serde(rename = "organizations:profiling")]
     Profiling,
 
-    /// forward compatibility
+    /// Unused.
+    ///
+    /// This used to control the initial experimental metrics extraction for sessions and has been
+    /// discontinued.
+    #[serde(rename = "organizations:metrics-extraction")]
+    Deprecated1,
+
+    /// Forward compatibility.
     #[serde(other)]
     Unknown,
 }
@@ -91,7 +97,10 @@ pub struct ProjectConfig {
     /// Configuration for operation breakdown. Will be emitted only if present.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub breakdowns_v2: Option<BreakdownsConfig>,
-    /// Configuration in relation to extracting metrics from transaction events.
+    /// Configuration for extracting metrics from sessions.
+    #[serde(skip_serializing_if = "SessionMetricsConfig::is_disabled")]
+    pub session_metrics: SessionMetricsConfig,
+    /// Configuration for extracting metrics from transaction events.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub transaction_metrics: Option<ErrorBoundary<TransactionMetricsConfig>>,
     /// The span attributes configuration.
@@ -117,6 +126,7 @@ impl Default for ProjectConfig {
             quotas: Vec::new(),
             dynamic_sampling: None,
             breakdowns_v2: None,
+            session_metrics: SessionMetricsConfig::default(),
             transaction_metrics: None,
             span_attributes: BTreeSet::new(),
             metric_conditional_tagging: Vec::new(),

diff --git a/relay-server/src/metrics_extraction/sessions.rs b/relay-server/src/metrics_extraction/sessions.rs
@@ -6,6 +6,46 @@ use relay_metrics::{DurationUnit, Metric, MetricUnit, MetricValue};
 
 use super::utils::with_tag;
 
+/// Namespace of session metricsfor the MRI.
+const METRIC_NAMESPACE: &str = "sessions";
+
+/// Current version of metrics extraction.
+const EXTRACT_VERSION: u16 = 1;
+
+/// Configuration for metric extraction from sessions.
+#[derive(Debug, Clone, Copy, Default, serde::Deserialize, serde::Serialize)]
+#[serde(default, rename_all = "camelCase")]
+pub struct SessionMetricsConfig {
+    /// The revision of the extraction algorithm.
+    ///
+    /// Provided the revision is lower than or equal to the revision supported by this Relay,
+    /// metrics are extracted. If the revision is higher than what this Relay supports, it does not
+    /// extract metrics from sessions, and instead forwards them to the upstream.
+    ///
+    /// Version `0` (default) disables extraction.
+    version: u16,
+
+    /// Drop sessions after successfully extracting metrics.
+    drop: bool,
+}
+
+impl SessionMetricsConfig {
+    /// Returns `true` if session metrics is enabled and compatible.
+    pub fn is_enabled(&self) -> bool {
+        self.version > 0 && self.version <= EXTRACT_VERSION
+    }
+
+    /// Returns `true` if Relay should not extract metrics from sessions.
+    pub fn is_disabled(&self) -> bool {
+        !self.is_enabled()
+    }
+
+    /// Returns `true` if the session should be dropped after extracting metrics.
+    pub fn should_drop(&self) -> bool {
+        self.drop
+    }
+}
+
 /// Convert contained nil UUIDs to None
 fn nil_to_none(distinct_id: Option<&String>) -> Option<&String> {
     let distinct_id = distinct_id?;
@@ -18,8 +58,6 @@ fn nil_to_none(distinct_id: Option<&String>) -> Option<&String> {
     Some(distinct_id)
 }
 
-const METRIC_NAMESPACE: &str = "sessions";
-
 pub fn extract_session_metrics<T: SessionLike>(
     attributes: &SessionAttributes,
     session: &T,

diff --git a/tests/integration/test_metrics.py b/tests/integration/test_metrics.py
@@ -196,7 +196,7 @@ def test_session_metrics_non_processing(
 
     if extract_metrics:
         # enable metrics extraction for the project
-        extra_config = {"config": {"features": ["organizations:metrics-extraction"]}}
+        extra_config = {"config": {"sessionMetrics": {"version": 1}}}
     else:
         extra_config = {}
 
@@ -322,7 +322,7 @@ def test_metrics_extracted_only_once(
     )
 
     # enable metrics extraction for the project
-    extra_config = {"config": {"features": ["organizations:metrics-extraction"]}}
+    extra_config = {"config": {"sessionMetrics": {"version": 1}}}
 
     project_id = 42
     mini_sentry.add_full_project_config(project_id, extra=extra_config)
@@ -358,7 +358,7 @@ def test_session_metrics_processing(
     project_id = 42
 
     # enable metrics extraction for the project
-    extra_config = {"config": {"features": ["organizations:metrics-extraction"]}}
+    extra_config = {"config": {"sessionMetrics": {"version": 1}}}
 
     mini_sentry.add_full_project_config(project_id, extra=extra_config)
 
@@ -466,7 +466,8 @@ def test_transaction_metrics(
     config = mini_sentry.project_configs[project_id]["config"]
     timestamp = datetime.now(tz=timezone.utc)
 
-    config["features"] = ["organizations:metrics-extraction"] if extract_metrics else []
+    if extract_metrics:
+        config["sessionMetrics"] = {"version": 1}
     config["breakdownsV2"] = {
         "span_ops": {"type": "spanOperations", "matches": ["react.mount"]}
     }