Make Config::clusters dynamic (#1125)

googleforgames · Feb 20, 2025 · d7a3ed8 · d7a3ed8
1 parent 96e3bd0
commit d7a3ed8
Show file tree

Hide file tree

Showing 24 changed files with 324 additions and 184 deletions.
diff --git a/benches/shared.rs b/benches/shared.rs
@@ -325,7 +325,7 @@ impl QuilkinLoop {
         let thread = spawn("quilkin", move || {
             let runtime = tokio::runtime::Runtime::new().unwrap();
             let config = Arc::new(quilkin::Config::default_non_agent());
-            config.clusters.modify(|clusters| {
+            config.dyn_cfg.clusters().unwrap().modify(|clusters| {
                 clusters
                     .insert_default([quilkin::net::endpoint::Endpoint::new(endpoint.into())].into())
             });

diff --git a/crates/test/src/lib.rs b/crates/test/src/lib.rs
@@ -490,7 +490,9 @@ impl Pail {
 
                 if !endpoints.is_empty() {
                     config
-                        .clusters
+                        .dyn_cfg
+                        .clusters()
+                        .unwrap()
                         .modify(|clusters| clusters.insert_default(endpoints));
                 }
 

diff --git a/crates/test/tests/proxy.rs b/crates/test/tests/proxy.rs
@@ -88,7 +88,9 @@ trace_test!(uring_receiver, {
 
     let config = std::sync::Arc::new(quilkin::Config::default_non_agent());
     config
-        .clusters
+        .dyn_cfg
+        .clusters()
+        .unwrap()
         .modify(|clusters| clusters.insert_default([endpoint.into()].into()));
 
     let socket = sb.client();
@@ -133,7 +135,9 @@ trace_test!(
 
         let config = std::sync::Arc::new(quilkin::Config::default_non_agent());
         config
-            .clusters
+            .dyn_cfg
+            .clusters()
+            .unwrap()
             .modify(|clusters| clusters.insert_default([endpoint.into()].into()));
 
         let pending_sends: Vec<_> = [

diff --git a/src/cli/service.rs b/src/cli/service.rs
@@ -499,9 +499,13 @@ impl Service {
         let filters = config
             .dyn_cfg
             .filters()
-            .clone()
-            .context("XDP requires a filter chain")?;
-        let clusters = config.clusters.clone();
+            .context("XDP requires a filter chain")?
+            .clone();
+        let clusters = config
+            .dyn_cfg
+            .clusters()
+            .context("XDP requires a cluster map")?
+            .clone();
 
         let config = crate::net::xdp::process::ConfigState { filters, clusters };
 

diff --git a/src/components/manage.rs b/src/components/manage.rs
@@ -39,10 +39,12 @@ impl Manage {
             mut shutdown_rx,
         }: RunArgs<Ready>,
     ) -> crate::Result<()> {
+        let Some(clusters) = config.dyn_cfg.clusters() else {
+            eyre::bail!("clusters were not configured, this is a configuration issue");
+        };
+
         if let Some(locality) = &self.locality {
-            config
-                .clusters
-                .modify(|map| map.update_unlocated_endpoints(locality.clone()));
+            clusters.modify(|map| map.update_unlocated_endpoints(locality.clone()));
         }
 
         let provider_task = match self.provider {

diff --git a/src/components/proxy.rs b/src/components/proxy.rs
@@ -121,6 +121,10 @@ impl Proxy {
             })
         });
 
+        let Some(clusters) = config.dyn_cfg.clusters() else {
+            eyre::bail!("empty clusters were not created")
+        };
+
         if !self.to.is_empty() {
             let endpoints = if let Some(tt) = &self.to_tokens {
                 let (unique, overflow) = 256u64.overflowing_pow(tt.length as _);
@@ -192,12 +196,12 @@ impl Proxy {
                     .collect()
             };
 
-            config.clusters.modify(|clusters| {
+            clusters.modify(|clusters| {
                 clusters.insert(None, endpoints);
             });
         }
 
-        if !config.clusters.read().has_endpoints() && self.management_servers.is_empty() {
+        if !clusters.read().has_endpoints() && self.management_servers.is_empty() {
             return Err(eyre::eyre!(
                  "`quilkin proxy` requires at least one `to` address or `management_server` endpoint."
              ));

diff --git a/src/components/proxy/packet_router.rs b/src/components/proxy/packet_router.rs
@@ -100,12 +100,16 @@ impl<P: PacketMut> DownstreamPacket<P> {
         sessions: &S,
         destinations: &mut Vec<crate::net::EndpointAddress>,
     ) -> Result<(), PipelineError> {
-        if !config.clusters.read().has_endpoints() {
+        let Some(clusters) = config
+            .dyn_cfg
+            .clusters()
+            .filter(|c| c.read().has_endpoints())
+        else {
             tracing::trace!("no upstream endpoints");
             return Err(PipelineError::NoUpstreamEndpoints);
-        }
+        };
 
-        let cm = config.clusters.clone_value();
+        let cm = clusters.clone_value();
         let Some(filters) = config.dyn_cfg.filters() else {
             return Err(PipelineError::Filter(crate::filters::FilterError::Custom(
                 "no filters loaded",