feat(metrics): Support dedicated topics per metrics usecase, drop metrics from unknown usecases [INGEST-1309]

CHANGELOG.md

@@ -24,6 +24,7 @@
 - Avoid potential panics when scrubbing minidumps. ([#1282](https://github.com/getsentry/relay/pull/1282))
 - Fix typescript profile validation. ([#1283](https://github.com/getsentry/relay/pull/1283))
 - Track memory footprint of metrics buckets. ([#1284](https://github.com/getsentry/relay/pull/1284), [#1287](https://github.com/getsentry/relay/pull/1287), [#1288](https://github.com/getsentry/relay/pull/1288))
+- Support dedicated topics per metrics usecase, drop metrics from unknown usecases. ([#1285](https://github.com/getsentry/relay/pull/1285))
 
 ## 22.5.0
 

relay-config/src/config.rs

@@ -779,8 +779,10 @@ pub enum KafkaTopic {
     OutcomesBilling,
     /// Session health updates.
     Sessions,
-    /// Aggregate Metrics.
-    Metrics,
+    /// Any metric that is extracted from sessions.
+    MetricsSessions,
+    /// Any metric that is extracted from transactions.
+    MetricsTransactions,
     /// Profiles
     Profiles,
 }
@@ -801,8 +803,14 @@ pub struct TopicAssignments {
     pub outcomes_billing: Option<TopicAssignment>,
     /// Session health topic name.
     pub sessions: TopicAssignment,
-    /// Metrics topic name.
+    /// Default topic name for all aggregate metrics. Specialized topics for session-based and
+    /// transaction-based metrics can be configured via `metrics_sessions` and
+    /// `metrics_transactions` each.
     pub metrics: TopicAssignment,
+    /// Topic name for metrics extracted from sessions. Defaults to the assignment of `metrics`.
+    pub metrics_sessions: Option<TopicAssignment>,
+    /// Topic name for metrics extracted from transactions. Defaults to the assignment of `metrics`.
+    pub metrics_transactions: Option<TopicAssignment>,
     /// Stacktrace topic name
     pub profiles: TopicAssignment,
 }
@@ -817,7 +825,10 @@ impl TopicAssignments {
             KafkaTopic::Outcomes => &self.outcomes,
             KafkaTopic::OutcomesBilling => self.outcomes_billing.as_ref().unwrap_or(&self.outcomes),
             KafkaTopic::Sessions => &self.sessions,
-            KafkaTopic::Metrics => &self.metrics,
+            KafkaTopic::MetricsSessions => self.metrics_sessions.as_ref().unwrap_or(&self.metrics),
+            KafkaTopic::MetricsTransactions => {
+                self.metrics_transactions.as_ref().unwrap_or(&self.metrics)
+            }
             KafkaTopic::Profiles => &self.profiles,
         }
     }
@@ -833,6 +844,8 @@ impl Default for TopicAssignments {
             outcomes_billing: None,
             sessions: "ingest-sessions".to_owned().into(),
             metrics: "ingest-metrics".to_owned().into(),
+            metrics_sessions: None,
+            metrics_transactions: None,
             profiles: "profiles".to_owned().into(),
         }
     }