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

Author: Dav1dde

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

@@ -17,9 +17,7 @@ use crate::services::metrics::RouterService;
 use crate::services::objectstore::Objectstore;
 #[cfg(feature = "processing")]
 use crate::services::objectstore::ObjectstoreService;
-use crate::services::outcome::{
-    OutcomeAggregator, OutcomeProducer, OutcomeProducerService, TrackOutcome,
-};
+use crate::services::outcome::{OutcomeProducerService, TrackOutcome};
 use crate::services::processor::{
     self, EnvelopeProcessor, EnvelopeProcessorService, EnvelopeProcessorServicePool,
 };
@@ -73,7 +71,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>,
@@ -201,13 +198,10 @@ impl ServiceState {
 
         let (aggregator, aggregator_rx) = channel(RouterService::name());
 
-        let outcome_producer = services.start(OutcomeProducerService::create(
-            config.clone(),
-            processor.clone(),
+        let outcome_aggregator = services.start(OutcomeProducerService::new(
+            Arc::clone(&config),
             aggregator.clone(),
-        )?);
-        let outcome_aggregator =
-            services.start(OutcomeAggregator::new(&config, outcome_producer.clone()));
+        ));
 
         let (global_config, global_config_rx) =
             GlobalConfigService::new(config.clone(), upstream_relay.clone());
@@ -365,7 +359,6 @@ impl ServiceState {
         let registry = Registry {
             processor,
             health_check,
-            outcome_producer,
             outcome_aggregator,
             relay_cache,
             global_config,
@@ -426,17 +419,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 +434,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
     }