feat(relay): Simplify Proxy Mode (#5165)

This is a big change, however this should not break anything. Users that
run Relay in Proxy mode should not experience any negative side effects.

With the recent changes to the Relay modes
(#5108,
#5057,
#5053) it became apparent that
the proxy mode is currently doing more than strictly necessary. That is,
the logic is currently more complex than it needs to be and as such it
also is less efficient than it could be. This PR tries to rectify this
by making Proxy mode more of a 'pure' Proxy by introducing
the `ProxyProcessorService` which is a simplified version of
the `EnvelopeProcessorService`. There are also some changes to the
`HealthCheckService`, `AutoscalingMetricService` and `RelayStats` to
make these compatible.

Fixes: https://linear.app/getsentry/issue/INGEST-581/simplify-proxy-mode

Author: tobias-wilfert

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+**Breaking Changes**:
+
+- Simplify proxy mode to forward without processing. ([#5165](https://github.com/getsentry/relay/pull/5165))
+
 **Bug Fixes**:
 
 - Make referer optional in Vercel Log Drain Transform. ([#5273](https://github.com/getsentry/relay/pull/5273))

relay-server/src/endpoints/autoscaling.rs

@@ -6,11 +6,14 @@ use std::fmt::Write;
 
 /// Returns internal metrics data for relay.
 pub async fn handle(state: ServiceState) -> (StatusCode, String) {
-    let data = match state
-        .autoscaling()
-        .send(AutoscalingMessageKind::Check)
-        .await
-    {
+    let Some(autoscaling) = state.autoscaling() else {
+        return (
+            StatusCode::NOT_FOUND,
+            "Autoscaling metrics not enabled".to_owned(),
+        );
+    };
+
+    let data = match autoscaling.send(AutoscalingMessageKind::Check).await {
         Ok(data) => data,
         Err(_) => {
             return (

relay-server/src/service.rs

@@ -20,6 +20,7 @@ use crate::services::processor::{
 };
 use crate::services::projects::cache::{ProjectCacheHandle, ProjectCacheService};
 use crate::services::projects::source::ProjectSource;
+use crate::services::proxy_processor::{ProxyAddrs, ProxyProcessorService};
 use crate::services::relays::{RelayCache, RelayCacheService};
 use crate::services::stats::RelayStats;
 #[cfg(feature = "processing")]
@@ -72,7 +73,7 @@ pub struct Registry {
     pub upstream_relay: Addr<UpstreamRelay>,
     pub envelope_buffer: PartitionedEnvelopeBuffer,
     pub project_cache_handle: ProjectCacheHandle,
-    pub autoscaling: Addr<AutoscalingMetrics>,
+    pub autoscaling: Option<Addr<AutoscalingMetrics>>,
 }
 
 /// Constructs a Tokio [`relay_system::Runtime`] configured for running [services](relay_system::Service).
@@ -181,7 +182,11 @@ impl ServiceState {
 
         // Create an address for the `EnvelopeProcessor`, which can be injected into the
         // other services.
-        let (processor, processor_rx) = channel(EnvelopeProcessorService::name());
+        let (processor, processor_rx) = match config.relay_mode() {
+            relay_config::RelayMode::Proxy => channel(ProxyProcessorService::name()),
+            relay_config::RelayMode::Managed => channel(EnvelopeProcessorService::name()),
+        };
+
         let outcome_producer = services.start(OutcomeProducerService::create(
             config.clone(),
             upstream_relay.clone(),
@@ -209,16 +214,6 @@ impl ServiceState {
         let project_cache_handle =
             ProjectCacheService::new(Arc::clone(&config), project_source).start_in(services);
 
-        let aggregator = RouterService::new(
-            handle.clone(),
-            config.default_aggregator_config().clone(),
-            config.secondary_aggregator_configs().clone(),
-            Some(processor.clone().recipient()),
-            project_cache_handle.clone(),
-        );
-        let aggregator_handle = aggregator.handle();
-        let aggregator = services.start(aggregator);
-
         let metric_outcomes = MetricOutcomes::new(outcome_aggregator.clone());
 
         #[cfg(feature = "processing")]
@@ -238,38 +233,11 @@ impl ServiceState {
             })
             .transpose()?;
 
-        let cogs = CogsService::new(&config);
-        let cogs = Cogs::new(CogsServiceRecorder::new(&config, services.start(cogs)));
-
         #[cfg(feature = "processing")]
         let global_rate_limits = redis_clients
             .as_ref()
             .map(|p| services.start(GlobalRateLimitsService::new(p.quotas.clone())));
 
-        let processor_pool = create_processor_pool(&config)?;
-        services.start_with(
-            EnvelopeProcessorService::new(
-                processor_pool.clone(),
-                config.clone(),
-                global_config_handle,
-                project_cache_handle.clone(),
-                cogs,
-                #[cfg(feature = "processing")]
-                redis_clients.clone(),
-                processor::Addrs {
-                    outcome_aggregator: outcome_aggregator.clone(),
-                    upstream_relay: upstream_relay.clone(),
-                    #[cfg(feature = "processing")]
-                    store_forwarder: store.clone(),
-                    aggregator: aggregator.clone(),
-                    #[cfg(feature = "processing")]
-                    global_rate_limits,
-                },
-                metric_outcomes.clone(),
-            ),
-            processor_rx,
-        );
-
         let envelope_buffer = PartitionedEnvelopeBuffer::create(
             config.spool_partitions(),
             config.clone(),
@@ -281,6 +249,75 @@ impl ServiceState {
             services,
         );
 
+        let (processor_pool, aggregator_handle, autoscaling) = match config.relay_mode() {
+            relay_config::RelayMode::Proxy => {
+                services.start_with(
+                    ProxyProcessorService::new(
+                        config.clone(),
+                        project_cache_handle.clone(),
+                        ProxyAddrs {
+                            outcome_aggregator: outcome_aggregator.clone(),
+                            upstream_relay: upstream_relay.clone(),
+                        },
+                    ),
+                    processor_rx,
+                );
+                (None, None, None)
+            }
+            relay_config::RelayMode::Managed => {
+                let processor_pool = create_processor_pool(&config)?;
+
+                let aggregator = RouterService::new(
+                    handle.clone(),
+                    config.default_aggregator_config().clone(),
+                    config.secondary_aggregator_configs().clone(),
+                    Some(processor.clone().recipient()),
+                    project_cache_handle.clone(),
+                );
+                let aggregator_handle = aggregator.handle();
+                let aggregator = services.start(aggregator);
+
+                let cogs = CogsService::new(&config);
+                let cogs = Cogs::new(CogsServiceRecorder::new(&config, services.start(cogs)));
+
+                services.start_with(
+                    EnvelopeProcessorService::new(
+                        processor_pool.clone(),
+                        config.clone(),
+                        global_config_handle,
+                        project_cache_handle.clone(),
+                        cogs,
+                        #[cfg(feature = "processing")]
+                        redis_clients.clone(),
+                        processor::Addrs {
+                            outcome_aggregator: outcome_aggregator.clone(),
+                            upstream_relay: upstream_relay.clone(),
+                            #[cfg(feature = "processing")]
+                            store_forwarder: store.clone(),
+                            aggregator: aggregator.clone(),
+                            #[cfg(feature = "processing")]
+                            global_rate_limits,
+                        },
+                        metric_outcomes.clone(),
+                    ),
+                    processor_rx,
+                );
+
+                let autoscaling = services.start(AutoscalingMetricService::new(
+                    memory_stat.clone(),
+                    envelope_buffer.clone(),
+                    handle.clone(),
+                    processor_pool.clone(),
+                ));
+
+                (
+                    Some(processor_pool),
+                    Some(aggregator_handle),
+                    Some(autoscaling),
+                )
+            }
+        };
+
         let health_check = services.start(HealthCheckService::new(
             config.clone(),
             MemoryChecker::new(memory_stat.clone(), config.clone()),
@@ -289,13 +326,6 @@ impl ServiceState {
             envelope_buffer.clone(),
         ));
 
-        let autoscaling = services.start(AutoscalingMetricService::new(
-            memory_stat.clone(),
-            envelope_buffer.clone(),
-            handle.clone(),
-            processor_pool.clone(),
-        ));
-
         services.start(RelayStats::new(
             config.clone(),
             handle.clone(),
@@ -348,8 +378,8 @@ impl ServiceState {
         &self.inner.memory_checker
     }
 
-    pub fn autoscaling(&self) -> &Addr<AutoscalingMetrics> {
-        &self.inner.registry.autoscaling
+    pub fn autoscaling(&self) -> Option<&Addr<AutoscalingMetrics>> {
+        self.inner.registry.autoscaling.as_ref()
     }
 
     /// Returns the V2 envelope buffer, if present.

relay-server/src/services/health_check.rs

@@ -84,7 +84,7 @@ impl StatusUpdate {
 pub struct HealthCheckService {
     config: Arc<Config>,
     memory_checker: MemoryChecker,
-    aggregator: RouterHandle,
+    aggregator: Option<RouterHandle>,
     upstream_relay: Addr<UpstreamRelay>,
     envelope_buffer: PartitionedEnvelopeBuffer,
 }
@@ -94,7 +94,7 @@ impl HealthCheckService {
     pub fn new(
         config: Arc<Config>,
         memory_checker: MemoryChecker,
-        aggregator: RouterHandle,
+        aggregator: Option<RouterHandle>,
         upstream_relay: Addr<UpstreamRelay>,
         envelope_buffer: PartitionedEnvelopeBuffer,
     ) -> Self {
@@ -145,7 +145,10 @@ impl HealthCheckService {
     }
 
     async fn aggregator_probe(&self) -> Status {
-        Status::from(self.aggregator.can_accept_metrics())
+        self.aggregator
+            .as_ref()
+            .map(|agg| Status::from(agg.can_accept_metrics()))
+            .unwrap_or(Status::Healthy)
     }
 
     async fn spool_health_probe(&self) -> Status {

relay-server/src/services/mod.rs

@@ -38,6 +38,7 @@ pub mod outcome;
 pub mod outcome_aggregator;
 pub mod processor;
 pub mod projects;
+pub mod proxy_processor;
 pub mod relays;
 pub mod server;
 pub mod stats;

relay-server/src/services/processor.rs

@@ -3358,7 +3358,7 @@ impl RateLimiter {
     }
 }
 
-fn encode_payload(body: &Bytes, http_encoding: HttpEncoding) -> Result<Bytes, std::io::Error> {
+pub fn encode_payload(body: &Bytes, http_encoding: HttpEncoding) -> Result<Bytes, std::io::Error> {
     let envelope_body: Vec<u8> = match http_encoding {
         HttpEncoding::Identity => return Ok(body.clone()),
         HttpEncoding::Deflate => {
@@ -3392,10 +3392,10 @@ fn encode_payload(body: &Bytes, http_encoding: HttpEncoding) -> Result<Bytes, st
 /// An upstream request that submits an envelope via HTTP.
 #[derive(Debug)]
 pub struct SendEnvelope {
-    envelope: TypedEnvelope<Processed>,
-    body: Bytes,
-    http_encoding: HttpEncoding,
-    project_cache: ProjectCacheHandle,
+    pub envelope: TypedEnvelope<Processed>,
+    pub body: Bytes,
+    pub http_encoding: HttpEncoding,
+    pub project_cache: ProjectCacheHandle,
 }
 
 impl UpstreamRequest for SendEnvelope {