feat(config): add leader election to configuration, and block mDS communication on being the leader (#1176)

Author: XAMPPRocky

Cargo.lock

@@ -101,6 +101,7 @@ hyper = { version = "1.5", features = ["http2", "http1", "server"] }
 hyper-rustls = { version = "0.27", default-features = false, features = [
     "http2",
     "webpki-roots",
+    "aws-lc-rs",
 ] }
 ipnetwork = "0.20.0" # keep in sync with maxmind
 k8s-openapi.workspace = true
@@ -151,6 +152,8 @@ libflate = "2.1.0"
 form_urlencoded = "1.2.1"
 enum_dispatch = "0.3.13"
 gxhash = "3.4.1"
+kube-leader-election = "0.40.0"
+rustls = "0.23.25"
 
 [dependencies.hyper-util]
 version = "0.1"
@@ -204,12 +207,12 @@ cached = { version = "0.54", default-features = false }
 eyre = "0.6.12"
 enum-map = "2.7.3"
 futures = "0.3.31"
-kube = { version = "0.98", features = [
+kube = { version = "0.99", features = [
     "runtime",
     "rustls-tls",
     "client",
 ], default-features = false }
-kube-core = { version = "0.98", default-features = false, features = [
+kube-core = { version = "0.99", default-features = false, features = [
     "schema",
 ] }
 k8s-openapi = { version = "0.24", features = ["v1_29", "schemars"] }

Cargo.toml

@@ -35,3 +35,6 @@ accepted = ["ISC", "OpenSSL"]
 
 [xxhash-rust]
 accepted = ["BSL-1.0"]
+
+[aws-lc-sys]
+accepted = ["OpenSSL"]

about.toml

@@ -33,6 +33,7 @@ futures.workspace = true
 k8s-openapi.workspace = true
 kube = { workspace = true, features = ["openssl-tls", "client", "derive", "runtime"] }
 quilkin = { path = "../../" }
+rustls = "0.23.25"
 serial_test = "3.2.0"
 tokio.workspace = true
 tracing.workspace = true

crates/agones/Cargo.toml

@@ -91,6 +91,7 @@ impl Client {
     /// * Removes previous test namespaces
     /// * Retrieves the `IMAGE_TAG` to test from env vars, and panics if it if not available.
     pub async fn new() -> Client {
+        let _provider = rustls::crypto::aws_lc_rs::default_provider().install_default();
         let mut client = CLIENT
             .get_or_init(|| async {
                 let client = kube::Client::try_default()

crates/agones/src/lib.rs

@@ -40,3 +40,5 @@ tracing.workspace = true
 tryhard.workspace = true
 uuid.workspace = true
 url.workspace = true
+http = "1.3.1"
+tower = { workspace = true, features = ["tokio", "tracing"] }

crates/xds/Cargo.toml

@@ -225,8 +225,13 @@ impl MdsClient {
         config: Arc<C>,
         is_healthy: Arc<AtomicBool>,
     ) -> Result<DeltaSubscription, Self> {
+        const LEADERSHIP_CHECK_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
         let identifier = String::from(&*self.identifier);
 
+        while config.is_leader() == Some(false) {
+            tokio::time::sleep(LEADERSHIP_CHECK_INTERVAL).await;
+        }
+
         let (mut ds, mut stream) =
             match DeltaServerStream::connect(self.client.clone(), identifier.clone()).await {
                 Ok(ds) => {
@@ -245,6 +250,12 @@ impl MdsClient {
                 tracing::trace!("starting relay client delta stream task");
 
                 loop {
+                    if config.is_leader() == Some(false) {
+                        tracing::debug!("not leader, delaying task");
+                        tokio::time::sleep(LEADERSHIP_CHECK_INTERVAL).await;
+                        continue;
+                    }
+
                     {
                         let control_plane = super::server::ControlPlane::from_arc(
                             config.clone(),
@@ -260,7 +271,16 @@ impl MdsClient {
                         let mut stream = control_plane.delta_aggregated_resources(stream).await?;
                         is_healthy.store(true, Ordering::SeqCst);
 
-                        while let Some(result) = stream.next().await {
+                        loop {
+                            if config.is_leader() == Some(false) {
+                                tracing::warn!("lost leader lock mid-stream, disconnecting");
+                                break;
+                            }
+
+                            let Some(result) = stream.next().await else {
+                                break;
+                            };
+
                             let response = result?;
                             tracing::trace!("received delta discovery response");
                             ds.send_response(response).await?;

crates/xds/src/client.rs

@@ -198,6 +198,10 @@ impl ClientTracker {
 pub trait Configuration: Send + Sync + Sized + 'static {
     fn identifier(&self) -> String;
 
+    /// Returns whether the current instance is considered the leader of a set
+    /// of replicas, if leader election is enabled in a config provider.
+    fn is_leader(&self) -> Option<bool>;
+
     fn apply_delta(
         &self,
         resource_type: &str,

crates/xds/src/config.rs

@@ -31,11 +31,12 @@ ignore = [
 deny = [
     { crate = "openssl-sys", use-instead = "rustls" },
     { crate = "openssl", use-instead = "rustls" },
-    { crate = "cmake", use-instead = "cc" },
+    { crate = "cmake", use-instead = "cc", wrappers = ["aws-lc-sys"] },
     { crate = "chrono", use-instead = "time", wrappers = [
         "k8s-openapi",
         "kube-client",
         "kube-core",
+        "kube-leader-election",
     ] },
 ]
 multiple-versions = "deny"
@@ -46,9 +47,6 @@ skip = [
 ]
 skip-tree = [
     { crate = "[email protected]", reason = "matchers is using an old version, https://github.com/hawkw/matchers/pull/5, but it's also barely maintained..." },
-    # Much like trust-dns this pulls in a ton of outdated dependencies, but it's _slightly_ better
-    #{ crate = "[email protected]", reason = "Uses _many_ outdated crates" },
-    { crate = "[email protected]", reason = "several crates use this old version" },
     { crate = "[email protected]", reason = "many crates use this old version" },
     { crate = "[email protected]", reason = "several crates use this old version" },
 ]
@@ -67,6 +65,7 @@ exceptions = [
     # This license should not really be used for code, but here we are
     { crate = "notify", allow = ["CC0-1.0"] },
     { crate = "webpki-roots", allow = ["MPL-2.0"] },
+    { crate = "aws-lc-sys", allow = ["OpenSSL"] },
 ]
 
 [[licenses.clarify]]

deny.toml

@@ -20,12 +20,13 @@ use std::{
     net::{IpAddr, SocketAddr},
     sync::{
         Arc,
-        atomic::{AtomicU64, Ordering::Relaxed},
+        atomic::{AtomicBool, AtomicU64, Ordering::Relaxed},
     },
     time::Duration,
 };
 
 use base64_serde::base64_serde_type;
+use once_cell::sync::Lazy;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
@@ -53,6 +54,21 @@ pub(crate) const BACKOFF_INITIAL_DELAY: Duration = Duration::from_millis(500);
 
 pub type ConfigMap = typemap_rev::TypeMap<dyn typemap_rev::CloneDebuggableStorage>;
 
+#[derive(Debug, Clone, Default)]
+#[repr(transparent)]
+pub(crate) struct LeaderLock(Arc<Lazy<Arc<AtomicBool>>>);
+
+impl LeaderLock {
+    pub(crate) fn load(&self) -> bool {
+        self.0.load(Relaxed)
+    }
+
+    pub(crate) fn store(&self, is_leader: bool) {
+        crate::metrics::leader_election(is_leader);
+        self.0.store(is_leader, Relaxed);
+    }
+}
+
 base64_serde_type!(pub Base64Standard, base64::engine::general_purpose::STANDARD);
 #[derive(Clone)]
 #[cfg_attr(test, derive(Debug))]
@@ -172,6 +188,8 @@ impl<'de> Deserialize<'de> for Config {
                     });
                 };
 
+                typemap.insert::<LeaderLock>(<_>::default());
+
                 Ok(Config {
                     dyn_cfg: DynamicConfig {
                         version: version.unwrap_or_default(),
@@ -309,6 +327,7 @@ impl<'de> Deserialize<'de> for DynamicConfig {
                     });
                 };
 
+                typemap.insert::<LeaderLock>(<_>::default());
                 Ok(DynamicConfig {
                     version: version.unwrap_or_default(),
                     id: id.map_or_else(default_id, |id| Slot::new(Some(id))),
@@ -337,6 +356,10 @@ impl typemap_rev::TypeMapKey for Agent {
     type Value = Agent;
 }
 
+impl typemap_rev::TypeMapKey for LeaderLock {
+    type Value = LeaderLock;
+}
+
 impl DynamicConfig {
     pub fn filters(&self) -> Option<&Slot<FilterChain>> {
         self.typemap.get::<FilterChain>()
@@ -353,6 +376,16 @@ impl DynamicConfig {
     pub fn agent(&self) -> Option<&Agent> {
         self.typemap.get::<Agent>()
     }
+
+    pub(crate) fn init_leader_lock(&self) -> LeaderLock {
+        self.typemap.get::<LeaderLock>().unwrap().clone()
+    }
+
+    pub(crate) fn leader_lock(&self) -> Option<&LeaderLock> {
+        self.typemap
+            .get::<LeaderLock>()
+            .filter(|ll| Lazy::get(&*ll.0).is_some())
+    }
 }
 
 #[cfg(test)]
@@ -383,6 +416,10 @@ impl quilkin_xds::config::Configuration for Config {
         String::clone(&self.id())
     }
 
+    fn is_leader(&self) -> Option<bool> {
+        self.dyn_cfg.leader_lock().map(|ll| ll.load())
+    }
+
     fn allow_request_processing(&self, resource_type: &str) -> bool {
         resource_type.parse::<ResourceType>().is_ok()
     }