feat: Clean up Datacenters which fail to respond to qcmp pings (#1267)

A proxy that is connected to a relay may miss an update where a
Datacenter is removed if there is a temporary disruption. This adds a
channel where the phoenix subsystem can report "bad nodes" that it has
failed to ping for a period, so that they can be cleaned up from the
Config.

---------

Co-authored-by: XAMPPRocky <[email protected]>

Author: glindstedt

crates/test/src/lib.rs

@@ -368,6 +368,7 @@ impl Pail {
                 let config_path = path.clone();
 
                 let (shutdown, shutdown_rx) = quilkin::signal::channel();
+                let pail_token = quilkin::signal::cancellation_token(shutdown_rx.clone());
 
                 let providers = quilkin::Providers::default().fs().fs_path(path);
                 let svc = quilkin::Service::default()
@@ -376,12 +377,13 @@ impl Pail {
                     .mds()
                     .mds_port(mds_port);
 
-                let config = Arc::new(crate::Config::new(
+                let config = crate::Config::new_rc(
                     Some("test-relay".into()),
                     Default::default(),
                     &providers,
                     &svc,
-                ));
+                    pail_token,
+                );
 
                 *config.dyn_cfg.id.lock() = spc.name.into();
                 let healthy = Arc::new(std::sync::atomic::AtomicBool::new(false));
@@ -452,6 +454,7 @@ impl Pail {
                     .collect::<Vec<_>>();
 
                 let (shutdown, shutdown_rx) = quilkin::signal::channel();
+                let pail_token = quilkin::signal::cancellation_token(shutdown_rx.clone());
 
                 let port = quilkin::net::socket_port(
                     &quilkin::net::raw_socket_with_reuse(0).expect("failed to bind qcmp socket"),
@@ -464,12 +467,13 @@ impl Pail {
                     .fs_path(path)
                     .grpc_push_endpoints(relay_servers);
 
-                let config = Arc::new(crate::Config::new(
+                let config = crate::Config::new_rc(
                     Some("test-agent".into()),
                     apc.icao_code,
                     &providers,
                     &svc,
-                ));
+                    pail_token,
+                );
 
                 *config.dyn_cfg.id.lock() = spc.name.into();
                 let acfg = config.clone();
@@ -539,12 +543,16 @@ impl Pail {
                     .phoenix_port(phoenix_port)
                     .termination_timeout(None);
 
-                let config = Arc::new(crate::Config::new(
+                let (shutdown, shutdown_rx) = quilkin::signal::channel();
+                let pail_token = quilkin::signal::cancellation_token(shutdown_rx.clone());
+
+                let config = crate::Config::new_rc(
                     Some("test-proxy".into()),
                     Default::default(),
                     &Default::default(),
                     &svc,
-                ));
+                    pail_token,
+                );
 
                 if let Some(cfg) = ppc.config {
                     if !cfg.clusters.is_empty() {
@@ -581,7 +589,6 @@ impl Pail {
                 *config.dyn_cfg.id.lock() = spc.name.into();
 
                 let (rttx, rtrx) = tokio::sync::mpsc::unbounded_channel();
-                let (shutdown, shutdown_rx) = quilkin::signal::channel();
 
                 let healthy = Arc::new(std::sync::atomic::AtomicBool::new(false));
                 let provider_task = providers.spawn_providers(

src/cli.rs

@@ -257,18 +257,19 @@ impl Cli {
         tracing::debug!(cli = ?self, "config parameters");
 
         let locality = self.locality.locality();
+        let shutdown_handler = crate::signal::spawn_handler();
+        let drive_token = crate::signal::cancellation_token(shutdown_handler.shutdown_rx());
 
-        let mut config = crate::Config::new(
+        let config = crate::Config::new_rc(
             self.service.id.clone(),
             self.locality.icao_code,
             &self.providers,
             &self.service,
+            drive_token.child_token(),
         );
         config.read_config(&self.config, locality.clone())?;
-        let config = Arc::new(config);
 
         let ready = Arc::<std::sync::atomic::AtomicBool>::default();
-        let shutdown_handler = crate::signal::spawn_handler();
         if self.admin.enabled {
             crate::components::admin::server(
                 config.clone(),

src/config.rs

@@ -74,11 +74,15 @@ impl LeaderLock {
     }
 }
 
+pub type BadNodeInformer = tokio::sync::mpsc::UnboundedSender<std::net::SocketAddr>;
+
 base64_serde_type!(pub Base64Standard, base64::engine::general_purpose::STANDARD);
 #[derive(Clone)]
 #[cfg_attr(test, derive(Debug))]
 pub struct Config {
     pub dyn_cfg: DynamicConfig,
+    bad_node_informer: Option<BadNodeInformer>,
+    cancellation_token: Option<tokio_util::sync::CancellationToken>,
 }
 
 #[cfg(test)]
@@ -356,7 +360,18 @@ fn resolve_id(id: Option<String>) -> String {
         .unwrap_or_else(uuid)
 }
 
+impl Drop for Config {
+    fn drop(&mut self) {
+        if let Some(token) = &self.cancellation_token {
+            token.cancel();
+        }
+    }
+}
+
 impl Config {
+    /// Creates and initializes a new Config
+    ///
+    /// This constructor is mainly intended for tests
     pub fn new(
         id: Option<String>,
         icao_code: IcaoCode,
@@ -370,15 +385,84 @@ impl Config {
                 icao_code: NotifyingIcaoCode::new(icao_code),
                 typemap: default_typemap(),
             },
+            bad_node_informer: None,
+            cancellation_token: None,
         };
         providers.init_config(&mut config);
         service.init_config(&mut config);
 
         config
     }
 
+    /// Creates and initializes a new Arc<Config>
+    ///
+    /// Spawns a tokio task as a side effect that will be stopped when the cancellation token is
+    /// cancelled. The token _will_ be cancelled when Config is dropped, so make sure to pass in a
+    /// child token if the token is intended to live longer than the Config.
+    pub fn new_rc(
+        id: Option<String>,
+        icao_code: IcaoCode,
+        providers: &crate::Providers,
+        service: &crate::Service,
+        cancellation_token: tokio_util::sync::CancellationToken,
+    ) -> Arc<Self> {
+        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<std::net::SocketAddr>();
+
+        let mut config = Self::new(id, icao_code, providers, service);
+        config.cancellation_token = Some(cancellation_token);
+        config.bad_node_informer = Some(tx);
+
+        let config = Arc::new(config);
+        config
+            .spawn_janitor(rx)
+            .expect("spawn_janitor() from new_rc() should not fail");
+
+        config
+    }
+
+    /// Spawns a janitor task to help out with cleaning stale Datacenter entries from the Config
+    fn spawn_janitor(
+        self: &Arc<Config>,
+        mut rx: tokio::sync::mpsc::UnboundedReceiver<std::net::SocketAddr>,
+    ) -> eyre::Result<()> {
+        let cancellation_token = self
+            .cancellation_token
+            .as_ref()
+            .ok_or_else(|| eyre::eyre!("cancellation token not set"))?
+            .clone();
+
+        // Ensure that we don't keep an owning reference to Arc<Config> so that Drop can occurr
+        let janitor_config_ref = Arc::downgrade(self);
+        tokio::spawn(async move {
+            loop {
+                tokio::select! {
+                    _ = cancellation_token.cancelled() => {
+                        break;
+                    }
+                    node = rx.recv() => {
+                        if let Some(node) = node {
+                            let ip = node.ip();
+                            if let Some(config) = janitor_config_ref.upgrade() {
+                                if let Some(datacenters) = config.dyn_cfg.datacenters() {
+                                    datacenters.modify(|wg| {
+                                        tracing::warn!(%ip, "removing datacenter from local state");
+                                        wg.remove(ip);
+                                    });
+                                }
+                            } else {
+                                break;
+                            }
+                        }
+                    }
+                }
+            }
+            tracing::trace!("Stopping janitor task");
+        });
+        Ok(())
+    }
+
     pub fn read_config(
-        &mut self,
+        self: &Arc<Self>,
         config_path: &std::path::Path,
         locality: Option<crate::net::endpoint::Locality>,
     ) -> Result<(), eyre::Error> {
@@ -409,6 +493,10 @@ impl Config {
         Ok(())
     }
 
+    pub fn bad_node_informer(&self) -> Option<BadNodeInformer> {
+        self.bad_node_informer.clone()
+    }
+
     /// Given a list of subscriptions and the current state of the calling client,
     /// construct a response with the current state of our resources that differ
     /// from those of the client