feat(server): Delay serving global config in managed mode before upstream response

CHANGELOG.md

@@ -31,6 +31,7 @@
 - Validate span timestamps and IDs in light normalization on renormalization. ([#2679](https://github.com/getsentry/relay/pull/2679))
 - Scrub all DB Core Data spans differently. ([#2686](https://github.com/getsentry/relay/pull/2686))
 - Support generic metrics extraction version 2. ([#2692](https://github.com/getsentry/relay/pull/2692))
+- Postpone processing till the global config is available. ([#2697](https://github.com/getsentry/relay/pull/2697))
 - Emit error on continued project config fetch failures after a time interval. ([#2700](https://github.com/getsentry/relay/pull/2700))
 
 ## 23.10.1

relay-server/src/actors/global_config.rs

@@ -105,32 +105,59 @@ pub struct Subscribe;
 /// frequency from upstream.
 pub enum GlobalConfigManager {
     /// Returns the most recent global config.
-    Get(relay_system::Sender<Arc<GlobalConfig>>),
+    Get(relay_system::Sender<Status>),
     /// Returns a [`watch::Receiver`] where global config updates will be sent to.
-    Subscribe(relay_system::Sender<watch::Receiver<Arc<GlobalConfig>>>),
+    Subscribe(relay_system::Sender<watch::Receiver<Status>>),
 }
 
 impl Interface for GlobalConfigManager {}
 
 impl FromMessage<Get> for GlobalConfigManager {
-    type Response = AsyncResponse<Arc<GlobalConfig>>;
+    type Response = AsyncResponse<Status>;
 
-    fn from_message(_: Get, sender: relay_system::Sender<Arc<GlobalConfig>>) -> Self {
+    fn from_message(_: Get, sender: relay_system::Sender<Status>) -> Self {
         Self::Get(sender)
     }
 }
 
 impl FromMessage<Subscribe> for GlobalConfigManager {
-    type Response = AsyncResponse<watch::Receiver<Arc<GlobalConfig>>>;
+    type Response = AsyncResponse<watch::Receiver<Status>>;
 
-    fn from_message(
-        _: Subscribe,
-        sender: relay_system::Sender<watch::Receiver<Arc<GlobalConfig>>>,
-    ) -> Self {
+    fn from_message(_: Subscribe, sender: relay_system::Sender<watch::Receiver<Status>>) -> Self {
         Self::Subscribe(sender)
     }
 }
 
+/// Describes the current fetching status of the [`GlobalConfig`] from the upstream.
+#[derive(Debug, Clone, Default)]
+pub enum Status {
+    /// Global config ready to be used by other services.
+    ///
+    /// This variant implies different things in different circumstances. In managed mode, it means
+    /// that we have received a config from upstream. In other modes the config is either
+    /// from a file or the default global config.
+    Ready(Arc<GlobalConfig>),
+    /// The global config is requested from the upstream but it has not arrived yet.
+    ///
+    /// This variant should never be sent after the first `Ready` has occured.
+    #[default]
+    Waiting,
+}
+
+impl Status {
+    fn is_ready(&self) -> bool {
+        matches!(self, Self::Ready(_))
+    }
+
+    /// Similar to Option::unwrap_or_default.
+    pub fn get_ready_or_default(self) -> Arc<GlobalConfig> {
+        match self {
+            Status::Ready(global_config) => global_config,
+            Status::Waiting => Arc::default(),
+        }
+    }
+}
+
 /// Service implementing the [`GlobalConfigManager`] interface.
 ///
 /// The service offers two alternatives to fetch the [`GlobalConfig`]:
@@ -140,7 +167,7 @@ impl FromMessage<Subscribe> for GlobalConfigManager {
 pub struct GlobalConfigService {
     config: Arc<Config>,
     /// Sender of the [`watch`] channel for the subscribers of the service.
-    global_config_watch: watch::Sender<Arc<GlobalConfig>>,
+    global_config_watch: watch::Sender<Status>,
     /// Sender of the internal channel to forward global configs from upstream.
     internal_tx: mpsc::Sender<UpstreamQueryResult>,
     /// Receiver of the internal channel to forward global configs from upstream.
@@ -161,7 +188,7 @@ impl GlobalConfigService {
     /// Creates a new [`GlobalConfigService`].
     pub fn new(config: Arc<Config>, upstream: Addr<UpstreamRelay>) -> Self {
         let (internal_tx, internal_rx) = mpsc::channel(1);
-        let (global_config_watch, _) = watch::channel(Arc::default());
+        let (global_config_watch, _) = watch::channel(Status::Waiting);
 
         Self {
             config,
@@ -177,7 +204,7 @@ impl GlobalConfigService {
     }
 
     /// Handles messages from external services.
-    fn handle_message(&self, message: GlobalConfigManager) {
+    fn handle_message(&mut self, message: GlobalConfigManager) {
         match message {
             GlobalConfigManager::Get(sender) => {
                 sender.send(self.global_config_watch.borrow().clone());
@@ -231,9 +258,14 @@ impl GlobalConfigService {
                 let mut success = false;
                 match config.global {
                     Some(global_config) => {
+                        // Log the first time we receive a global config from upstream.
+                        if !self.global_config_watch.borrow().is_ready() {
+                            relay_log::info!("received global config from upstream");
+                        }
                         // Notifying subscribers only fails when there are no
                         // subscribers.
-                        self.global_config_watch.send(Arc::new(global_config)).ok();
+                        self.global_config_watch
+                            .send_replace(Status::Ready(Arc::new(global_config)));
                         success = true;
                         self.last_fetched = Instant::now();
                     }
@@ -277,24 +309,29 @@ impl Service for GlobalConfigService {
 
             relay_log::info!("global config service starting");
             if self.config.relay_mode() == RelayMode::Managed {
-                relay_log::info!("serving global configs fetched from upstream");
+                relay_log::info!("requesting global config from upstream");
                 self.request_global_config();
             } else {
                 match GlobalConfig::load(self.config.path()) {
                     Ok(Some(from_file)) => {
                         relay_log::info!("serving static global config loaded from file");
-                        self.global_config_watch.send(Arc::new(from_file)).ok();
+                        self.global_config_watch
+                            .send_replace(Status::Ready(Arc::new(from_file)));
                     }
                     Ok(None) => {
                         relay_log::info!(
                                 "serving default global configs due to lacking static global config file"
                             );
+                        self.global_config_watch
+                            .send_replace(Status::Ready(Arc::default()));
                     }
                     Err(e) => {
                         relay_log::error!("failed to load global config from file: {}", e);
                         relay_log::info!(
                                 "serving default global configs due to failure to load global config from file"
                             );
+                        self.global_config_watch
+                            .send_replace(Status::Ready(Arc::default()));
                     }
                 }
             };

relay-server/src/actors/processor.rs

@@ -66,7 +66,6 @@ use {
 };
 
 use crate::actors::envelopes::{EnvelopeManager, SendEnvelope, SendEnvelopeError, SubmitEnvelope};
-use crate::actors::global_config::{GlobalConfigManager, Subscribe};
 use crate::actors::outcome::{DiscardReason, Outcome, TrackOutcome};
 use crate::actors::project::ProjectState;
 use crate::actors::project_cache::ProjectCache;
@@ -319,6 +318,9 @@ struct ProcessEnvelopeState<'a> {
 
     /// Reservoir evaluator that we use for dynamic sampling.
     reservoir: ReservoirEvaluator<'a>,
+
+    /// Global config used for envelope processing.
+    global_config: Arc<GlobalConfig>,
 }
 
 impl<'a> ProcessEnvelopeState<'a> {
@@ -438,6 +440,7 @@ pub struct ProcessEnvelope {
     pub project_state: Arc<ProjectState>,
     pub sampling_project_state: Option<Arc<ProjectState>>,
     pub reservoir_counters: ReservoirCounters,
+    pub global_config: Arc<GlobalConfig>,
 }
 
 /// Parses a list of metrics or metric buckets and pushes them to the project's aggregator.
@@ -539,7 +542,6 @@ impl FromMessage<RateLimitBuckets> for EnvelopeProcessor {
 /// This service handles messages in a worker pool with configurable concurrency.
 #[derive(Clone)]
 pub struct EnvelopeProcessorService {
-    global_config: Arc<GlobalConfig>,
     inner: Arc<InnerProcessor>,
 }
 
@@ -549,7 +551,6 @@ struct InnerProcessor {
     redis_pool: Option<RedisPool>,
     envelope_manager: Addr<EnvelopeManager>,
     project_cache: Addr<ProjectCache>,
-    global_config: Addr<GlobalConfigManager>,
     outcome_aggregator: Addr<TrackOutcome>,
     #[cfg(feature = "processing")]
     aggregator: Addr<Aggregator>,
@@ -568,7 +569,6 @@ impl EnvelopeProcessorService {
         envelope_manager: Addr<EnvelopeManager>,
         outcome_aggregator: Addr<TrackOutcome>,
         project_cache: Addr<ProjectCache>,
-        global_config: Addr<GlobalConfigManager>,
         upstream_relay: Addr<UpstreamRelay>,
         #[cfg(feature = "processing")] aggregator: Addr<Aggregator>,
     ) -> Self {
@@ -591,7 +591,6 @@ impl EnvelopeProcessorService {
             config,
             envelope_manager,
             project_cache,
-            global_config,
             outcome_aggregator,
             upstream_relay,
             geoip_lookup,
@@ -600,7 +599,6 @@ impl EnvelopeProcessorService {
         };
 
         Self {
-            global_config: Arc::default(),
             inner: Arc::new(inner),
         }
     }
@@ -1341,6 +1339,7 @@ impl EnvelopeProcessorService {
             project_state,
             sampling_project_state,
             reservoir_counters,
+            global_config,
         } = message;
 
         let envelope = managed_envelope.envelope_mut();
@@ -1395,6 +1394,7 @@ impl EnvelopeProcessorService {
             managed_envelope,
             has_profile: false,
             reservoir,
+            global_config,
         })
     }
 
@@ -2706,7 +2706,7 @@ impl EnvelopeProcessorService {
                 enable_trimming: true,
                 measurements: Some(DynamicMeasurementsConfig::new(
                     state.project_state.config().measurements.as_ref(),
-                    self.global_config.measurements.as_ref(),
+                    state.global_config.measurements.as_ref(),
                 )),
             };
 
@@ -3062,25 +3062,13 @@ impl EnvelopeProcessorService {
 impl Service for EnvelopeProcessorService {
     type Interface = EnvelopeProcessor;
 
-    fn spawn_handler(mut self, mut rx: relay_system::Receiver<Self::Interface>) {
+    fn spawn_handler(self, mut rx: relay_system::Receiver<Self::Interface>) {
         let thread_count = self.inner.config.cpu_concurrency();
         relay_log::info!("starting {thread_count} envelope processing workers");
 
         tokio::spawn(async move {
             let semaphore = Arc::new(Semaphore::new(thread_count));
 
-            let Ok(mut subscription) = self.inner.global_config.send(Subscribe).await else {
-                // TODO(iker): we accept this sub-optimal error handling. TBD
-                // the approach to deal with failures on the subscription
-                // mechanism.
-                relay_log::error!("failed to subscribe to GlobalConfigService");
-                return;
-            };
-
-            // In case we use static global config, the watch wont be updated repeatedly, so we
-            // should immediatly use the content of the watch.
-            self.global_config = subscription.borrow().clone();
-
             loop {
                 let next_msg = async {
                     let permit_result = semaphore.clone().acquire_owned().await;
@@ -3093,7 +3081,6 @@ impl Service for EnvelopeProcessorService {
                 tokio::select! {
                    biased;
 
-                    Ok(()) = subscription.changed() => self.global_config = subscription.borrow().clone(),
                     (Some(message), Ok(permit)) = next_msg => {
                         let service = self.clone();
                         tokio::task::spawn_blocking(move || {
@@ -3361,6 +3348,7 @@ mod tests {
                 has_profile: false,
                 event_metrics_extracted: false,
                 reservoir: dummy_reservoir(),
+                global_config: Arc::default(),
             }
         };
 
@@ -3543,7 +3531,6 @@ mod tests {
         let (outcome_aggregator, _) = mock_service("outcome_aggregator", (), |&mut (), _| {});
         let (project_cache, _) = mock_service("project_cache", (), |&mut (), _| {});
         let (upstream_relay, _) = mock_service("upstream_relay", (), |&mut (), _| {});
-        let (global_config, _) = mock_service("global_config", (), |&mut (), _| {});
         #[cfg(feature = "processing")]
         let (aggregator, _) = mock_service("aggregator", (), |&mut (), _| {});
         let inner = InnerProcessor {
@@ -3557,13 +3544,11 @@ mod tests {
             #[cfg(feature = "processing")]
             redis_pool: None,
             geoip_lookup: None,
-            global_config,
             #[cfg(feature = "processing")]
             aggregator,
         };
 
         EnvelopeProcessorService {
-            global_config: Arc::default(),
             inner: Arc::new(inner),
         }
     }
@@ -3598,6 +3583,7 @@ mod tests {
             project_state: Arc::new(ProjectState::allowed()),
             sampling_project_state: None,
             reservoir_counters: ReservoirCounters::default(),
+            global_config: Arc::default(),
         };
 
         let envelope_response = processor.process(message).unwrap();
@@ -3620,6 +3606,7 @@ mod tests {
             project_state: Arc::new(ProjectState::allowed()),
             sampling_project_state,
             reservoir_counters: ReservoirCounters::default(),
+            global_config: Arc::default(),
         };
 
         let envelope_response = processor.process(message).unwrap();
@@ -3831,6 +3818,7 @@ mod tests {
             project_state: Arc::new(project_state),
             sampling_project_state: None,
             reservoir_counters: ReservoirCounters::default(),
+            global_config: Arc::default(),
         };
 
         let envelope_response = processor.process(message).unwrap();
@@ -3902,6 +3890,7 @@ mod tests {
             project_state: Arc::new(ProjectState::allowed()),
             sampling_project_state: None,
             reservoir_counters: ReservoirCounters::default(),
+            global_config: Arc::default(),
         };
 
         let envelope_response = processor.process(message).unwrap();
@@ -3951,6 +3940,7 @@ mod tests {
             project_state: Arc::new(ProjectState::allowed()),
             sampling_project_state: None,
             reservoir_counters: ReservoirCounters::default(),
+            global_config: Arc::default(),
         };
 
         let envelope_response = processor.process(message).unwrap();
@@ -4008,6 +3998,7 @@ mod tests {
             project_state: Arc::new(ProjectState::allowed()),
             sampling_project_state: None,
             reservoir_counters: ReservoirCounters::default(),
+            global_config: Arc::default(),
         };
 
         let envelope_response = processor.process(message).unwrap();