feat(relay): Simplify Proxy Mode

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 {