ref(outcomes): Remove outcomes aggregator, implement forwarding of metrics as client reports

CHANGELOG.md

@@ -9,6 +9,7 @@
 **Internal**:
 
 - Require timestamps and verification in auth signatures. ([#6069](https://github.com/getsentry/relay/pull/6069))
+- Internally handle outcomes as metrics. ([#6107](https://github.com/getsentry/relay/pull/6107))
 - Have relay generate metric billing outcomes. ([#6066](https://github.com/getsentry/relay/pull/6066))
 - Update sentry-conventions to 0.12.0.
 

relay-config/src/config.rs

@@ -1233,25 +1233,6 @@ pub enum NormalizationLevel {
     Full,
 }
 
-/// Configuration values for the outcome aggregator
-#[derive(Serialize, Deserialize, Debug)]
-#[serde(default)]
-pub struct OutcomeAggregatorConfig {
-    /// Defines the width of the buckets into which outcomes are aggregated, in seconds.
-    pub bucket_interval: u64,
-    /// Defines how often all buckets are flushed, in seconds.
-    pub flush_interval: u64,
-}
-
-impl Default for OutcomeAggregatorConfig {
-    fn default() -> Self {
-        Self {
-            bucket_interval: 60,
-            flush_interval: 120,
-        }
-    }
-}
-
 /// Configuration options for objectstore's auth scheme.
 #[derive(Serialize, Deserialize)]
 pub struct ObjectstoreAuthConfig {
@@ -1418,8 +1399,6 @@ pub struct Outcomes {
     /// Defines the source string registered in the outcomes originating from
     /// this Relay (typically something like the region or the layer).
     pub source: Option<String>,
-    /// Configures the outcome aggregator.
-    pub aggregator: OutcomeAggregatorConfig,
 }
 
 impl Default for Outcomes {
@@ -1429,7 +1408,6 @@ impl Default for Outcomes {
             batch_size: 1000,
             batch_interval: 500,
             source: None,
-            aggregator: OutcomeAggregatorConfig::default(),
         }
     }
 }
@@ -2163,11 +2141,6 @@ impl Config {
         self.values.outcomes.source.as_deref()
     }
 
-    /// Returns the width of the buckets into which outcomes are aggregated, in seconds.
-    pub fn outcome_aggregator(&self) -> &OutcomeAggregatorConfig {
-        &self.values.outcomes.aggregator
-    }
-
     /// Returns logging configuration.
     pub fn logging(&self) -> &relay_log::LogConfig {
         &self.values.logging

relay-server/src/service.rs

@@ -18,7 +18,8 @@ use crate::services::objectstore::Objectstore;
 #[cfg(feature = "processing")]
 use crate::services::objectstore::ObjectstoreService;
 use crate::services::outcome::{
-    OutcomeAggregator, OutcomeProducer, OutcomeProducerService, TrackOutcome,
+    ClientReportOutcomeProducerService, NullOutcomeProducerService, OutcomeProducerService,
+    TrackOutcome,
 };
 use crate::services::processor::{
     self, EnvelopeProcessor, EnvelopeProcessorService, EnvelopeProcessorServicePool,
@@ -39,7 +40,7 @@ use anyhow::Result;
 use axum::extract::FromRequestParts;
 use axum::http::request::Parts;
 use relay_cogs::Cogs;
-use relay_config::Config;
+use relay_config::{Config, EmitOutcomes, RelayMode};
 #[cfg(feature = "processing")]
 use relay_config::{RedisConfigRef, RedisConfigsRef};
 #[cfg(feature = "processing")]
@@ -73,7 +74,6 @@ pub enum ServiceError {
 #[derive(Clone, Debug)]
 pub struct Registry {
     pub health_check: Addr<HealthCheck>,
-    pub outcome_producer: Addr<OutcomeProducer>,
     pub outcome_aggregator: Addr<TrackOutcome>,
     pub processor: Addr<EnvelopeProcessor>,
     pub relay_cache: Addr<RelayCache>,
@@ -195,19 +195,25 @@ impl ServiceState {
         // Create an address for the `EnvelopeProcessor`, which can be injected into the
         // other services.
         let (processor, processor_rx) = match config.relay_mode() {
-            relay_config::RelayMode::Proxy => channel(ProxyProcessorService::name()),
-            relay_config::RelayMode::Managed => channel(EnvelopeProcessorService::name()),
+            RelayMode::Proxy => channel(ProxyProcessorService::name()),
+            RelayMode::Managed => channel(EnvelopeProcessorService::name()),
         };
 
         let (aggregator, aggregator_rx) = channel(RouterService::name());
 
-        let outcome_producer = services.start(OutcomeProducerService::create(
-            config.clone(),
-            processor.clone(),
-            aggregator.clone(),
-        )?);
-        let outcome_aggregator =
-            services.start(OutcomeAggregator::new(&config, outcome_producer.clone()));
+        let outcome_aggregator = match config.emit_outcomes() {
+            EmitOutcomes::None => services.start(NullOutcomeProducerService::new()),
+            _ => match config.relay_mode() {
+                RelayMode::Proxy => services.start(ClientReportOutcomeProducerService::new(
+                    &config,
+                    processor.clone(),
+                )),
+                RelayMode::Managed => services.start(OutcomeProducerService::new(
+                    Arc::clone(&config),
+                    aggregator.clone(),
+                )),
+            },
+        };
 
         let (global_config, global_config_rx) =
             GlobalConfigService::new(config.clone(), upstream_relay.clone());
@@ -267,7 +273,7 @@ impl ServiceState {
         );
 
         let (processor_pool, aggregator_handle, autoscaling) = match config.relay_mode() {
-            relay_config::RelayMode::Proxy => {
+            RelayMode::Proxy => {
                 services.start_with(
                     ProxyProcessorService::new(
                         config.clone(),
@@ -281,7 +287,7 @@ impl ServiceState {
                 );
                 (None, None, None)
             }
-            relay_config::RelayMode::Managed => {
+            RelayMode::Managed => {
                 let processor_pool = create_processor_pool(&config)?;
 
                 let router = RouterService::new(
@@ -365,7 +371,6 @@ impl ServiceState {
         let registry = Registry {
             processor,
             health_check,
-            outcome_producer,
             outcome_aggregator,
             relay_cache,
             global_config,
@@ -426,17 +431,12 @@ impl ServiceState {
         &self.inner.registry.health_check
     }
 
-    /// Returns the address of the [`OutcomeProducer`] service.
-    pub fn outcome_producer(&self) -> &Addr<OutcomeProducer> {
-        &self.inner.registry.outcome_producer
-    }
-
-    /// Returns the address of the [`OutcomeProducer`] service.
+    /// Returns the address of the [`UpstreamRelay`] service.
     pub fn upstream_relay(&self) -> &Addr<UpstreamRelay> {
         &self.inner.registry.upstream_relay
     }
 
-    /// Returns the address of the [`OutcomeProducer`] service.
+    /// Returns the address of the [`EnvelopeProcessor`] service.
     pub fn processor(&self) -> &Addr<EnvelopeProcessor> {
         &self.inner.registry.processor
     }
@@ -446,7 +446,7 @@ impl ServiceState {
         &self.inner.registry.global_config
     }
 
-    /// Returns the address of the [`OutcomeProducer`] service.
+    /// Returns the address of the [`TrackOutcome`] service.
     pub fn outcome_aggregator(&self) -> &Addr<TrackOutcome> {
         &self.inner.registry.outcome_aggregator
     }