Move eap switch to store

Author: noahsmartin

relay-dynamic-config/src/feature.rs

@@ -138,15 +138,6 @@ pub enum Feature {
     /// Enable the experimental Trace Attachment pipeline in Relay.
     #[serde(rename = "projects:trace-attachment-processing")]
     TraceAttachmentProcessing,
-    /// Enable EAP (Event Analytics Platform) double-write for user sessions.
-    ///
-    /// When enabled, session data is sent both through the legacy metrics pipeline
-    /// and directly to the snuba-items topic as TRACE_ITEM_TYPE_USER_SESSION.
-    /// This enables migration to the new EAP-based user sessions storage.
-    ///
-    /// Serialized as `organizations:user-sessions-eap`.
-    #[serde(rename = "organizations:user-sessions-eap")]
-    UserSessionsEap,
     /// Forward compatibility.
     #[doc(hidden)]
     #[serde(other)]

relay-dynamic-config/src/global.rs

@@ -176,6 +176,20 @@ pub struct Options {
     )]
     pub objectstore_attachments_sample_rate: f32,
 
+    /// Rollout rate for the EAP (Event Analytics Platform) double-write for user sessions.
+    ///
+    /// When rolled out, session data is sent both through the legacy metrics pipeline
+    /// and directly to the snuba-items topic as TRACE_ITEM_TYPE_USER_SESSION.
+    ///
+    /// Rate needs to be between `0.0` and `1.0`.
+    /// Rollout is determined deterministically per organization ID.
+    #[serde(
+        rename = "relay.sessions-eap.rollout-rate",
+        deserialize_with = "default_on_error",
+        skip_serializing_if = "is_default"
+    )]
+    pub sessions_eap_rollout_rate: f32,
+
     /// All other unknown options.
     #[serde(flatten)]
     other: HashMap<String, Value>,

relay-server/src/processing/sessions/mod.rs

@@ -104,7 +104,6 @@ impl processing::Processor for SessionsProcessor {
         let sessions = self.limiter.enforce_quotas(sessions, ctx).await?;
 
         let sessions = process::extract_metrics(sessions, ctx);
-
         Ok(Output::metrics(sessions))
     }
 }

relay-server/src/services/processor.rs

@@ -2449,18 +2449,18 @@ impl EnvelopeProcessorService {
 
     /// Processes metric buckets and sends them to kafka.
     ///
-    /// Applies rate limiting and cardinality limiting, then submits to the store.
-    /// If `UserSessionsEap` is enabled, session metrics are also sent to EAP.
+    /// This function runs the following steps:
+    ///  - cardinality limiting
+    ///  - rate limiting
+    ///  - submit to `StoreForwarder`
     #[cfg(feature = "processing")]
     async fn encode_metrics_processing(
         &self,
         message: FlushBuckets,
         store_forwarder: &Addr<Store>,
     ) {
         use crate::constants::DEFAULT_EVENT_RETENTION;
-        use crate::services::store::{StoreMetrics, StoreSessionMetricsEap};
-        use relay_dynamic_config::Feature;
-        use relay_metrics::MetricNamespace;
+        use crate::services::store::StoreMetrics;
 
         for ProjectBuckets {
             buckets,
@@ -2487,28 +2487,13 @@ impl EnvelopeProcessorService {
                 .event_retention
                 .unwrap_or(DEFAULT_EVENT_RETENTION);
 
-            // Always send all buckets to the legacy metrics path.
+            // The store forwarder takes care of bucket splitting internally, so we can submit the
+            // entire list of buckets. There is no batching needed here.
             store_forwarder.send(StoreMetrics {
-                buckets: buckets.clone(),
+                buckets,
                 scoping,
                 retention,
             });
-
-            // If EAP double-write is enabled, also send session metrics to EAP.
-            if project_info.config.features.has(Feature::UserSessionsEap) {
-                let session_buckets: Vec<_> = buckets
-                    .into_iter()
-                    .filter(|b| b.name.namespace() == MetricNamespace::Sessions)
-                    .collect();
-
-                if !session_buckets.is_empty() {
-                    store_forwarder.send(StoreSessionMetricsEap {
-                        buckets: session_buckets,
-                        scoping,
-                        retention,
-                    });
-                }
-            }
         }
     }
 

relay-server/src/services/store.rs

@@ -103,16 +103,6 @@ pub struct StoreMetrics {
     pub retention: u16,
 }
 
-/// Publishes session metric buckets to EAP via the snuba-items topic.
-///
-/// Used when the `UserSessionsEap` feature flag is enabled for double-write.
-#[derive(Clone, Debug)]
-pub struct StoreSessionMetricsEap {
-    pub buckets: Vec<Bucket>,
-    pub scoping: Scoping,
-    pub retention: u16,
-}
-
 /// Publishes a log item to the Sentry core application through Kafka.
 #[derive(Debug)]
 pub struct StoreTraceItem {
@@ -185,8 +175,6 @@ pub enum Store {
     Envelope(StoreEnvelope),
     /// Aggregated generic metrics.
     Metrics(StoreMetrics),
-    /// Session metrics routed to EAP (Event Analytics Platform).
-    SessionMetricsEap(StoreSessionMetricsEap),
     /// A singular [`TraceItem`].
     TraceItem(Managed<StoreTraceItem>),
     /// A singular Span.
@@ -201,7 +189,6 @@ impl Store {
         match self {
             Store::Envelope(_) => "envelope",
             Store::Metrics(_) => "metrics",
-            Store::SessionMetricsEap(_) => "session_metrics_eap",
             Store::TraceItem(_) => "log",
             Store::Span(_) => "span",
             Store::ProfileChunk(_) => "profile_chunk",
@@ -227,14 +214,6 @@ impl FromMessage<StoreMetrics> for Store {
     }
 }
 
-impl FromMessage<StoreSessionMetricsEap> for Store {
-    type Response = NoResponse;
-
-    fn from_message(message: StoreSessionMetricsEap, _: ()) -> Self {
-        Self::SessionMetricsEap(message)
-    }
-}
-
 impl FromMessage<Managed<StoreTraceItem>> for Store {
     type Response = NoResponse;
 
@@ -294,7 +273,6 @@ impl StoreService {
             match message {
                 Store::Envelope(message) => self.handle_store_envelope(message),
                 Store::Metrics(message) => self.handle_store_metrics(message),
-                Store::SessionMetricsEap(message) => self.handle_store_session_metrics_eap(message),
                 Store::TraceItem(message) => self.handle_store_trace_item(message),
                 Store::Span(message) => self.handle_store_span(message),
                 Store::ProfileChunk(message) => self.handle_store_profile_chunk(message),
@@ -543,9 +521,22 @@ impl StoreService {
         let global_config = self.global_config.current();
         let mut encoder = BucketEncoder::new(&global_config);
 
+        // Check if this organization is rolled out for sessions EAP double-write.
+        let sessions_eap_rollout_rate = global_config.options.sessions_eap_rollout_rate;
+        let emit_sessions_to_eap =
+            utils::is_rolled_out(scoping.organization_id.value(), sessions_eap_rollout_rate)
+                .is_keep();
+
         let now = UnixTimestamp::now();
         let mut delay_stats = ByNamespace::<(u64, u64, u64)>::default();
 
+        // Collect session buckets for EAP if rolled out.
+        let mut session_buckets = if emit_sessions_to_eap {
+            Vec::new()
+        } else {
+            Vec::with_capacity(0)
+        };
+
         for mut bucket in buckets {
             let namespace = encoder.prepare(&mut bucket);
 
@@ -557,6 +548,11 @@ impl StoreService {
                 *max = (*max).max(delay);
             }
 
+            // Collect session buckets for EAP double-write.
+            if emit_sessions_to_eap && namespace == MetricNamespace::Sessions {
+                session_buckets.push(bucket.clone());
+            }
+
             // Create a local bucket view to avoid splitting buckets unnecessarily. Since we produce
             // each bucket separately, we only need to split buckets that exceed the size, but not
             // batches.
@@ -588,6 +584,11 @@ impl StoreService {
             }
         }
 
+        // Also emit session buckets to EAP if rolled out.
+        if !session_buckets.is_empty() {
+            self.emit_session_metrics_to_eap(&session_buckets, scoping, retention);
+        }
+
         if let Some(error) = error {
             relay_log::error!(
                 error = &error as &dyn std::error::Error,
@@ -614,8 +615,11 @@ impl StoreService {
         }
     }
 
-    /// Sends session metric buckets to EAP (snuba-items topic) as TraceItems.
-    fn handle_store_session_metrics_eap(&self, message: StoreSessionMetricsEap) {
+    /// Emits session metric buckets to EAP (snuba-items topic) as TraceItems.
+    ///
+    /// This is called when the organization is rolled out for sessions EAP double-write
+    /// via `sessions_eap_rollout_rate`.
+    fn emit_session_metrics_to_eap(&self, buckets: &[Bucket], scoping: Scoping, retention: u16) {
         use sentry_protos::snuba::v1::{AnyValue, TraceItem, TraceItemType, any_value};
         use std::collections::HashMap;
 
@@ -624,11 +628,6 @@ impl StoreService {
             0x90, 0xa1,
         ]);
 
-        let StoreSessionMetricsEap {
-            buckets,
-            scoping,
-            retention,
-        } = message;
         let now = UnixTimestamp::now();
         let mut error = None;
         let mut row_count: u64 = 0;
@@ -696,8 +695,8 @@ impl StoreService {
                     trace_id: uuid.to_string(),
                     item_id: uuid.as_bytes()[..16].into(),
                     item_type: i32::from(TraceItemType::UserSession),
-                    timestamp: timestamp.clone(),
-                    received: received.clone(),
+                    timestamp,
+                    received,
                     retention_days: retention.into(),
                     downsampled_retention_days: retention.into(),
                     attributes,

relay-server/src/statsd.rs

@@ -826,8 +826,8 @@ pub enum RelayCounters {
     SpanV2Produced,
     /// Number of session metric buckets produced to the EAP snuba-items topic.
     ///
-    /// This metric is only emitted when the `UserSessionsEap` feature flag is enabled,
-    /// routing aggregated session metrics to EAP (Event Analytics Platform) instead of
+    /// This metric is only emitted when the organization is rolled out via `sessions_eap_rollout_rate`,
+    /// routing aggregated session metrics to EAP (Event Analytics Platform) in addition to
     /// the legacy ingest-metrics topic.
     ///
     /// This metric is tagged with: