From f02650876af73c32d6d8bbc5b465593d90a2f92f Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:56:58 +0200 Subject: [PATCH 01/47] fix(docker): remove unnecessary build and runtime packages (upstream c952b74) --- Dockerfile | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/Dockerfile b/Dockerfile index 9ef2dca..e6f3960 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,6 +1,6 @@ FROM rust:1.93.1-bookworm AS builder -RUN apt update && apt-get install -y pkg-config libjemalloc-dev libssl-dev && apt-get clean +RUN apt-get update && apt-get install -y --no-install-recommends pkg-config libssl-dev && rm -rf /var/lib/apt/lists/* WORKDIR /app @@ -12,7 +12,9 @@ RUN cargo build --release FROM debian:bookworm AS base -RUN apt-get update && apt install -y libssl-dev ca-certificates libjemalloc-dev && apt-get clean +RUN apt-get update \ + && apt-get install --yes --no-install-recommends libssl3 ca-certificates \ + && rm -rf /var/lib/apt/lists/* COPY --from=builder /app/target/release/robotlb /usr/local/bin/ ENV PATH=/usr/local/bin:$PATH From aab1633b83a308ed89d02a44f506ac56d58e62e3 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:57:07 +0200 Subject: [PATCH 02/47] fix(controller): add bounded per-service failure backoff (upstream 058bc73) --- src/backoff.rs | 128 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 128 insertions(+) create mode 100644 src/backoff.rs diff --git a/src/backoff.rs b/src/backoff.rs new file mode 100644 index 0000000..a267764 --- /dev/null +++ b/src/backoff.rs @@ -0,0 +1,128 @@ +use std::{ + collections::HashMap, + sync::Mutex, + time::{Duration, Instant}, +}; + +const FIRST_DELAY: Duration = Duration::from_secs(5); +const MAX_DELAY: Duration = Duration::from_secs(300); +/// A service deleted before it got the robotlb finalizer is not reconciled again to be +/// forgotten on success, so an entry that has not failed for this long is dropped. It is +/// well above `MAX_DELAY` plus one rate limit pause; a service held at the rate limit gate +/// for longer starts short again. +const FORGET_AFTER: Duration = Duration::from_secs(3600); + +/// Retry delay of each service whose reconciliation keeps failing. +/// +/// It doubles from `FIRST_DELAY` up to `MAX_DELAY` like the upstream cloud-provider +/// service controller: +/// +#[derive(Debug, Default)] +pub struct ErrorBackoff { + failures: Mutex>, +} + +#[derive(Debug)] +struct Failures { + count: u32, + last: Instant, +} + +impl ErrorBackoff { + /// Record a failed reconciliation of `service` and return how long to wait before the next one. + pub fn on_failure(&self, service: &str, now: Instant) -> Duration { + let mut failures = self + .failures + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + failures.retain(|_, failure| now.saturating_duration_since(failure.last) <= FORGET_AFTER); + let failure = failures.entry(service.to_string()).or_insert(Failures { + count: 0, + last: now, + }); + let delay = FIRST_DELAY + .saturating_mul(2_u32.saturating_pow(failure.count)) + .min(MAX_DELAY); + failure.count = failure.count.saturating_add(1); + failure.last = now; + drop(failures); + delay + } + + /// Start the delay of `service` short again. + pub fn forget(&self, service: &str) { + self.failures + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(service); + } + + #[cfg(test)] + fn tracked(&self) -> usize { + self.failures + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .len() + } +} + +#[cfg(test)] +mod tests { + use super::{ErrorBackoff, FORGET_AFTER}; + use std::time::{Duration, Instant}; + + #[test] + fn repeated_failures_double_the_delay_up_to_the_cap() { + let backoff = ErrorBackoff::default(); + let mut now = Instant::now(); + let mut delays = Vec::new(); + for _ in 0..9 { + let delay = backoff.on_failure("shop/web", now); + delays.push(delay.as_secs()); + now += delay; + } + assert_eq!(delays, vec![5, 10, 20, 40, 80, 160, 300, 300, 300]); + } + + #[test] + fn services_back_off_independently() { + let backoff = ErrorBackoff::default(); + let now = Instant::now(); + backoff.on_failure("shop/web", now); + backoff.on_failure("shop/web", now); + assert_eq!(backoff.on_failure("shop/api", now), Duration::from_secs(5)); + assert_eq!(backoff.on_failure("shop/web", now), Duration::from_secs(20)); + } + + #[test] + fn a_forgotten_service_starts_short_again() { + let backoff = ErrorBackoff::default(); + let now = Instant::now(); + backoff.on_failure("shop/web", now); + backoff.on_failure("shop/web", now); + backoff.forget("shop/web"); + assert_eq!(backoff.on_failure("shop/web", now), Duration::from_secs(5)); + } + + #[test] + fn a_long_failing_service_is_not_forgotten() { + let backoff = ErrorBackoff::default(); + let mut now = Instant::now(); + for _ in 0..20 { + now += backoff.on_failure("shop/web", now); + } + assert_eq!( + backoff.on_failure("shop/web", now), + Duration::from_secs(300) + ); + } + + #[test] + fn services_that_stopped_failing_are_dropped() { + let backoff = ErrorBackoff::default(); + let now = Instant::now(); + backoff.on_failure("shop/deleted", now); + backoff.on_failure("shop/web", now + FORGET_AFTER + Duration::from_secs(1)); + assert_eq!(backoff.tracked(), 1); + } +} From 93aa17c91ae377be4ca7f4870769f95e4c06891e Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:57:33 +0200 Subject: [PATCH 03/47] feat(lb): support optional validated cluster identifier (upstream 5ca9e17) --- src/config.rs | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/config.rs b/src/config.rs index 858f6da..23e1f1e 100644 --- a/src/config.rs +++ b/src/config.rs @@ -13,6 +13,10 @@ pub struct OperatorConfig { #[arg(short = 't', long, env = "ROBOTLB_HCLOUD_TOKEN")] pub hcloud_token: String, + /// Optional cluster prefix for default LB names when sharing a Hetzner project. + #[arg(long, env = "ROBOTLB_CLUSTER_NAME", value_parser = crate::lb::parse_cluster_name)] + pub cluster_name: Option, + /// Default network to use for load balancers. /// If not set, then only network from the service annotation will be used. #[arg(long, env = "ROBOTLB_DEFAULT_NETWORK", default_value = None)] @@ -140,6 +144,7 @@ mod tests { .expect("config should parse"); assert_eq!(config.hcloud_token, "token"); + assert_eq!(config.cluster_name, None); assert_eq!(config.default_network, None); assert!(config.dynamic_node_selector); assert_eq!(config.default_lb_retries, 3); @@ -167,6 +172,8 @@ mod tests { "robotlb", "--hcloud-token", "token", + "--cluster-name", + "prod", "--default-network", "prod-net", "--default-lb-retries", @@ -204,6 +211,7 @@ mod tests { .expect("config should parse"); assert_eq!(config.default_network.as_deref(), Some("prod-net")); + assert_eq!(config.cluster_name.as_deref(), Some("prod")); assert!(config.dynamic_node_selector); assert_eq!(config.default_lb_retries, 5); assert_eq!(config.default_lb_timeout, 7); From f28ee58ddca7a6377fea86f3b6e2a2e024770f72 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:58:01 +0200 Subject: [PATCH 04/47] feat(lb): generate stable cluster-prefixed names with hashed length cap (upstream 5ca9e17) --- src/lb/config.rs | 84 +++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 80 insertions(+), 4 deletions(-) diff --git a/src/lb/config.rs b/src/lb/config.rs index 289785b..9bab515 100644 --- a/src/lb/config.rs +++ b/src/lb/config.rs @@ -66,10 +66,10 @@ pub fn parse_load_balancer_config( .get(consts::LB_NAME_ANNOTATION) .cloned() .unwrap_or_else(|| { - format!( - "{}.{}", - svc.name_any(), - svc.namespace().unwrap_or_else(|| "default".to_string()) + default_name( + config.cluster_name.as_deref(), + &svc.name_any(), + &svc.namespace().unwrap_or_else(|| "default".to_string()), ) }); @@ -115,6 +115,43 @@ pub fn parse_load_balancer_config( }) } +const MAX_LB_NAME: usize = 128; + +/// A stable 64-bit name suffix. Balancer names are external identities, so the +/// hash algorithm must never change between operator releases. +fn fnv1a64(value: &str) -> u64 { + value.bytes().fold(0xcbf2_9ce4_8422_2325, |hash, byte| { + (hash ^ u64::from(byte)).wrapping_mul(0x0100_0000_01b3) + }) +} + +fn default_name(cluster: Option<&str>, service: &str, namespace: &str) -> String { + let name = cluster.map_or_else( + || format!("{service}.{namespace}"), + |cluster| format!("{cluster}.{service}.{namespace}"), + ); + if name.len() <= MAX_LB_NAME { + return name; + } + // Service, namespace and validated cluster names are ASCII DNS labels. + let suffix = format!("-{:016x}", fnv1a64(&name)); + format!("{}{}", &name[..MAX_LB_NAME - suffix.len()], suffix) +} + +/// Validate the DNS-label cluster prefix at startup instead of failing on an API call. +pub fn parse_cluster_name(value: &str) -> Result { + let ascii_alnum = |ch: char| ch.is_ascii_lowercase() || ch.is_ascii_digit(); + let valid = (1..=63).contains(&value.len()) + && value.chars().all(|ch| ascii_alnum(ch) || ch == '-') + && value.starts_with(ascii_alnum) + && value.ends_with(ascii_alnum); + if valid { + Ok(value.to_string()) + } else { + Err("must be a DNS label: 1 to 63 lowercase letters, digits or '-', starting and ending with a letter or digit".to_string()) + } +} + /// Parse a numeric or boolean annotation from a service. fn parse_annotation(svc: &Service, key: &str) -> RobotLBResult> where @@ -151,6 +188,7 @@ mod tests { fn base_config() -> OperatorConfig { OperatorConfig { hcloud_token: "token".to_string(), + cluster_name: None, default_network: Some("default-net".to_string()), dynamic_node_selector: true, default_lb_retries: 3, @@ -209,6 +247,44 @@ mod tests { assert!(parsed.public_interface); } + #[test] + fn cluster_prefix_only_changes_unannotated_names() { + let mut config = base_config(); + config.cluster_name = Some("prod".to_string()); + let svc = service_with_annotations([]); + let parsed = parse_load_balancer_config(&svc, &config).unwrap(); + assert_eq!(parsed.name, "prod.svc-name.default"); + let annotated = service_with_annotations([(consts::LB_NAME_ANNOTATION, "custom")]); + assert_eq!( + parse_load_balancer_config(&annotated, &config).unwrap().name, + "custom" + ); + } + + #[test] + fn stable_hashed_names_fit_hetzner_limit() { + let part = "a".repeat(63); + let whole = format!("{part}.{part}.{part}"); + let generated = default_name(Some(&part), &part, &part); + assert_eq!(generated.len(), 128); + assert_eq!(generated, format!("{}-{:016x}", &whole[..111], fnv1a64(&whole))); + assert_eq!(fnv1a64("foobar"), 0x8594_4171_f739_67e8); + assert_ne!( + default_name(Some(&part), &part, &"b".repeat(63)), + default_name(Some(&part), &part, &"c".repeat(63)) + ); + } + + #[test] + fn invalid_cluster_names_are_rejected() { + for invalid in ["", "Prod", "prod.eu", "-prod", "prod-", "some_name", "prød", &"a".repeat(64)] { + assert!(parse_cluster_name(invalid).is_err(), "{invalid:?}"); + } + for valid in ["a", "prod", "eu-1", &"a".repeat(63)] { + assert_eq!(parse_cluster_name(valid).as_deref(), Ok(valid)); + } + } + #[test] fn namespace_is_part_of_default_balancer_name() { let config = base_config(); From 23d07a0b596a8ce5dc21f5c99b5d2ae6ed9dfc17 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:58:10 +0200 Subject: [PATCH 05/47] feat(lb): expose cluster name parser for CLI validation --- src/lb/mod.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/lb/mod.rs b/src/lb/mod.rs index 68926d8..032209a 100644 --- a/src/lb/mod.rs +++ b/src/lb/mod.rs @@ -31,6 +31,7 @@ use crate::{ pub(crate) use api::{HcloudLoadBalancerApi, LiveHcloudLoadBalancerApi}; pub(crate) use config::parse_load_balancer_config; +pub use config::parse_cluster_name; use types::{ServiceReconcileAction, TargetReconcileAction, normalize_ip, service_matches_desired}; /// Struct representing a load balancer. From f900b564825e9bcf58af260a0c43c06c16c8c2d8 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:58:45 +0200 Subject: [PATCH 06/47] feat(controller): share upstream backoff and reconciliation memo across services (058bc73, 5f018d3) --- src/main.rs | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/main.rs b/src/main.rs index e0ea12e..4c99d92 100644 --- a/src/main.rs +++ b/src/main.rs @@ -41,6 +41,7 @@ use crate::{ metrics::Metrics, }; +pub mod backoff; pub mod config; pub mod consts; pub mod controller; @@ -49,6 +50,7 @@ pub mod finalizers; pub mod health; pub mod label_filter; pub mod lb; +pub mod memo; pub mod metrics; pub mod otel_tracing; pub mod prometheus_exporter; @@ -72,6 +74,10 @@ pub struct CurrentContext { pub metrics: Arc, /// Hetzner project-wide API throttle shared across all Services. pub rate_limit: Arc, + /// Failure backoff scoped to each Service UID. + pub error_backoff: Arc, + /// Remember successful desired state to skip redundant Hetzner requests. + pub memo: Arc, /// Suppress repeated identical Kubernetes Warning Events. pub(crate) event_throttle: Arc, } @@ -93,6 +99,8 @@ impl CurrentContext { is_leader, metrics, rate_limit: Arc::default(), + error_backoff: Arc::default(), + memo: Arc::default(), event_throttle: Arc::default(), } } From 9240431790214c22b17d858038bec9ab7698d92d Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:59:04 +0200 Subject: [PATCH 07/47] feat(controller): memoize successfully applied target and service state (upstream 5f018d3) --- src/memo.rs | 300 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 300 insertions(+) create mode 100644 src/memo.rs diff --git a/src/memo.rs b/src/memo.rs new file mode 100644 index 0000000..45834f1 --- /dev/null +++ b/src/memo.rs @@ -0,0 +1,300 @@ +use k8s_openapi::api::core::v1::Service; +use kube::{Resource, ResourceExt}; +use std::{ + collections::{BTreeMap, BTreeSet, HashMap}, + hash::{BuildHasher, DefaultHasher, Hash, Hasher}, + sync::{Mutex, PoisonError}, + time::{Duration, Instant}, +}; + +/// What each service looked like after its last successful reconcile, so that a +/// watch event that changed nothing robotlb acts on costs no Hetzner requests. +#[derive(Default)] +pub struct ReconcileMemo { + entries: Mutex>, +} + +impl ReconcileMemo { + /// How long the recorded state stays valid, when the service is still in it. + pub fn unchanged(&self, uid: &str, fingerprint: u64, now: Instant) -> Option { + let (recorded, until) = *self + .entries + .lock() + .unwrap_or_else(PoisonError::into_inner) + .get(uid)?; + if recorded != fingerprint { + return None; + } + Some(until.saturating_duration_since(now)).filter(|left| !left.is_zero()) + } + + /// Keep the state for `valid_for`, the time until the reconcile is due anyway. + pub fn record(&self, uid: &str, fingerprint: u64, now: Instant, valid_for: Duration) { + self.entries + .lock() + .unwrap_or_else(PoisonError::into_inner) + .insert(uid.to_string(), (fingerprint, now + valid_for)); + } + + /// Drop the entry of a service after a failed reconcile, or once robotlb no + /// longer serves it. + pub fn settle(&self, svc: &Service, succeeded: bool) { + let serves = + svc.meta().deletion_timestamp.is_none() + && crate::controller::ensure_service_is_supported(svc).is_ok(); + if !(succeeded && serves) { + if let Some(uid) = svc.uid() { + self.forget(&uid); + } + } + } + + pub fn forget(&self, uid: &str) { + self.entries + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(uid); + } +} + +/// A hash of everything a reconcile acts on: the spec, the robotlb annotations, and +/// the targets and ports computed from them. +pub fn fingerprint( + svc: &Service, + targets: &[String], + services: &HashMap, +) -> u64 { + // `DefaultHasher::new` uses fixed keys, unlike `RandomState`, so the same input + // hashes the same on every call within the process. + let mut hasher = DefaultHasher::new(); + k8s_openapi::serde_json::to_string(&svc.spec) + .unwrap_or_default() + .hash(&mut hasher); + for annotation in svc + .annotations() + .iter() + .filter(|(key, _)| key.starts_with("robotlb/")) + { + annotation.hash(&mut hasher); + } + targets.iter().collect::>().hash(&mut hasher); + services + .iter() + .collect::>() + .hash(&mut hasher); + hasher.finish() +} + +#[cfg(test)] +mod tests { + use super::{fingerprint, ReconcileMemo}; + use crate::consts; + use k8s_openapi::{ + api::core::v1::{ + LoadBalancerIngress, LoadBalancerStatus, Service, ServicePort, ServiceSpec, + ServiceStatus, + }, + apimachinery::pkg::apis::meta::v1::{ObjectMeta, Time}, + }; + use kube::ResourceExt; + use std::{ + collections::HashMap, + time::{Duration, Instant}, + }; + + const RESYNC: Duration = Duration::from_secs(300); + + fn service() -> Service { + Service { + metadata: ObjectMeta { + name: Some("web".to_string()), + uid: Some("uid-1".to_string()), + annotations: Some( + [( + consts::LB_LOCATION_ANNOTATION.to_string(), + "fsn1".to_string(), + )] + .into(), + ), + ..Default::default() + }, + spec: Some(ServiceSpec { + type_: Some("LoadBalancer".to_string()), + ports: Some(vec![ServicePort { + port: 80, + node_port: Some(30080), + ..Default::default() + }]), + ..Default::default() + }), + ..Default::default() + } + } + + fn fingerprint_of(svc: &Service, targets: &[&str], ports: &[(i32, i32)]) -> u64 { + let targets = targets.iter().map(ToString::to_string).collect::>(); + fingerprint( + svc, + &targets, + &ports.iter().copied().collect::>(), + ) + } + + fn print(svc: &Service) -> u64 { + fingerprint_of(svc, &["10.0.0.1"], &[(80, 30080)]) + } + + #[test] + fn an_unchanged_service_is_skipped_until_the_resync() { + let memo = ReconcileMemo::default(); + let start = Instant::now(); + memo.record("uid-1", 7, start, RESYNC); + assert_eq!( + memo.unchanged("uid-1", 7, start + Duration::from_secs(100)), + Some(Duration::from_secs(200)) + ); + assert_eq!(memo.unchanged("uid-1", 7, start + RESYNC), None); + assert_eq!( + memo.unchanged("uid-1", 7, start + RESYNC + Duration::from_secs(1)), + None + ); + } + + #[test] + fn a_shorter_validity_ends_the_skip_sooner() { + let memo = ReconcileMemo::default(); + let start = Instant::now(); + let retry = Duration::from_secs(30); + memo.record("uid-1", 7, start, retry); + assert_eq!( + memo.unchanged("uid-1", 7, start + Duration::from_secs(10)), + Some(Duration::from_secs(20)) + ); + assert_eq!(memo.unchanged("uid-1", 7, start + retry), None); + } + + #[test] + fn a_changed_or_unknown_service_is_reconciled() { + let memo = ReconcileMemo::default(); + let start = Instant::now(); + assert_eq!(memo.unchanged("uid-1", 7, start), None); + memo.record("uid-1", 7, start, RESYNC); + assert_eq!(memo.unchanged("uid-1", 8, start), None); + assert_eq!(memo.unchanged("uid-2", 7, start), None); + } + + #[test] + fn a_forgotten_service_is_reconciled() { + let memo = ReconcileMemo::default(); + let start = Instant::now(); + memo.record("uid-1", 7, start, RESYNC); + memo.forget("uid-1"); + assert_eq!(memo.unchanged("uid-1", 7, start), None); + } + + #[test] + fn a_new_record_restarts_the_resync() { + let memo = ReconcileMemo::default(); + let start = Instant::now(); + memo.record("uid-1", 7, start, RESYNC); + memo.record("uid-1", 8, start + RESYNC, RESYNC); + assert_eq!(memo.unchanged("uid-1", 8, start + RESYNC), Some(RESYNC)); + } + + fn settled(svc: &Service, succeeded: bool) -> bool { + let memo = ReconcileMemo::default(); + let start = Instant::now(); + memo.record("uid-1", 7, start, RESYNC); + memo.settle(svc, succeeded); + memo.unchanged("uid-1", 7, start).is_some() + } + + #[test] + fn a_successful_reconcile_keeps_the_entry() { + assert!(settled(&service(), true)); + } + + #[test] + fn a_failed_reconcile_drops_the_entry() { + assert!(!settled(&service(), false)); + } + + #[test] + fn a_released_or_deleted_service_drops_the_entry() { + let mut svc = service(); + svc.spec.as_mut().unwrap().type_ = Some("ClusterIP".to_string()); + assert!(!settled(&svc, true)); + let mut svc = service(); + svc.spec.as_mut().unwrap().load_balancer_class = Some("other".to_string()); + assert!(!settled(&svc, true)); + let mut svc = service(); + svc.metadata.deletion_timestamp = Some(Time(k8s_openapi::chrono::Utc::now())); + assert!(!settled(&svc, true)); + } + + #[test] + fn the_fingerprint_is_stable() { + assert_eq!(print(&service()), print(&service())); + } + + #[test] + fn a_spec_change_changes_the_fingerprint() { + let mut svc = service(); + svc.spec.as_mut().unwrap().external_traffic_policy = Some("Local".to_string()); + assert_ne!(print(&svc), print(&service())); + } + + #[test] + fn a_robotlb_annotation_changes_the_fingerprint() { + let mut svc = service(); + svc.annotations_mut().insert( + consts::LB_LOCATION_LABEL_NAME.to_string(), + "nbg1".to_string(), + ); + assert_ne!(print(&svc), print(&service())); + let mut svc = service(); + svc.annotations_mut().insert( + consts::LB_PRIVATE_IP_ANNOTATION.to_string(), + "10.0.0.9".to_string(), + ); + assert_ne!(print(&svc), print(&service())); + } + + #[test] + fn the_targets_change_the_fingerprint() { + let svc = service(); + let one = fingerprint_of(&svc, &["10.0.0.1"], &[(80, 30080)]); + let two = fingerprint_of(&svc, &["10.0.0.1", "10.0.0.2"], &[(80, 30080)]); + let reordered = fingerprint_of(&svc, &["10.0.0.2", "10.0.0.1"], &[(80, 30080)]); + assert_ne!(one, two); + assert_eq!(two, reordered); + } + + #[test] + fn the_ports_change_the_fingerprint() { + let svc = service(); + let one = fingerprint_of(&svc, &["10.0.0.1"], &[(80, 30080)]); + let other = fingerprint_of(&svc, &["10.0.0.1"], &[(80, 30081)]); + assert_ne!(one, other); + } + + #[test] + fn status_and_foreign_metadata_leave_the_fingerprint_alone() { + let mut svc = service(); + svc.status = Some(ServiceStatus { + load_balancer: Some(LoadBalancerStatus { + ingress: Some(vec![LoadBalancerIngress { + ip: Some("192.0.2.1".to_string()), + ..Default::default() + }]), + }), + ..Default::default() + }); + svc.metadata.resource_version = Some("42".to_string()); + svc.annotations_mut().insert( + "metallb.io/ip-allocated-from-pool".to_string(), + "x".to_string(), + ); + assert_eq!(print(&svc), print(&service())); + } +} From c70ef3bf72f6c212a08c1ca93c137bba354e0cba Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 11:59:16 +0200 Subject: [PATCH 08/47] feat(controller): configurable bounded periodic resync (upstream dc37dcf) --- src/config.rs | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/src/config.rs b/src/config.rs index 23e1f1e..da20379 100644 --- a/src/config.rs +++ b/src/config.rs @@ -27,6 +27,15 @@ pub struct OperatorConfig { #[arg(long, env = "ROBOTLB_DYNAMIC_NODE_SELECTOR", default_value = "true")] pub dynamic_node_selector: bool, + /// Maximum seconds between full Hetzner drift checks, even with no changes. + #[arg( + long, + env = "ROBOTLB_RESYNC_INTERVAL", + default_value = "300", + value_parser = clap::value_parser!(u64).range(1..=31_536_000) + )] + pub resync_interval: u64, + /// Default load balancer healthcheck retries cound. #[arg(long, env = "ROBOTLB_DEFAULT_LB_RETRIES", default_value = "3")] pub default_lb_retries: i32, @@ -147,6 +156,7 @@ mod tests { assert_eq!(config.cluster_name, None); assert_eq!(config.default_network, None); assert!(config.dynamic_node_selector); + assert_eq!(config.resync_interval, 300); assert_eq!(config.default_lb_retries, 3); assert_eq!(config.default_lb_timeout, 10); assert_eq!(config.default_lb_interval, 15); @@ -174,6 +184,8 @@ mod tests { "token", "--cluster-name", "prod", + "--resync-interval", + "120", "--default-network", "prod-net", "--default-lb-retries", @@ -211,6 +223,7 @@ mod tests { .expect("config should parse"); assert_eq!(config.default_network.as_deref(), Some("prod-net")); + assert_eq!(config.resync_interval, 120); assert_eq!(config.cluster_name.as_deref(), Some("prod")); assert!(config.dynamic_node_selector); assert_eq!(config.default_lb_retries, 5); From be1246ddc016f9f6ed73faee3c6750890893023f Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:00:01 +0200 Subject: [PATCH 09/47] fix(hcloud): preserve network error cause chains and structured Hetzner messages (upstream issue 77) --- src/error.rs | 67 ++++++++++++++++++++++++++++++++++++++++------------ 1 file changed, 52 insertions(+), 15 deletions(-) diff --git a/src/error.rs b/src/error.rs index 66b086e..2940ef4 100644 --- a/src/error.rs +++ b/src/error.rs @@ -30,6 +30,43 @@ pub fn redact(message: &str, token: &str) -> String { safe } +/// Format a Hetzner error with its concrete network/TLS/DNS cause. +/// The caller redacts the configured token before writing it to logs or Events. +fn describe_hcloud(error: &hcloud::apis::Error) -> String { + use std::error::Error as _; + let (mut description, mut cause): (String, Option<&(dyn std::error::Error + 'static)>) = + match error { + hcloud::apis::Error::Reqwest(inner) => (inner.to_string(), inner.source()), + hcloud::apis::Error::Io(inner) => (inner.to_string(), inner.source()), + hcloud::apis::Error::Serde(inner) => (inner.to_string(), inner.source()), + hcloud::apis::Error::ResponseError(response) => { + let detail = k8s_openapi::serde_json::from_str::( + &response.content, + ) + .ok() + .and_then(|value| { + Some(format!( + "{}: {}", + value.pointer("/error/code")?.as_str()?, + value.pointer("/error/message")?.as_str()?, + )) + }); + let message = detail.map_or_else( + || response.status.to_string(), + |text| format!("{}: {text}", response.status), + ); + (message, None) + } + }; + while let Some(source) = cause { + description.push_str(": "); + description.push_str(&source.to_string()); + cause = source.source(); + } + // The same message is also emitted in Kubernetes Events, which are line-oriented. + description.split_whitespace().collect::>().join(" ") +} + /// All possible errors that can occur in the robotlb operator. #[derive(Debug, Error)] pub enum RobotLBError { @@ -47,7 +84,7 @@ pub enum RobotLBError { HCloudError(String), #[error("Invalid load balancer configuration: {0}")] InvalidLoadBalancerConfiguration(String), - #[error("Cannot list Hetzner load balancer types: {0}")] + #[error("Cannot list Hetzner load balancer types: {}", describe_hcloud(.0))] HCloudListLoadBalancerTypesError( #[from] hcloud::apis::Error, @@ -60,62 +97,62 @@ pub enum RobotLBError { ServiceWithoutSelector, // HCloud API errors - #[error("Cannot attach load balancer to a network. Reason: {0}")] + #[error("Cannot attach load balancer to a network. Reason: {}", describe_hcloud(.0))] HCloudLBAttachToNetworkError( #[from] hcloud::apis::Error, ), - #[error("Cannot detach load balancer from network. Reason: {0}")] + #[error("Cannot detach load balancer from network. Reason: {}", describe_hcloud(.0))] HcloudLBDetachFromNetworkError( #[from] hcloud::apis::Error, ), - #[error("Cannot add load balancer target. Reason: {0}")] + #[error("Cannot add load balancer target. Reason: {}", describe_hcloud(.0))] HcloudLBAddTargetError( #[from] hcloud::apis::Error, ), - #[error("Cannot remove load balancer target. Reason: {0}")] + #[error("Cannot remove load balancer target. Reason: {}", describe_hcloud(.0))] HcloudLBRemoveTargetError( #[from] hcloud::apis::Error, ), - #[error("Cannot add service to load balancer. Reason: {0}")] + #[error("Cannot add service to load balancer. Reason: {}", describe_hcloud(.0))] HcloudLBAddServiceError( #[from] hcloud::apis::Error, ), - #[error("Cannot remove service from load balancer. Reason: {0}")] + #[error("Cannot remove service from load balancer. Reason: {}", describe_hcloud(.0))] HcloudLBRemoveServiceError( #[from] hcloud::apis::Error, ), - #[error("Cannot create load balancer. Reason: {0}")] + #[error("Cannot create load balancer. Reason: {}", describe_hcloud(.0))] HcloudLBCreateError( #[from] hcloud::apis::Error, ), - #[error("Cannot delete load balancer. Reason: {0}")] + #[error("Cannot delete load balancer. Reason: {}", describe_hcloud(.0))] HcloudLBDeleteError( #[from] hcloud::apis::Error, ), - #[error("Cannot get load balancer. Reason: {0}")] + #[error("Cannot get load balancer. Reason: {}", describe_hcloud(.0))] HcloudLBGetError( #[from] hcloud::apis::Error, ), - #[error("Cannot update service. Reason: {0}")] + #[error("Cannot update service. Reason: {}", describe_hcloud(.0))] HcloudLBUpdateServiceError( #[from] hcloud::apis::Error, ), - #[error("Cannot change type of load balancer. Reason: {0}")] + #[error("Cannot change type of load balancer. Reason: {}", describe_hcloud(.0))] HcloudLBChangeType( #[from] hcloud::apis::Error, ), - #[error("Cannot change algorithm of load balancer. Reason: {0}")] + #[error("Cannot change algorithm of load balancer. Reason: {}", describe_hcloud(.0))] HcloudLBChangeAlgorithm( #[from] hcloud::apis::Error, ), - #[error("Cannot list networks. Reason: {0}")] + #[error("Cannot list networks. Reason: {}", describe_hcloud(.0))] HcloudListNetworksError( #[from] hcloud::apis::Error, ), - #[error("Cannot list load balancers. Reason: {0}")] + #[error("Cannot list load balancers. Reason: {}", describe_hcloud(.0))] HcloudListLoadBalancersError( #[from] hcloud::apis::Error, ), From c31f887dbff7120dede1f423442ec4d2d1401b30 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:00:43 +0200 Subject: [PATCH 10/47] feat(controller): avoid redundant Hetzner calls and back off failures (upstream 5f018d3, 058bc73) --- src/controller/mod.rs | 43 ++++++++++++++++++++++++++++++++++++++----- 1 file changed, 38 insertions(+), 5 deletions(-) diff --git a/src/controller/mod.rs b/src/controller/mod.rs index 317203c..16d5a65 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -4,6 +4,7 @@ //! with the `robotlb` load balancer class. mod event_throttle; +mod node_changes; mod nodes; mod status; @@ -16,6 +17,7 @@ use std::{ }; use futures::StreamExt; +use crate::memo; use k8s_openapi::api::core::v1::Service; use kube::{ Resource, ResourceExt, @@ -24,7 +26,7 @@ use kube::{ controller::Action, events::{Event, EventType, Recorder, Reporter}, reflector::ObjectRef, - watcher, + watcher, WatchStreamExt, }, }; @@ -41,7 +43,7 @@ pub(crate) use nodes::{ }; pub(crate) use status::{build_ingress, patch_ingress_status}; -const SUCCESS_REQUEUE_SECS: u64 = 180; +const SUCCESS_REQUEUE_SECS: u64 = 300; /// Check if a service is supported by robotlb. /// @@ -97,6 +99,10 @@ pub async fn reconcile_service( ) -> RobotLBResult { let mut timer = ReconcileTimer::new(context.metrics.clone()); let result = reconcile_service_inner(svc.clone(), context.clone()).await; + context.memo.settle(&svc, result.is_ok()); + if matches!(result, Ok(_) | Err(RobotLBError::SkipService)) { + context.error_backoff.forget(&service_key(&svc)); + } if let Err(error) = &result && !matches!(error, RobotLBError::SkipService) { @@ -144,6 +150,12 @@ async fn reconcile_service_inner( reconcile_load_balancer(lb, svc, context).await } +fn service_key(svc: &Service) -> String { + svc.uid().unwrap_or_else(|| { + format!("{}/{}", svc.namespace().unwrap_or_default(), svc.name_any()) + }) +} + /// Display the last error on the Service without leaking Hetzner API credentials. /// Event publishing errors must not hide the original reconciliation failure. async fn publish_reconciliation_warning( @@ -239,6 +251,16 @@ pub async fn reconcile_load_balancer( apply_desired_state(&mut lb, desired_targets, desired_services); + let uid = svc.uid(); + let fingerprint = memo::fingerprint(&svc, &lb.targets, &lb.services); + if let Some(wait) = uid + .as_deref() + .and_then(|uid| context.memo.unchanged(uid, fingerprint, std::time::Instant::now())) + { + tracing::debug!("No effective configuration change; skipping Hetzner API"); + return Ok(Action::requeue(wait)); + } + let hcloud_lb = lb.reconcile().await?; let ingress = build_ingress( @@ -249,12 +271,16 @@ pub async fn reconcile_load_balancer( ); patch_ingress_status(&svc, &context, ingress).await?; - Ok(Action::requeue(Duration::from_secs(SUCCESS_REQUEUE_SECS))) + let resync = Duration::from_secs(context.config.resync_interval); + if let Some(uid) = uid { + context.memo.record(&uid, fingerprint, std::time::Instant::now(), resync); + } + Ok(Action::requeue(resync)) } /// Handle the error during reconciliation. #[allow(clippy::needless_pass_by_value)] -pub fn on_error(_: Arc, error: &RobotLBError, context: Arc) -> Action { +pub fn on_error(svc: Arc, error: &RobotLBError, context: Arc) -> Action { if error.is_rate_limited() { let delay = context.rate_limit.on_rate_limit(std::time::Instant::now()); tracing::warn!( @@ -263,7 +289,14 @@ pub fn on_error(_: Arc, error: &RobotLBError, context: Arc Action { From 0e35869d8e0ca077946bca996229e47f4f584cf1 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:01:08 +0200 Subject: [PATCH 11/47] test(controller): exercise failure requeues through shared backoff and rate-limit gate --- src/controller/mod.rs | 69 ++++++++++++++++++------------------------- 1 file changed, 28 insertions(+), 41 deletions(-) diff --git a/src/controller/mod.rs b/src/controller/mod.rs index 16d5a65..bc1e754 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -43,7 +43,6 @@ pub(crate) use nodes::{ }; pub(crate) use status::{build_ingress, patch_ingress_status}; -const SUCCESS_REQUEUE_SECS: u64 = 300; /// Check if a service is supported by robotlb. /// @@ -278,34 +277,28 @@ pub async fn reconcile_load_balancer( Ok(Action::requeue(resync)) } -/// Handle the error during reconciliation. +/// Schedule failures independently, but pause the entire project on rate limits. #[allow(clippy::needless_pass_by_value)] pub fn on_error(svc: Arc, error: &RobotLBError, context: Arc) -> Action { - if error.is_rate_limited() { - let delay = context.rate_limit.on_rate_limit(std::time::Instant::now()); - tracing::warn!( - ?delay, - "Hetzner API rate limit reached; pausing all Services" - ); - return Action::requeue(delay); - } - if matches!(error, RobotLBError::SkipService) { - return Action::await_change(); - } - Action::requeue( - context - .error_backoff - .on_failure(&service_key(&svc), std::time::Instant::now()), - ) + error_action(error, &context.rate_limit, &context.error_backoff, + std::time::Instant::now(), &service_key(&svc)) } -const fn action_for_error(error: &RobotLBError) -> Action { +fn error_action( + error: &RobotLBError, + rate_limit: &crate::rate_limit::RateLimitGate, + backoff: &crate::backoff::ErrorBackoff, + now: std::time::Instant, + service: &str, +) -> Action { match error { RobotLBError::SkipService => Action::await_change(), - RobotLBError::InvalidLoadBalancerConfiguration(_) => { - Action::requeue(Duration::from_secs(180)) + error if error.is_rate_limited() => { + let delay = rate_limit.on_rate_limit(now); + tracing::warn!(?delay, "Hetzner API rate limit reached; pausing all Services"); + Action::requeue(delay) } - _ => Action::requeue(Duration::from_secs(30)), + _ => Action::requeue(backoff.on_failure(service, now)), } } @@ -436,24 +429,18 @@ mod tests { } #[test] - fn on_error_awaits_change_for_skipped_service() { - let action = action_for_error(&RobotLBError::SkipService); - assert_eq!(action, Action::await_change()); - } - - #[test] - fn invalid_configuration_requeues_less_frequently() { - assert_eq!( - action_for_error(&RobotLBError::InvalidLoadBalancerConfiguration( - "test".to_string() - )), - Action::requeue(Duration::from_secs(180)) - ); - } - - #[test] - fn on_error_requeues_for_non_skip_errors() { - let action = action_for_error(&RobotLBError::ServiceWithoutSelector); - assert_eq!(action, Action::requeue(Duration::from_secs(30))); + fn failure_backoff_is_independent_for_each_service() { + let gate = crate::rate_limit::RateLimitGate::default(); + let backoff = crate::backoff::ErrorBackoff::default(); + let now = std::time::Instant::now(); + let error = RobotLBError::HCloudError("test".to_string()); + assert_eq!(error_action(&error, &gate, &backoff, now, "shop/web"), + Action::requeue(Duration::from_secs(5))); + assert_eq!(error_action(&error, &gate, &backoff, now, "shop/web"), + Action::requeue(Duration::from_secs(10))); + assert_eq!(error_action(&error, &gate, &backoff, now, "shop/api"), + Action::requeue(Duration::from_secs(5))); + assert_eq!(error_action(&RobotLBError::SkipService, &gate, &backoff, now, "shop/web"), + Action::await_change()); } } From 32548c8fc889b98e03ef4762aa2f6d33b7da5873 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:01:36 +0200 Subject: [PATCH 12/47] perf(controller): detect meaningful Node watch events without heartbeat churn (upstream dc37dcf) --- src/controller/node_changes.rs | 134 +++++++++++++++++++++++++++++++++ 1 file changed, 134 insertions(+) create mode 100644 src/controller/node_changes.rs diff --git a/src/controller/node_changes.rs b/src/controller/node_changes.rs new file mode 100644 index 0000000..5183f09 --- /dev/null +++ b/src/controller/node_changes.rs @@ -0,0 +1,134 @@ +//! Detect node changes that actually affect Hetzner target selection. +//! Kubelet heartbeats must not trigger reconciling every LoadBalancer Service. + +use std::{ + collections::{hash_map::DefaultHasher, HashMap}, + hash::{Hash, Hasher}, +}; + +use k8s_openapi::api::core::v1::Node; +use kube::{runtime::watcher::Event, ResourceExt}; + +#[derive(Debug, Default)] +pub struct NodeChanges { + seen: HashMap, + relisted: HashMap, + synced: bool, +} + +impl NodeChanges { + /// True for a node change affecting targets, including relist/deletions. + pub fn observe(&mut self, event: &Event) -> bool { + match event { + Event::Init => { + self.relisted.clear(); + false + } + Event::InitApply(node) => { + self.relisted.insert(node.name_any(), fingerprint(node)); + false + } + Event::InitDone => { + let changed = !self.synced || self.relisted != self.seen; + self.seen = std::mem::take(&mut self.relisted); + self.synced = true; + changed + } + Event::Apply(node) => { + let print = fingerprint(node); + self.seen.insert(node.name_any(), print) != Some(print) + } + Event::Delete(node) => self.seen.remove(&node.name_any()).is_some(), + } + } +} + +fn fingerprint(node: &Node) -> u64 { + let mut hasher = DefaultHasher::new(); + node.labels().hash(&mut hasher); + node.spec + .as_ref() + .and_then(|spec| spec.unschedulable) + .hash(&mut hasher); + if let Some(status) = &node.status { + status + .conditions + .iter() + .flatten() + .find(|condition| condition.type_ == "Ready") + .map(|condition| condition.status.as_str()) + .hash(&mut hasher); + for address in status.addresses.iter().flatten() { + (address.type_.as_str(), address.address.as_str()).hash(&mut hasher); + } + } + hasher.finish() +} + +#[cfg(test)] +mod tests { + use super::*; + use k8s_openapi::{ + api::core::v1::{NodeAddress, NodeCondition, NodeSpec, NodeStatus}, + apimachinery::pkg::apis::meta::v1::ObjectMeta, + }; + + fn node(name: &str, ready: &str) -> Node { + Node { + metadata: ObjectMeta { + name: Some(name.to_string()), + ..Default::default() + }, + spec: Some(NodeSpec::default()), + status: Some(NodeStatus { + addresses: Some(vec![NodeAddress { + type_: "InternalIP".to_string(), + address: "192.0.2.10".to_string(), + }]), + conditions: Some(vec![NodeCondition { + type_: "Ready".to_string(), + status: ready.to_string(), + ..Default::default() + }]), + ..Default::default() + }), + } + } + + #[test] + fn ignores_heartbeats_but_reacts_to_node_target_fields() { + let mut changes = NodeChanges::default(); + assert!(!changes.observe(&Event::Init)); + assert!(!changes.observe(&Event::InitApply(node("a", "True")))); + assert!(changes.observe(&Event::InitDone)); + let mut heartbeat = node("a", "True"); + heartbeat.status.as_mut().unwrap().conditions.as_mut().unwrap()[0].message = + Some("kubelet heartbeat".to_string()); + assert!(!changes.observe(&Event::Apply(heartbeat))); + assert!(changes.observe(&Event::Apply(node("a", "False")))); + assert!(!changes.observe(&Event::Apply(node("a", "False")))); + + let mut cordoned = node("a", "False"); + cordoned.spec.as_mut().unwrap().unschedulable = Some(true); + assert!(changes.observe(&Event::Apply(cordoned.clone()))); + cordoned.status.as_mut().unwrap().addresses.as_mut().unwrap()[0].address = + "192.0.2.11".to_string(); + assert!(changes.observe(&Event::Apply(cordoned))); + assert!(changes.observe(&Event::Delete(node("a", "False")))); + } + + #[test] + fn relist_and_new_node_detection() { + let mut changes = NodeChanges::default(); + changes.observe(&Event::Init); + changes.observe(&Event::InitApply(node("a", "True"))); + assert!(changes.observe(&Event::InitDone)); + changes.observe(&Event::Init); + changes.observe(&Event::InitApply(node("a", "True"))); + assert!(!changes.observe(&Event::InitDone)); + changes.observe(&Event::Init); + changes.observe(&Event::InitApply(node("a", "False"))); + assert!(changes.observe(&Event::InitDone)); + assert!(changes.observe(&Event::Apply(node("b", "True")))); + } +} From 3e5693e95ae34d7737602a6a2fbe331a75f2023e Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:02:02 +0200 Subject: [PATCH 13/47] perf(controller): wake services on relevant Node changes and retain scheduled resync (upstream dc37dcf) --- src/controller/mod.rs | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/src/controller/mod.rs b/src/controller/mod.rs index bc1e754..78ec59f 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -318,6 +318,35 @@ pub fn map_endpoint_slice_to_service( Some(ObjectRef::new(&service_name).within(&namespace)) } +/// Wake all Services only when target-relevant Node state changes. +/// Periodic resync still detects out-of-band Hetzner changes. +fn node_change_signals( + client: kube::Client, +) -> impl futures::Stream + Send + Sync { + let (sender, receiver) = futures::channel::mpsc::unbounded(); + tokio::spawn(async move { + let mut changes = node_changes::NodeChanges::default(); + let mut events = std::pin::pin!( + watcher::watcher( + kube::Api::::all(client), + watcher::Config::default(), + ) + .default_backoff() + ); + while let Some(event) = events.next().await { + match event { + Ok(event) => { + if changes.observe(&event) && sender.unbounded_send(()).is_err() { + return; + } + } + Err(error) => tracing::warn!(%error, "Node watch failed"), + } + } + }); + receiver +} + /// Create and run the controller. pub async fn run( kube_client: kube::Client, @@ -330,6 +359,7 @@ pub async fn run( kube::Api::::all(kube_client), watcher::Config::default(), ) + .reconcile_all_on(node_change_signals(context.client.clone())) .watches( kube::Api::::all(context.client.clone()), watcher::Config::default(), From 87f2bc850f08340c2136d54ef9cb87da843923a1 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:02:36 +0200 Subject: [PATCH 14/47] docs(helm): expose cluster prefix and drift resync settings --- helm/values.yaml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/helm/values.yaml b/helm/values.yaml index 64d5042..ef4be6d 100644 --- a/helm/values.yaml +++ b/helm/values.yaml @@ -16,6 +16,10 @@ replicaCount: 1 envs: ROBOTLB_LOG_LEVEL: "INFO" + # Optional cluster prefix when multiple clusters share one Hetzner project. + # ROBOTLB_CLUSTER_NAME: "prod" + # Full drift resync interval; Node and EndpointSlice changes trigger earlier. + ROBOTLB_RESYNC_INTERVAL: "300" existingSecrets: [] From 0e0a65fea63102601e280e3f70a24f02614b0190 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:02:53 +0200 Subject: [PATCH 15/47] docs: explain memoized reconciliation, targeted watches, retries and cluster-scoped LB naming --- README.md | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/README.md b/README.md index c5dbfd5..905d700 100644 --- a/README.md +++ b/README.md @@ -231,3 +231,34 @@ spec: ## Star History [![Star History Chart](https://api.star-history.com/svg?repos=forkline/robotlb&type=Date)](https://star-history.com/#forkline/robotlb&Date) + +## Reconciliation and multi-cluster operation + +robotlb handles `LoadBalancer` Services whose `spec.loadBalancerClass` is unset +or set to `robotlb`. When other LoadBalancer controllers run in the cluster, +set `spec.loadBalancerClass: robotlb` **when creating the Service** to prevent +competing controllers from repeatedly overwriting `status.loadBalancer`. +The class is effectively immutable after creation; recreating the Service can +replace the LB and change its IP. + +Node changes that affect targets (readiness, labels, cordon state or addresses) +trigger reconciliation immediately; kubelet heartbeats do not. EndpointSlice +updates also reconcile their Service. On otherwise unchanged Services, the +operator compares the desired Service spec, `robotlb/` annotations, target IPs +and listener ports with the last successful reconcile and skips Hetzner requests. +Every `ROBOTLB_RESYNC_INTERVAL` seconds (default **300**) it performs a full +check anyway to recover out-of-band changes made in Hetzner. + +A failing Service retries independently with exponential backoff from 5 seconds +up to 5 minutes, without suppressing reconciliation triggered by Kubernetes +watch events. Project-wide Hetzner API rate limits use the separate shared gate. +Errors remain visible in logs and Kubernetes Warning Events. + +When several clusters share one Hetzner Cloud project, set +`ROBOTLB_CLUSTER_NAME` (for example `prod`) to a DNS-label prefix +of 1–63 lowercase letters, digits and hyphens. Default LB names then become +`..`, with a stable hash suffix when exceeding +Hetzner's 128-character limit. Explicit `robotlb/balancer` names are not +prefixed. Existing owned balancers continue to be selected by immutable Service +UID labels, so changing the prefix does **not** rename or adopt them. +Unlabelled legacy balancers are not automatically adopted. From 988f645bc21e19425e4099fdccf67371315170b9 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:03:54 +0200 Subject: [PATCH 16/47] test(memo): use Forkline annotation keys in upstream fingerprint tests --- src/memo.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/memo.rs b/src/memo.rs index 45834f1..6bb6b07 100644 --- a/src/memo.rs +++ b/src/memo.rs @@ -248,7 +248,7 @@ mod tests { fn a_robotlb_annotation_changes_the_fingerprint() { let mut svc = service(); svc.annotations_mut().insert( - consts::LB_LOCATION_LABEL_NAME.to_string(), + consts::LB_LOCATION_ANNOTATION.to_string(), "nbg1".to_string(), ); assert_ne!(print(&svc), print(&service())); From 63810bd61bb8586dfd0ef4166c52ba3a04a6a090 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:04:39 +0200 Subject: [PATCH 17/47] test(config): set resync interval in controller test fixture --- src/lb/config.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/lb/config.rs b/src/lb/config.rs index 9bab515..4bf6fd2 100644 --- a/src/lb/config.rs +++ b/src/lb/config.rs @@ -191,6 +191,7 @@ mod tests { cluster_name: None, default_network: Some("default-net".to_string()), dynamic_node_selector: true, + resync_interval: 300, default_lb_retries: 3, default_lb_timeout: 10, default_lb_interval: 15, From d105d0877237b0a83ec783eb6c22a2e4a9536256 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:05:26 +0200 Subject: [PATCH 18/47] fix(ci): adapt upstream tests and Clippy documentation to kube 3 --- src/memo.rs | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/src/memo.rs b/src/memo.rs index 6bb6b07..9ac0d39 100644 --- a/src/memo.rs +++ b/src/memo.rs @@ -42,10 +42,10 @@ impl ReconcileMemo { let serves = svc.meta().deletion_timestamp.is_none() && crate::controller::ensure_service_is_supported(svc).is_ok(); - if !(succeeded && serves) { - if let Some(uid) = svc.uid() { - self.forget(&uid); - } + if !(succeeded && serves) + && let Some(uid) = svc.uid() + { + self.forget(&uid); } } @@ -94,7 +94,7 @@ mod tests { LoadBalancerIngress, LoadBalancerStatus, Service, ServicePort, ServiceSpec, ServiceStatus, }, - apimachinery::pkg::apis::meta::v1::{ObjectMeta, Time}, + apimachinery::pkg::apis::meta::v1::ObjectMeta, }; use kube::ResourceExt; use std::{ @@ -228,7 +228,9 @@ mod tests { svc.spec.as_mut().unwrap().load_balancer_class = Some("other".to_string()); assert!(!settled(&svc, true)); let mut svc = service(); - svc.metadata.deletion_timestamp = Some(Time(k8s_openapi::chrono::Utc::now())); + svc.metadata.deletion_timestamp = Some( + k8s_openapi::serde_json::from_str("\\"2026-01-01T00:00:00Z\\"").unwrap(), + ); assert!(!settled(&svc, true)); } From ac2d3fe9f00fb7df5b2517afe92f140bc4497470 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:05:29 +0200 Subject: [PATCH 19/47] fix(ci): adapt upstream tests and Clippy documentation to kube 3 --- src/lb/config.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/lb/config.rs b/src/lb/config.rs index 4bf6fd2..ee82899 100644 --- a/src/lb/config.rs +++ b/src/lb/config.rs @@ -139,6 +139,10 @@ fn default_name(cluster: Option<&str>, service: &str, namespace: &str) -> String } /// Validate the DNS-label cluster prefix at startup instead of failing on an API call. +/// +/// # Errors +/// +/// Returns an error if the name is not a valid DNS label. pub fn parse_cluster_name(value: &str) -> Result { let ascii_alnum = |ch: char| ch.is_ascii_lowercase() || ch.is_ascii_digit(); let valid = (1..=63).contains(&value.len()) From 8be59f9b9c0b59517320ec9398f75bc2b27db596 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:05:31 +0200 Subject: [PATCH 20/47] fix(ci): adapt upstream tests and Clippy documentation to kube 3 --- src/controller/node_changes.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/controller/node_changes.rs b/src/controller/node_changes.rs index 5183f09..358d439 100644 --- a/src/controller/node_changes.rs +++ b/src/controller/node_changes.rs @@ -1,5 +1,5 @@ //! Detect node changes that actually affect Hetzner target selection. -//! Kubelet heartbeats must not trigger reconciling every LoadBalancer Service. +//! Kubelet heartbeats must not trigger reconciling every `LoadBalancer` Service. use std::{ collections::{hash_map::DefaultHasher, HashMap}, From a73bae7207cb1acfeae5c639b0d21a3627db14e6 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:05:47 +0200 Subject: [PATCH 21/47] test(memo): deserialize Kubernetes deletion timestamp without chrono dependency --- src/memo.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/memo.rs b/src/memo.rs index 9ac0d39..7f6dbdb 100644 --- a/src/memo.rs +++ b/src/memo.rs @@ -229,7 +229,7 @@ mod tests { assert!(!settled(&svc, true)); let mut svc = service(); svc.metadata.deletion_timestamp = Some( - k8s_openapi::serde_json::from_str("\\"2026-01-01T00:00:00Z\\"").unwrap(), + k8s_openapi::serde_json::from_str(r#""2026-01-01T00:00:00Z""#).unwrap(), ); assert!(!settled(&svc, true)); } From 81afb8a33c3a1ec512ff3e8eed2c89fdd98f80fa Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:07:22 +0200 Subject: [PATCH 22/47] style(rust): apply rustfmt changes from CI --- src/controller/mod.rs | 67 ++++++++++++++++++++++++++----------------- 1 file changed, 41 insertions(+), 26 deletions(-) diff --git a/src/controller/mod.rs b/src/controller/mod.rs index 78ec59f..ce437b1 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -16,17 +16,17 @@ use std::{ time::Duration, }; -use futures::StreamExt; use crate::memo; +use futures::StreamExt; use k8s_openapi::api::core::v1::Service; use kube::{ Resource, ResourceExt, runtime::{ - Controller, + Controller, WatchStreamExt, controller::Action, events::{Event, EventType, Recorder, Reporter}, reflector::ObjectRef, - watcher, WatchStreamExt, + watcher, }, }; @@ -43,7 +43,6 @@ pub(crate) use nodes::{ }; pub(crate) use status::{build_ingress, patch_ingress_status}; - /// Check if a service is supported by robotlb. /// /// Returns an error if the service is not a `LoadBalancer` type or @@ -150,9 +149,8 @@ async fn reconcile_service_inner( } fn service_key(svc: &Service) -> String { - svc.uid().unwrap_or_else(|| { - format!("{}/{}", svc.namespace().unwrap_or_default(), svc.name_any()) - }) + svc.uid() + .unwrap_or_else(|| format!("{}/{}", svc.namespace().unwrap_or_default(), svc.name_any())) } /// Display the last error on the Service without leaking Hetzner API credentials. @@ -252,10 +250,11 @@ pub async fn reconcile_load_balancer( let uid = svc.uid(); let fingerprint = memo::fingerprint(&svc, &lb.targets, &lb.services); - if let Some(wait) = uid - .as_deref() - .and_then(|uid| context.memo.unchanged(uid, fingerprint, std::time::Instant::now())) - { + if let Some(wait) = uid.as_deref().and_then(|uid| { + context + .memo + .unchanged(uid, fingerprint, std::time::Instant::now()) + }) { tracing::debug!("No effective configuration change; skipping Hetzner API"); return Ok(Action::requeue(wait)); } @@ -272,7 +271,9 @@ pub async fn reconcile_load_balancer( let resync = Duration::from_secs(context.config.resync_interval); if let Some(uid) = uid { - context.memo.record(&uid, fingerprint, std::time::Instant::now(), resync); + context + .memo + .record(&uid, fingerprint, std::time::Instant::now(), resync); } Ok(Action::requeue(resync)) } @@ -280,8 +281,13 @@ pub async fn reconcile_load_balancer( /// Schedule failures independently, but pause the entire project on rate limits. #[allow(clippy::needless_pass_by_value)] pub fn on_error(svc: Arc, error: &RobotLBError, context: Arc) -> Action { - error_action(error, &context.rate_limit, &context.error_backoff, - std::time::Instant::now(), &service_key(&svc)) + error_action( + error, + &context.rate_limit, + &context.error_backoff, + std::time::Instant::now(), + &service_key(&svc), + ) } fn error_action( @@ -295,7 +301,10 @@ fn error_action( RobotLBError::SkipService => Action::await_change(), error if error.is_rate_limited() => { let delay = rate_limit.on_rate_limit(now); - tracing::warn!(?delay, "Hetzner API rate limit reached; pausing all Services"); + tracing::warn!( + ?delay, + "Hetzner API rate limit reached; pausing all Services" + ); Action::requeue(delay) } _ => Action::requeue(backoff.on_failure(service, now)), @@ -320,9 +329,7 @@ pub fn map_endpoint_slice_to_service( /// Wake all Services only when target-relevant Node state changes. /// Periodic resync still detects out-of-band Hetzner changes. -fn node_change_signals( - client: kube::Client, -) -> impl futures::Stream + Send + Sync { +fn node_change_signals(client: kube::Client) -> impl futures::Stream + Send + Sync { let (sender, receiver) = futures::channel::mpsc::unbounded(); tokio::spawn(async move { let mut changes = node_changes::NodeChanges::default(); @@ -464,13 +471,21 @@ mod tests { let backoff = crate::backoff::ErrorBackoff::default(); let now = std::time::Instant::now(); let error = RobotLBError::HCloudError("test".to_string()); - assert_eq!(error_action(&error, &gate, &backoff, now, "shop/web"), - Action::requeue(Duration::from_secs(5))); - assert_eq!(error_action(&error, &gate, &backoff, now, "shop/web"), - Action::requeue(Duration::from_secs(10))); - assert_eq!(error_action(&error, &gate, &backoff, now, "shop/api"), - Action::requeue(Duration::from_secs(5))); - assert_eq!(error_action(&RobotLBError::SkipService, &gate, &backoff, now, "shop/web"), - Action::await_change()); + assert_eq!( + error_action(&error, &gate, &backoff, now, "shop/web"), + Action::requeue(Duration::from_secs(5)) + ); + assert_eq!( + error_action(&error, &gate, &backoff, now, "shop/web"), + Action::requeue(Duration::from_secs(10)) + ); + assert_eq!( + error_action(&error, &gate, &backoff, now, "shop/api"), + Action::requeue(Duration::from_secs(5)) + ); + assert_eq!( + error_action(&RobotLBError::SkipService, &gate, &backoff, now, "shop/web"), + Action::await_change() + ); } } From aeba2545030b0403d901a3345a630564f973e949 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:07:25 +0200 Subject: [PATCH 23/47] style(rust): apply rustfmt changes from CI --- src/controller/node_changes.rs | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/src/controller/node_changes.rs b/src/controller/node_changes.rs index 358d439..c5f3a7d 100644 --- a/src/controller/node_changes.rs +++ b/src/controller/node_changes.rs @@ -102,8 +102,14 @@ mod tests { assert!(!changes.observe(&Event::InitApply(node("a", "True")))); assert!(changes.observe(&Event::InitDone)); let mut heartbeat = node("a", "True"); - heartbeat.status.as_mut().unwrap().conditions.as_mut().unwrap()[0].message = - Some("kubelet heartbeat".to_string()); + heartbeat + .status + .as_mut() + .unwrap() + .conditions + .as_mut() + .unwrap()[0] + .message = Some("kubelet heartbeat".to_string()); assert!(!changes.observe(&Event::Apply(heartbeat))); assert!(changes.observe(&Event::Apply(node("a", "False")))); assert!(!changes.observe(&Event::Apply(node("a", "False")))); @@ -111,8 +117,14 @@ mod tests { let mut cordoned = node("a", "False"); cordoned.spec.as_mut().unwrap().unschedulable = Some(true); assert!(changes.observe(&Event::Apply(cordoned.clone()))); - cordoned.status.as_mut().unwrap().addresses.as_mut().unwrap()[0].address = - "192.0.2.11".to_string(); + cordoned + .status + .as_mut() + .unwrap() + .addresses + .as_mut() + .unwrap()[0] + .address = "192.0.2.11".to_string(); assert!(changes.observe(&Event::Apply(cordoned))); assert!(changes.observe(&Event::Delete(node("a", "False")))); } From f1d497807f88bf85a34d8913171b507693fbf99d Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:07:29 +0200 Subject: [PATCH 24/47] style(rust): apply rustfmt changes from CI --- src/lb/config.rs | 20 +++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/src/lb/config.rs b/src/lb/config.rs index ee82899..c721afc 100644 --- a/src/lb/config.rs +++ b/src/lb/config.rs @@ -261,7 +261,9 @@ mod tests { assert_eq!(parsed.name, "prod.svc-name.default"); let annotated = service_with_annotations([(consts::LB_NAME_ANNOTATION, "custom")]); assert_eq!( - parse_load_balancer_config(&annotated, &config).unwrap().name, + parse_load_balancer_config(&annotated, &config) + .unwrap() + .name, "custom" ); } @@ -272,7 +274,10 @@ mod tests { let whole = format!("{part}.{part}.{part}"); let generated = default_name(Some(&part), &part, &part); assert_eq!(generated.len(), 128); - assert_eq!(generated, format!("{}-{:016x}", &whole[..111], fnv1a64(&whole))); + assert_eq!( + generated, + format!("{}-{:016x}", &whole[..111], fnv1a64(&whole)) + ); assert_eq!(fnv1a64("foobar"), 0x8594_4171_f739_67e8); assert_ne!( default_name(Some(&part), &part, &"b".repeat(63)), @@ -282,7 +287,16 @@ mod tests { #[test] fn invalid_cluster_names_are_rejected() { - for invalid in ["", "Prod", "prod.eu", "-prod", "prod-", "some_name", "prød", &"a".repeat(64)] { + for invalid in [ + "", + "Prod", + "prod.eu", + "-prod", + "prod-", + "some_name", + "prød", + &"a".repeat(64), + ] { assert!(parse_cluster_name(invalid).is_err(), "{invalid:?}"); } for valid in ["a", "prod", "eu-1", &"a".repeat(63)] { From b13fb20e936bc6f36b8121b49c013eed474a5e22 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:07:32 +0200 Subject: [PATCH 25/47] style(rust): apply rustfmt changes from CI --- src/lb/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/lb/mod.rs b/src/lb/mod.rs index 032209a..91dd882 100644 --- a/src/lb/mod.rs +++ b/src/lb/mod.rs @@ -30,8 +30,8 @@ use crate::{ }; pub(crate) use api::{HcloudLoadBalancerApi, LiveHcloudLoadBalancerApi}; -pub(crate) use config::parse_load_balancer_config; pub use config::parse_cluster_name; +pub(crate) use config::parse_load_balancer_config; use types::{ServiceReconcileAction, TargetReconcileAction, normalize_ip, service_matches_desired}; /// Struct representing a load balancer. From c7bf240a1501e318b4e55d35e49eba5120775289 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:07:36 +0200 Subject: [PATCH 26/47] style(rust): apply rustfmt changes from CI --- src/memo.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/memo.rs b/src/memo.rs index 7f6dbdb..805212a 100644 --- a/src/memo.rs +++ b/src/memo.rs @@ -87,7 +87,7 @@ pub fn fingerprint( #[cfg(test)] mod tests { - use super::{fingerprint, ReconcileMemo}; + use super::{ReconcileMemo, fingerprint}; use crate::consts; use k8s_openapi::{ api::core::v1::{ From 8d5567ea52e1c7bf5f0083167f48ba426d96edb0 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:09:49 +0200 Subject: [PATCH 27/47] style: reconcile rustfmt changes from follow-up CI --- src/controller/node_changes.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/controller/node_changes.rs b/src/controller/node_changes.rs index c5f3a7d..d690728 100644 --- a/src/controller/node_changes.rs +++ b/src/controller/node_changes.rs @@ -2,12 +2,12 @@ //! Kubelet heartbeats must not trigger reconciling every `LoadBalancer` Service. use std::{ - collections::{hash_map::DefaultHasher, HashMap}, + collections::{HashMap, hash_map::DefaultHasher}, hash::{Hash, Hasher}, }; use k8s_openapi::api::core::v1::Node; -use kube::{runtime::watcher::Event, ResourceExt}; +use kube::{ResourceExt, runtime::watcher::Event}; #[derive(Debug, Default)] pub struct NodeChanges { From 947d1220467c50d9b13864db95c35bee269ff156 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:09:52 +0200 Subject: [PATCH 28/47] style: reconcile rustfmt changes from follow-up CI --- src/memo.rs | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/src/memo.rs b/src/memo.rs index 805212a..3db3c85 100644 --- a/src/memo.rs +++ b/src/memo.rs @@ -39,9 +39,8 @@ impl ReconcileMemo { /// Drop the entry of a service after a failed reconcile, or once robotlb no /// longer serves it. pub fn settle(&self, svc: &Service, succeeded: bool) { - let serves = - svc.meta().deletion_timestamp.is_none() - && crate::controller::ensure_service_is_supported(svc).is_ok(); + let serves = svc.meta().deletion_timestamp.is_none() + && crate::controller::ensure_service_is_supported(svc).is_ok(); if !(succeeded && serves) && let Some(uid) = svc.uid() { @@ -228,9 +227,8 @@ mod tests { svc.spec.as_mut().unwrap().load_balancer_class = Some("other".to_string()); assert!(!settled(&svc, true)); let mut svc = service(); - svc.metadata.deletion_timestamp = Some( - k8s_openapi::serde_json::from_str(r#""2026-01-01T00:00:00Z""#).unwrap(), - ); + svc.metadata.deletion_timestamp = + Some(k8s_openapi::serde_json::from_str(r#""2026-01-01T00:00:00Z""#).unwrap()); assert!(!settled(&svc, true)); } From fc7929e4e3c7effca2f25f84e52ac1baf67d84b6 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:10:45 +0200 Subject: [PATCH 29/47] test(hcloud): cover structured provider failures and non-JSON responses --- src/error.rs | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/src/error.rs b/src/error.rs index 2940ef4..32af12d 100644 --- a/src/error.rs +++ b/src/error.rs @@ -261,6 +261,32 @@ mod redaction_tests { assert_eq!(redact("empty token", ""), "empty token"); } + #[test] + fn hetzner_response_describes_status_code_and_message() { + use hcloud::apis::{Error, ResponseContent, load_balancers_api::ListLoadBalancersError}; + + let error = Error::::ResponseError(ResponseContent { + status: 503_u16.try_into().unwrap(), + content: r#"{"error":{"code":"robot_unavailable","message":"back end busy"}}"#.to_string(), + entity: None, + }); + let description = super::describe_hcloud(&error); + assert!(description.contains("503"), "{description}"); + assert!(description.contains("robot_unavailable"), "{description}"); + assert!(description.contains("back end busy"), "{description}"); + } + + #[test] + fn malformed_hetzner_response_falls_back_to_http_status() { + use hcloud::apis::{Error, ResponseContent, load_balancers_api::ListLoadBalancersError}; + let error = Error::::ResponseError(ResponseContent { + status: 502_u16.try_into().unwrap(), + content: "bad gateway".to_string(), + entity: None, + }); + assert!(super::describe_hcloud(&error).contains("502")); + } + #[test] fn redacts_non_ascii_surrounding_text() { assert_eq!( From 37c379bbd1100fe00539eb93225f44633196c4a7 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 12:13:00 +0200 Subject: [PATCH 30/47] style(hcloud): apply rustfmt to provider error regression tests --- src/error.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/error.rs b/src/error.rs index 32af12d..90f167b 100644 --- a/src/error.rs +++ b/src/error.rs @@ -267,7 +267,8 @@ mod redaction_tests { let error = Error::::ResponseError(ResponseContent { status: 503_u16.try_into().unwrap(), - content: r#"{"error":{"code":"robot_unavailable","message":"back end busy"}}"#.to_string(), + content: r#"{"error":{"code":"robot_unavailable","message":"back end busy"}}"# + .to_string(), entity: None, }); let description = super::describe_hcloud(&error); From e4610956c4c80a19f52a133b9933e46efa1f9ba7 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 13:02:23 +0200 Subject: [PATCH 31/47] fix(controller): distinguish deferred work from successful reconciliation; retry unassigned LB ingress promptly --- src/controller/mod.rs | 102 +++++++++++++++++++++++++++++++++++++----- 1 file changed, 92 insertions(+), 10 deletions(-) diff --git a/src/controller/mod.rs b/src/controller/mod.rs index ce437b1..8841b85 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -81,6 +81,29 @@ pub fn ensure_service_is_supported(svc: &Service) -> RobotLBResult<()> { Ok(()) } +/// Whether reconciliation completed or intentionally deferred without applying state. +/// Deferred attempts must not clear the Service's failure backoff or populate the memo. +#[derive(Debug)] +enum ReconcileOutcome { + Applied(Action), + Deferred(Action), +} + +impl ReconcileOutcome { + const fn into_action(self) -> Action { + match self { + Self::Applied(action) | Self::Deferred(action) => action, + } + } +} + +fn should_reset_failure_backoff(result: &RobotLBResult) -> bool { + matches!( + result, + Ok(ReconcileOutcome::Applied(_)) | Err(RobotLBError::SkipService) + ) +} + /// Reconcile the service. /// /// This function is called by the controller for each service. @@ -97,8 +120,10 @@ pub async fn reconcile_service( ) -> RobotLBResult { let mut timer = ReconcileTimer::new(context.metrics.clone()); let result = reconcile_service_inner(svc.clone(), context.clone()).await; - context.memo.settle(&svc, result.is_ok()); - if matches!(result, Ok(_) | Err(RobotLBError::SkipService)) { + context + .memo + .settle(&svc, matches!(result, Ok(ReconcileOutcome::Applied(_)))); + if should_reset_failure_backoff(&result) { context.error_backoff.forget(&service_key(&svc)); } if let Err(error) = &result @@ -107,13 +132,13 @@ pub async fn reconcile_service( timer.set_failed(); publish_reconciliation_warning(&svc, &context, error).await; } - result + result.map(ReconcileOutcome::into_action) } async fn reconcile_service_inner( svc: Arc, context: Arc, -) -> RobotLBResult { +) -> RobotLBResult { if !context.is_leader.load(Ordering::Relaxed) { return Err(RobotLBError::SkipService); } @@ -122,7 +147,7 @@ async fn reconcile_service_inner( ?delay, "Hetzner API rate limit pause; skipping reconciliation" ); - return Ok(Action::requeue(delay)); + return Ok(ReconcileOutcome::Deferred(Action::requeue(delay))); } // Release resources even if a Service changed its type/class, and even if @@ -132,7 +157,7 @@ async fn reconcile_service_inner( { LoadBalancer::cleanup_service(&svc, &context).await?; finalizers::remove(context.client.clone(), &svc).await?; - return Ok(Action::await_change()); + return Ok(ReconcileOutcome::Applied(Action::await_change())); } ensure_service_is_supported(&svc)?; @@ -227,11 +252,11 @@ fn event_note(error: &RobotLBError, token: &str) -> String { /// # Errors /// /// Returns an error if node discovery fails, or if the Kubernetes/Hetzner API call fails. -pub async fn reconcile_load_balancer( +async fn reconcile_load_balancer( mut lb: LoadBalancer, svc: Arc, context: Arc, -) -> RobotLBResult { +) -> RobotLBResult { let desired_ip_type = node_ip_type(&lb); let nodes = discover_target_nodes(&svc, &context).await?; @@ -256,7 +281,7 @@ pub async fn reconcile_load_balancer( .unchanged(uid, fingerprint, std::time::Instant::now()) }) { tracing::debug!("No effective configuration change; skipping Hetzner API"); - return Ok(Action::requeue(wait)); + return Ok(ReconcileOutcome::Applied(Action::requeue(wait))); } let hcloud_lb = lb.reconcile().await?; @@ -267,6 +292,19 @@ pub async fn reconcile_load_balancer( lb.proxy_mode, lb.public_interface, ); + // Hetzner may acknowledge creation or network attachment before assigning an + // address. A successful API response is not a complete LoadBalancer reconcile + // until Kubernetes can advertise an ingress address. In particular, never + // memoize an empty result and postpone address discovery for five minutes. + if should_retry_missing_ingress(&ingress) { + tracing::info!( + service = %svc.name_any(), + "Load balancer has no ingress address yet; retrying shortly" + ); + return Ok(ReconcileOutcome::Deferred(Action::requeue( + Duration::from_secs(15), + ))); + } patch_ingress_status(&svc, &context, ingress).await?; let resync = Duration::from_secs(context.config.resync_interval); @@ -275,7 +313,11 @@ pub async fn reconcile_load_balancer( .memo .record(&uid, fingerprint, std::time::Instant::now(), resync); } - Ok(Action::requeue(resync)) + Ok(ReconcileOutcome::Applied(Action::requeue(resync))) +} + +fn should_retry_missing_ingress(ingress: &[k8s_openapi::api::core::v1::LoadBalancerIngress]) -> bool { + ingress.is_empty() } /// Schedule failures independently, but pause the entire project on rate limits. @@ -465,6 +507,46 @@ mod tests { assert!(note.is_char_boundary(note.len())); } + #[test] + fn deferred_reconciliations_preserve_backoff() { + let rate_limit = crate::rate_limit::RateLimitGate::default(); + let backoff = crate::backoff::ErrorBackoff::default(); + let now = std::time::Instant::now(); + let svc = "shop/web"; + let error = RobotLBError::HCloudError("temporary outage".to_string()); + let first = error_action(&error, &rate_limit, &backoff, now, svc); + assert_eq!(first, Action::requeue(Duration::from_secs(5))); + let delayed = Ok(ReconcileOutcome::Deferred(Action::requeue( + Duration::from_secs(60), + ))); + assert!(!should_reset_failure_backoff(&delayed)); + // This is the exact shared-429 deferral path: it must not call forget(). + if should_reset_failure_backoff(&delayed) { + backoff.forget(svc); + } + assert_eq!( + error_action(&error, &rate_limit, &backoff, now, svc), + Action::requeue(Duration::from_secs(10)) + ); + assert!(should_reset_failure_backoff(&Ok(ReconcileOutcome::Applied( + Action::await_change() + )))); + assert!(should_reset_failure_backoff(&Err(RobotLBError::SkipService))); + assert!(!should_reset_failure_backoff(&Err(error))); + } + + #[test] + fn an_unassigned_lb_address_is_not_considered_reconciled() { + // Neither a new public balancer nor a private network attachment should + // be cached while Hetzner has not returned a routable ingress address. + assert!(should_retry_missing_ingress(&[])); + let ingress = vec![k8s_openapi::api::core::v1::LoadBalancerIngress { + ip: Some("192.0.2.2".to_string()), + ..Default::default() + }]; + assert!(!should_retry_missing_ingress(&ingress)); + } + #[test] fn failure_backoff_is_independent_for_each_service() { let gate = crate::rate_limit::RateLimitGate::default(); From 4e2779c1d191510a592f1e78445c3021eea80b13 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 13:03:53 +0200 Subject: [PATCH 32/47] fix(controller): filter EndpointSlice churn by effective target node placement and observe deletions --- src/controller/slice_changes.rs | 202 ++++++++++++++++++++++++++++++++ 1 file changed, 202 insertions(+) create mode 100644 src/controller/slice_changes.rs diff --git a/src/controller/slice_changes.rs b/src/controller/slice_changes.rs new file mode 100644 index 0000000..6319e92 --- /dev/null +++ b/src/controller/slice_changes.rs @@ -0,0 +1,202 @@ +//! Track target-relevant EndpointSlice changes without reconciling on every +//! endpoint readiness or address update. A second metadata-only watch detects +//! deletions, which kube's controller watch mapper cannot identify. + +use std::{ + collections::{BTreeSet, HashMap, HashSet}, + sync::{Mutex, PoisonError}, +}; + +use k8s_openapi::api::discovery::v1::EndpointSlice; +use kube::{Resource, ResourceExt, runtime::watcher::Event}; + +type SliceKey = (String, String); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum SliceTargetMode { + /// For selector-based Local Services, actual targets are based on Pod + /// placement, not on the readiness of individual endpoints. + PodNodes, + /// Selectorless Services use ready EndpointSlice nodes for targets. + ReadyEndpointNodes, +} + +fn key(slice: &K) -> SliceKey { + (slice.namespace().unwrap_or_default(), slice.name_any()) +} + +fn target_nodes(slice: &EndpointSlice, mode: SliceTargetMode) -> BTreeSet { + slice + .endpoints + .iter() + .filter(|endpoint| { + mode == SliceTargetMode::PodNodes + || endpoint + .conditions + .as_ref() + .and_then(|conditions| conditions.ready) + != Some(false) + }) + .filter_map(|endpoint| endpoint.node_name.clone()) + .collect() +} + +#[derive(Debug, Default)] +struct State { + observed: HashMap>, + live: HashSet, + relisted: HashSet, +} + +/// A shared cache for the controller's EndpointSlice mapper and the deletion watch. +#[derive(Debug, Default)] +pub(super) struct SliceChanges { + state: Mutex, +} + +impl SliceChanges { + /// True when this slice's relevant node set changed or first appeared. + pub fn observe(&self, slice: &EndpointSlice, mode: SliceTargetMode) -> bool { + let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner); + let next = target_nodes(slice, mode); + state.observed.insert(key(slice), next.clone()) != Some(next) + } + + /// Remove stale snapshots when a Service no longer selects this resource. + pub fn forget(&self, slice: &EndpointSlice) { + self.state + .lock() + .unwrap_or_else(PoisonError::into_inner) + .observed + .remove(&key(slice)); + } + + /// True when a previously observed slice with target nodes disappears. + /// + /// A metadata-only watch avoids transferring full endpoint lists twice. + pub fn track(&self, event: &Event) -> bool { + let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner); + match event { + Event::Init => { + state.relisted.clear(); + false + } + Event::InitApply(slice) => { + state.relisted.insert(key(slice)); + false + } + Event::InitDone => { + let live = std::mem::take(&mut state.relisted); + let removed = state.observed.iter().any(|(name, nodes)| { + !live.contains(name) && !nodes.is_empty() + }); + state.observed.retain(|name, _| live.contains(name)); + state.live = live; + removed + } + Event::Apply(slice) => { + state.live.insert(key(slice)); + false + } + Event::Delete(slice) => { + let name = key(slice); + state.live.remove(&name); + state + .observed + .remove(&name) + .is_some_and(|nodes| !nodes.is_empty()) + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use k8s_openapi::{ + api::discovery::v1::{Endpoint, EndpointConditions}, + apimachinery::pkg::apis::meta::v1::ObjectMeta, + }; + + fn slice(ready: bool, node: &str) -> EndpointSlice { + EndpointSlice { + metadata: ObjectMeta { + name: Some("web-abc".to_string()), + namespace: Some("shop".to_string()), + ..Default::default() + }, + endpoints: vec![Endpoint { + addresses: vec!["192.0.2.10".to_string()], + node_name: Some(node.to_string()), + conditions: Some(EndpointConditions { + ready: Some(ready), + ..Default::default() + }), + ..Default::default() + }], + ..Default::default() + } + } + + #[test] + fn pod_placement_mode_ignores_readiness_and_address_churn() { + let changes = SliceChanges::default(); + let ready = slice(true, "node-a"); + assert!(changes.observe(&ready, SliceTargetMode::PodNodes)); + let mut unready = slice(false, "node-a"); + unready.endpoints[0].addresses = vec!["192.0.2.11".to_string()]; + assert!(!changes.observe(&unready, SliceTargetMode::PodNodes)); + assert!(!changes.observe(&ready, SliceTargetMode::PodNodes)); + assert!(changes.observe(&slice(true, "node-b"), SliceTargetMode::PodNodes)); + } + + #[test] + fn selectorless_mode_tracks_last_ready_endpoint_on_each_node() { + let changes = SliceChanges::default(); + assert!(changes.observe(&slice(true, "node-a"), SliceTargetMode::ReadyEndpointNodes)); + assert!(changes.observe(&slice(false, "node-a"), SliceTargetMode::ReadyEndpointNodes)); + assert!(!changes.observe(&slice(false, "node-a"), SliceTargetMode::ReadyEndpointNodes)); + assert!(changes.observe(&slice(true, "node-a"), SliceTargetMode::ReadyEndpointNodes)); + } + + #[test] + fn a_deleted_target_slice_triggers_reconciliation() { + let changes = SliceChanges::default(); + let initial = slice(true, "node-a"); + changes.observe(&initial, SliceTargetMode::ReadyEndpointNodes); + assert!(!changes.track(&Event::::Init)); + assert!(!changes.track(&Event::InitApply(initial.clone()))); + assert!(!changes.track(&Event::::InitDone)); + assert!(changes.track(&Event::Delete(initial.clone()))); + assert!(!changes.track(&Event::Delete(initial))); + } + + #[test] + fn relisting_detects_deleted_slices_even_if_deletion_event_was_lost() { + let changes = SliceChanges::default(); + let initial = slice(true, "node-a"); + changes.observe(&initial, SliceTargetMode::ReadyEndpointNodes); + changes.track(&Event::::Init); + changes.track(&Event::InitApply(initial)); + assert!(!changes.track(&Event::::InitDone)); + assert!(!changes.track(&Event::::Init)); + assert!(changes.track(&Event::::InitDone)); + } + + #[test] + fn unobserved_and_empty_slices_do_not_trigger_delete_signals() { + let changes = SliceChanges::default(); + let initial = slice(false, "node-a"); + changes.observe(&initial, SliceTargetMode::ReadyEndpointNodes); + assert!(!changes.track(&Event::Delete(initial))); + } + + #[test] + fn forgetting_a_slice_restores_first_observation() { + let changes = SliceChanges::default(); + let initial = slice(true, "node-a"); + assert!(changes.observe(&initial, SliceTargetMode::PodNodes)); + changes.forget(&initial); + assert!(changes.observe(&initial, SliceTargetMode::PodNodes)); + } +} From edd3f8c9477ab22c4220c916185bd87b09ab290a Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 13:04:43 +0200 Subject: [PATCH 33/47] fix(controller): filter EndpointSlice updates by target placement while handling slice deletion via metadata watch --- src/controller/mod.rs | 104 ++++++++++++++++++++++++++++++++++++++---- 1 file changed, 95 insertions(+), 9 deletions(-) diff --git a/src/controller/mod.rs b/src/controller/mod.rs index 8841b85..49d5721 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -6,10 +6,12 @@ mod event_throttle; mod node_changes; mod nodes; +mod slice_changes; mod status; use event_throttle::EventKey; pub(crate) use event_throttle::EventThrottle; +use slice_changes::{SliceChanges, SliceTargetMode}; use std::{ sync::{Arc, atomic::Ordering}, @@ -369,15 +371,37 @@ pub fn map_endpoint_slice_to_service( Some(ObjectRef::new(&service_name).within(&namespace)) } -/// Wake all Services only when target-relevant Node state changes. -/// Periodic resync still detects out-of-band Hetzner changes. -fn node_change_signals(client: kube::Client) -> impl futures::Stream + Send + Sync { +/// Whether an EndpointSlice change can move the nodes selected by a Service. +/// Preserve Pod-based Local discovery rather than switching to EndpointSlice readiness. +fn slice_target_mode(svc: &Service, dynamic_node_selector: bool) -> Option { + if !dynamic_node_selector || ensure_service_is_supported(svc).is_err() { + return None; + } + let spec = svc.spec.as_ref()?; + if spec.external_traffic_policy.as_deref() != Some("Local") { + return None; + } + Some(if spec.selector.as_ref().is_some_and(|selector| !selector.is_empty()) { + SliceTargetMode::PodNodes + } else { + SliceTargetMode::ReadyEndpointNodes + }) +} + +/// Wake all Services only when target-relevant Nodes change or an observed +/// EndpointSlice is removed. Ordinary EndpointSlice changes use the targeted mapper. +fn node_change_signals( + client: kube::Client, + slice_changes: Arc, +) -> impl futures::Stream + Send + Sync { let (sender, receiver) = futures::channel::mpsc::unbounded(); + let node_sender = sender.clone(); + let node_client = client.clone(); tokio::spawn(async move { let mut changes = node_changes::NodeChanges::default(); let mut events = std::pin::pin!( watcher::watcher( - kube::Api::::all(client), + kube::Api::::all(node_client), watcher::Config::default(), ) .default_backoff() @@ -385,7 +409,7 @@ fn node_change_signals(client: kube::Client) -> impl futures::Stream while let Some(event) = events.next().await { match event { Ok(event) => { - if changes.observe(&event) && sender.unbounded_send(()).is_err() { + if changes.observe(&event) && node_sender.unbounded_send(()).is_err() { return; } } @@ -393,6 +417,29 @@ fn node_change_signals(client: kube::Client) -> impl futures::Stream } } }); + + // kube-runtime's watch mapper receives an object but not the event kind. + // A cheap second watch on only EndpointSlice metadata ensures deletion + // events (and deletions during relists) cannot leave stale target nodes. + tokio::spawn(async move { + let mut events = std::pin::pin!( + watcher::metadata_watcher( + kube::Api::::all(client), + watcher::Config::default(), + ) + .default_backoff() + ); + while let Some(event) = events.next().await { + match event { + Ok(event) => { + if slice_changes.track(&event) && sender.unbounded_send(()).is_err() { + return; + } + } + Err(error) => tracing::warn!(%error, "EndpointSlice metadata watch failed"), + } + } + }); receiver } @@ -404,15 +451,33 @@ pub async fn run( ) { let controller_shutdown = shutdown.clone(); let hcloud_token = context.config.hcloud_token.clone(); - Controller::new( + let controller = Controller::new( kube::Api::::all(kube_client), watcher::Config::default(), - ) - .reconcile_all_on(node_change_signals(context.client.clone())) + ); + let services = controller.store(); + let slice_changes = Arc::new(SliceChanges::default()); + let dynamic_node_selector = context.config.dynamic_node_selector; + controller + .reconcile_all_on(node_change_signals( + context.client.clone(), + slice_changes.clone(), + )) .watches( kube::Api::::all(context.client.clone()), watcher::Config::default(), - |endpoint_slice| map_endpoint_slice_to_service(&endpoint_slice), + move |endpoint_slice| { + let service_ref = map_endpoint_slice_to_service(&endpoint_slice)?; + let service = services.get(&service_ref)?; + match slice_target_mode(&service, dynamic_node_selector) { + Some(mode) if slice_changes.observe(&endpoint_slice, mode) => Some(service_ref), + Some(_) => None, + None => { + slice_changes.forget(&endpoint_slice); + None + } + } + }, ) .run(reconcile_service, on_error, context) .take_until(controller_shutdown.cancelled()) @@ -547,6 +612,27 @@ mod tests { assert!(!should_retry_missing_ingress(&ingress)); } + #[test] + fn slice_watches_are_only_used_for_dynamic_local_targets() { + let mut svc = service_with_spec(service_spec("LoadBalancer", None, vec![])); + assert_eq!(slice_target_mode(&svc, true), None); + svc.spec.as_mut().unwrap().external_traffic_policy = Some("Local".to_string()); + assert_eq!( + slice_target_mode(&svc, true), + Some(SliceTargetMode::ReadyEndpointNodes) + ); + svc.spec.as_mut().unwrap().selector = Some( + [("app".to_string(), "web".to_string())].into(), + ); + assert_eq!( + slice_target_mode(&svc, true), + Some(SliceTargetMode::PodNodes) + ); + assert_eq!(slice_target_mode(&svc, false), None); + svc.spec.as_mut().unwrap().load_balancer_class = Some("other".to_string()); + assert_eq!(slice_target_mode(&svc, true), None); + } + #[test] fn failure_backoff_is_independent_for_each_service() { let gate = crate::rate_limit::RateLimitGate::default(); From bfdc3b3c1225023c1d66f604530ba0250ddedae9 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 13:05:37 +0200 Subject: [PATCH 34/47] build(docker): use the tested lockfile for production release builds --- Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index e6f3960..c8fb94d 100644 --- a/Dockerfile +++ b/Dockerfile @@ -8,7 +8,7 @@ COPY . . ENV RUST_BACKTRACE=1 ENV JEMALLOC_SYS_WITH_MALLOC_CONF="background_thread:true,tcache:false,dirty_decay_ms:100,muzzy_decay_ms:100,abort_conf:true" -RUN cargo build --release +RUN cargo build --release --locked FROM debian:bookworm AS base From e073b389708ca150292e680d81a2a8201682a5e9 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 13:05:49 +0200 Subject: [PATCH 35/47] test(ci): build and smoke-test the actual runtime Docker image --- .github/workflows/tests.yaml | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index 80b25d8..6dc667c 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -17,6 +17,19 @@ jobs: - name: Run Rust unit tests run: cargo test --all-targets --locked + docker-smoke: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Build locked release image + run: docker build --progress=plain -t robotlb:pr . + - name: Verify runtime dependencies and CLI startup + run: | + docker run --rm --entrypoint /bin/sh robotlb:pr -c ' + ldd /usr/local/bin/robotlb + /usr/local/bin/robotlb --help > /dev/null + ' + helm-unittest: runs-on: ubuntu-latest steps: From 94cf07153fd983504b511560fceed342d0b9c3b0 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 13:06:37 +0200 Subject: [PATCH 36/47] style(controller): satisfy rustfmt and const-function Clippy checks --- src/controller/mod.rs | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/src/controller/mod.rs b/src/controller/mod.rs index 49d5721..32b12b3 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -99,7 +99,7 @@ impl ReconcileOutcome { } } -fn should_reset_failure_backoff(result: &RobotLBResult) -> bool { +const fn should_reset_failure_backoff(result: &RobotLBResult) -> bool { matches!( result, Ok(ReconcileOutcome::Applied(_)) | Err(RobotLBError::SkipService) @@ -318,7 +318,9 @@ async fn reconcile_load_balancer( Ok(ReconcileOutcome::Applied(Action::requeue(resync))) } -fn should_retry_missing_ingress(ingress: &[k8s_openapi::api::core::v1::LoadBalancerIngress]) -> bool { +const fn should_retry_missing_ingress( + ingress: &[k8s_openapi::api::core::v1::LoadBalancerIngress], +) -> bool { ingress.is_empty() } @@ -593,10 +595,12 @@ mod tests { error_action(&error, &rate_limit, &backoff, now, svc), Action::requeue(Duration::from_secs(10)) ); - assert!(should_reset_failure_backoff(&Ok(ReconcileOutcome::Applied( - Action::await_change() - )))); - assert!(should_reset_failure_backoff(&Err(RobotLBError::SkipService))); + assert!(should_reset_failure_backoff(&Ok( + ReconcileOutcome::Applied(Action::await_change()) + ))); + assert!(should_reset_failure_backoff(&Err( + RobotLBError::SkipService + ))); assert!(!should_reset_failure_backoff(&Err(error))); } From 8a0e79cebd8af1e27765722f30942d26efecbd9e Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 13:07:38 +0200 Subject: [PATCH 37/47] docs: describe filtered EndpointSlice triggers, deferred provisioning and preserved backoff --- README.md | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 905d700..04bdc25 100644 --- a/README.md +++ b/README.md @@ -242,17 +242,23 @@ The class is effectively immutable after creation; recreating the Service can replace the LB and change its IP. Node changes that affect targets (readiness, labels, cordon state or addresses) -trigger reconciliation immediately; kubelet heartbeats do not. EndpointSlice -updates also reconcile their Service. On otherwise unchanged Services, the -operator compares the desired Service spec, `robotlb/` annotations, target IPs -and listener ports with the last successful reconcile and skips Hetzner requests. +trigger reconciliation immediately; kubelet heartbeats do not. For `Local` +Services, EndpointSlice updates reconcile only when relevant target nodes change: +Pod-selector Services ignore endpoint readiness/address churn, while selectorless +Services track ready endpoint nodes. EndpointSlice deletions also trigger +reconciliation. On otherwise unchanged Services, the operator compares the +desired Service spec, `robotlb/` annotations, target IPs and listener ports with +the last successful reconcile and skips Hetzner requests. Every `ROBOTLB_RESYNC_INTERVAL` seconds (default **300**) it performs a full check anyway to recover out-of-band changes made in Hetzner. A failing Service retries independently with exponential backoff from 5 seconds up to 5 minutes, without suppressing reconciliation triggered by Kubernetes -watch events. Project-wide Hetzner API rate limits use the separate shared gate. -Errors remain visible in logs and Kubernetes Warning Events. +watch events. Project-wide Hetzner API rate limits use the separate shared gate +without resetting the per-Service failure backoff. After Hetzner creates an LB, +the operator retries every 15 seconds until an ingress address is available; +an incomplete provisioning response is never cached. Errors remain visible in +logs and Kubernetes Warning Events. When several clusters share one Hetzner Cloud project, set `ROBOTLB_CLUSTER_NAME` (for example `prod`) to a DNS-label prefix From 977f4b40c9ba4025c556bd14d85969cd95d6c712 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 13:08:22 +0200 Subject: [PATCH 38/47] style(controller): apply rustfmt and clarify EndpointSlice documentation --- src/controller/mod.rs | 128 ++++++++++++++++++++++-------------------- 1 file changed, 67 insertions(+), 61 deletions(-) diff --git a/src/controller/mod.rs b/src/controller/mod.rs index 32b12b3..84b6cf2 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -373,8 +373,8 @@ pub fn map_endpoint_slice_to_service( Some(ObjectRef::new(&service_name).within(&namespace)) } -/// Whether an EndpointSlice change can move the nodes selected by a Service. -/// Preserve Pod-based Local discovery rather than switching to EndpointSlice readiness. +/// Whether an `EndpointSlice` change can move the nodes selected by a Service. +/// Preserve Pod-based Local discovery rather than switching to `EndpointSlice` readiness. fn slice_target_mode(svc: &Service, dynamic_node_selector: bool) -> Option { if !dynamic_node_selector || ensure_service_is_supported(svc).is_err() { return None; @@ -383,15 +383,21 @@ fn slice_target_mode(svc: &Service, dynamic_node_selector: bool) -> Option, @@ -461,59 +467,61 @@ pub async fn run( let slice_changes = Arc::new(SliceChanges::default()); let dynamic_node_selector = context.config.dynamic_node_selector; controller - .reconcile_all_on(node_change_signals( - context.client.clone(), - slice_changes.clone(), - )) - .watches( - kube::Api::::all(context.client.clone()), - watcher::Config::default(), - move |endpoint_slice| { - let service_ref = map_endpoint_slice_to_service(&endpoint_slice)?; - let service = services.get(&service_ref)?; - match slice_target_mode(&service, dynamic_node_selector) { - Some(mode) if slice_changes.observe(&endpoint_slice, mode) => Some(service_ref), - Some(_) => None, - None => { - slice_changes.forget(&endpoint_slice); - None - } - } - }, - ) - .run(reconcile_service, on_error, context) - .take_until(controller_shutdown.cancelled()) - .for_each(|reconciliation_result| { - let token = hcloud_token.clone(); - async move { - match reconciliation_result { - Ok((service, _action)) => { - tracing::info!( - "Reconciliation of a service {} was successful", - service.name - ); - } - Err(kube::runtime::controller::Error::ReconcilerFailed( - RobotLBError::SkipService, - _, - )) => {} - Err(kube::runtime::controller::Error::ReconcilerFailed(error, service)) => { - tracing::error!( - service = %service, - error = %crate::error::redact(&error.to_string(), &token), - "Service reconciliation failed" - ); + .reconcile_all_on(node_change_signals( + context.client.clone(), + slice_changes.clone(), + )) + .watches( + kube::Api::::all( + context.client.clone(), + ), + watcher::Config::default(), + move |endpoint_slice| { + let service_ref = map_endpoint_slice_to_service(&endpoint_slice)?; + let service = services.get(&service_ref)?; + match slice_target_mode(&service, dynamic_node_selector) { + Some(mode) if slice_changes.observe(&endpoint_slice, mode) => Some(service_ref), + Some(_) => None, + None => { + slice_changes.forget(&endpoint_slice); + None + } } - Err(error) => { - tracing::error!( - error = %crate::error::redact(&error.to_string(), &token), - "Controller reconciliation failed" - ); + }, + ) + .run(reconcile_service, on_error, context) + .take_until(controller_shutdown.cancelled()) + .for_each(|reconciliation_result| { + let token = hcloud_token.clone(); + async move { + match reconciliation_result { + Ok((service, _action)) => { + tracing::info!( + "Reconciliation of a service {} was successful", + service.name + ); + } + Err(kube::runtime::controller::Error::ReconcilerFailed( + RobotLBError::SkipService, + _, + )) => {} + Err(kube::runtime::controller::Error::ReconcilerFailed(error, service)) => { + tracing::error!( + service = %service, + error = %crate::error::redact(&error.to_string(), &token), + "Service reconciliation failed" + ); + } + Err(error) => { + tracing::error!( + error = %crate::error::redact(&error.to_string(), &token), + "Controller reconciliation failed" + ); + } } } - } - }) - .await; + }) + .await; } #[cfg(test)] @@ -625,9 +633,7 @@ mod tests { slice_target_mode(&svc, true), Some(SliceTargetMode::ReadyEndpointNodes) ); - svc.spec.as_mut().unwrap().selector = Some( - [("app".to_string(), "web".to_string())].into(), - ); + svc.spec.as_mut().unwrap().selector = Some([("app".to_string(), "web".to_string())].into()); assert_eq!( slice_target_mode(&svc, true), Some(SliceTargetMode::PodNodes) From eec43d7de5a9d0ae69c58b279ada6b1b720968df Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 13:08:27 +0200 Subject: [PATCH 39/47] style(controller): apply rustfmt and clarify EndpointSlice documentation --- src/controller/slice_changes.rs | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/src/controller/slice_changes.rs b/src/controller/slice_changes.rs index 6319e92..2083cda 100644 --- a/src/controller/slice_changes.rs +++ b/src/controller/slice_changes.rs @@ -1,4 +1,4 @@ -//! Track target-relevant EndpointSlice changes without reconciling on every +//! Track target-relevant `EndpointSlice` changes without reconciling on every //! endpoint readiness or address update. A second metadata-only watch detects //! deletions, which kube's controller watch mapper cannot identify. @@ -17,7 +17,7 @@ pub(super) enum SliceTargetMode { /// For selector-based Local Services, actual targets are based on Pod /// placement, not on the readiness of individual endpoints. PodNodes, - /// Selectorless Services use ready EndpointSlice nodes for targets. + /// Selectorless Services use ready `EndpointSlice` nodes for targets. ReadyEndpointNodes, } @@ -48,7 +48,7 @@ struct State { relisted: HashSet, } -/// A shared cache for the controller's EndpointSlice mapper and the deletion watch. +/// A shared cache for the controller's `EndpointSlice` mapper and the deletion watch. #[derive(Debug, Default)] pub(super) struct SliceChanges { state: Mutex, @@ -87,9 +87,10 @@ impl SliceChanges { } Event::InitDone => { let live = std::mem::take(&mut state.relisted); - let removed = state.observed.iter().any(|(name, nodes)| { - !live.contains(name) && !nodes.is_empty() - }); + let removed = state + .observed + .iter() + .any(|(name, nodes)| !live.contains(name) && !nodes.is_empty()); state.observed.retain(|name, _| live.contains(name)); state.live = live; removed From b74a2cd23bb4c7b50e8ba6194c193de413ec04a8 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 15:12:14 +0200 Subject: [PATCH 40/47] fix(memo): cache successful ingress status and track bounded provisioning --- src/memo.rs | 129 ++++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 106 insertions(+), 23 deletions(-) diff --git a/src/memo.rs b/src/memo.rs index 3db3c85..57d0269 100644 --- a/src/memo.rs +++ b/src/memo.rs @@ -1,4 +1,4 @@ -use k8s_openapi::api::core::v1::Service; +use k8s_openapi::api::core::v1::{LoadBalancerIngress, Service}; use kube::{Resource, ResourceExt}; use std::{ collections::{BTreeMap, BTreeSet, HashMap}, @@ -11,29 +11,71 @@ use std::{ /// watch event that changed nothing robotlb acts on costs no Hetzner requests. #[derive(Default)] pub struct ReconcileMemo { - entries: Mutex>, + entries: Mutex)>>, + pending_ingress: Mutex>, } impl ReconcileMemo { /// How long the recorded state stays valid, when the service is still in it. - pub fn unchanged(&self, uid: &str, fingerprint: u64, now: Instant) -> Option { - let (recorded, until) = *self + pub fn unchanged( + &self, + uid: &str, + fingerprint: u64, + now: Instant, + ) -> Option<(Duration, Vec)> { + let (recorded, until, ingress) = self .entries .lock() .unwrap_or_else(PoisonError::into_inner) - .get(uid)?; + .get(uid)? + .clone(); if recorded != fingerprint { return None; } - Some(until.saturating_duration_since(now)).filter(|left| !left.is_zero()) + let left = until.saturating_duration_since(now); + (!left.is_zero()).then_some((left, ingress)) } - /// Keep the state for `valid_for`, the time until the reconcile is due anyway. - pub fn record(&self, uid: &str, fingerprint: u64, now: Instant, valid_for: Duration) { + /// Keep the successful desired state and ingress status for `valid_for`. + pub fn record( + &self, + uid: &str, + fingerprint: u64, + now: Instant, + valid_for: Duration, + ingress: Vec, + ) { self.entries .lock() .unwrap_or_else(PoisonError::into_inner) - .insert(uid.to_string(), (fingerprint, now + valid_for)); + .insert(uid.to_string(), (fingerprint, now + valid_for, ingress)); + } + + /// An incomplete ingress is normal briefly; escalate only after the deadline. + /// Changes to the effective configuration restart the deadline. + pub fn missing_ingress_overdue( + &self, + uid: &str, + fingerprint: u64, + now: Instant, + deadline: Duration, + ) -> bool { + let mut pending = self + .pending_ingress + .lock() + .unwrap_or_else(PoisonError::into_inner); + let entry = pending.entry(uid.to_string()).or_insert((fingerprint, now)); + if entry.0 != fingerprint { + *entry = (fingerprint, now); + } + now.saturating_duration_since(entry.1) >= deadline + } + + pub fn clear_pending(&self, uid: &str) { + self.pending_ingress + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(uid); } /// Drop the entry of a service after a failed reconcile, or once robotlb no @@ -41,10 +83,13 @@ impl ReconcileMemo { pub fn settle(&self, svc: &Service, succeeded: bool) { let serves = svc.meta().deletion_timestamp.is_none() && crate::controller::ensure_service_is_supported(svc).is_ok(); - if !(succeeded && serves) - && let Some(uid) = svc.uid() - { - self.forget(&uid); + if let Some(uid) = svc.uid() { + if !(succeeded && serves) { + self.forget(&uid); + } + if !serves { + self.clear_pending(&uid); + } } } @@ -147,10 +192,10 @@ mod tests { fn an_unchanged_service_is_skipped_until_the_resync() { let memo = ReconcileMemo::default(); let start = Instant::now(); - memo.record("uid-1", 7, start, RESYNC); + memo.record("uid-1", 7, start, RESYNC, vec![]); assert_eq!( memo.unchanged("uid-1", 7, start + Duration::from_secs(100)), - Some(Duration::from_secs(200)) + Some((Duration::from_secs(200), vec![])) ); assert_eq!(memo.unchanged("uid-1", 7, start + RESYNC), None); assert_eq!( @@ -164,10 +209,10 @@ mod tests { let memo = ReconcileMemo::default(); let start = Instant::now(); let retry = Duration::from_secs(30); - memo.record("uid-1", 7, start, retry); + memo.record("uid-1", 7, start, retry, vec![]); assert_eq!( memo.unchanged("uid-1", 7, start + Duration::from_secs(10)), - Some(Duration::from_secs(20)) + Some((Duration::from_secs(20), vec![])) ); assert_eq!(memo.unchanged("uid-1", 7, start + retry), None); } @@ -177,16 +222,54 @@ mod tests { let memo = ReconcileMemo::default(); let start = Instant::now(); assert_eq!(memo.unchanged("uid-1", 7, start), None); - memo.record("uid-1", 7, start, RESYNC); + memo.record("uid-1", 7, start, RESYNC, vec![]); assert_eq!(memo.unchanged("uid-1", 8, start), None); assert_eq!(memo.unchanged("uid-2", 7, start), None); } + #[test] + fn cached_ingress_supports_status_repair_without_hcloud() { + let memo = ReconcileMemo::default(); + let start = Instant::now(); + let ingress = vec![LoadBalancerIngress { + ip: Some("192.0.2.10".to_string()), + ..Default::default() + }]; + memo.record("uid-1", 7, start, RESYNC, ingress.clone()); + assert_eq!( + memo.unchanged("uid-1", 7, start), + Some((RESYNC, ingress)) + ); + } + + #[test] + fn pending_ingress_times_out_and_resets_on_change_or_recovery() { + let memo = ReconcileMemo::default(); + let start = Instant::now(); + let timeout = Duration::from_secs(180); + assert!(!memo.missing_ingress_overdue("uid-1", 7, start, timeout)); + assert!(!memo.missing_ingress_overdue( + "uid-1", + 7, + start + Duration::from_secs(179), + timeout + )); + assert!(memo.missing_ingress_overdue("uid-1", 7, start + timeout, timeout)); + assert!(!memo.missing_ingress_overdue("uid-1", 8, start + timeout, timeout)); + memo.clear_pending("uid-1"); + assert!(!memo.missing_ingress_overdue( + "uid-1", + 8, + start + timeout + Duration::from_secs(1), + timeout + )); + } + #[test] fn a_forgotten_service_is_reconciled() { let memo = ReconcileMemo::default(); let start = Instant::now(); - memo.record("uid-1", 7, start, RESYNC); + memo.record("uid-1", 7, start, RESYNC, vec![]); memo.forget("uid-1"); assert_eq!(memo.unchanged("uid-1", 7, start), None); } @@ -195,15 +278,15 @@ mod tests { fn a_new_record_restarts_the_resync() { let memo = ReconcileMemo::default(); let start = Instant::now(); - memo.record("uid-1", 7, start, RESYNC); - memo.record("uid-1", 8, start + RESYNC, RESYNC); - assert_eq!(memo.unchanged("uid-1", 8, start + RESYNC), Some(RESYNC)); + memo.record("uid-1", 7, start, RESYNC, vec![]); + memo.record("uid-1", 8, start + RESYNC, RESYNC, vec![]); + assert_eq!(memo.unchanged("uid-1", 8, start + RESYNC), Some((RESYNC, vec![]))); } fn settled(svc: &Service, succeeded: bool) -> bool { let memo = ReconcileMemo::default(); let start = Instant::now(); - memo.record("uid-1", 7, start, RESYNC); + memo.record("uid-1", 7, start, RESYNC, vec![]); memo.settle(svc, succeeded); memo.unchanged("uid-1", 7, start).is_some() } From 3bf8f828ce868e96122e576a778fbf6ffc6b1c17 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 15:12:46 +0200 Subject: [PATCH 41/47] fix(controller): reconcile on EndpointSlice ownership changes and relists --- src/controller/slice_changes.rs | 87 ++++++++++++++++++++++++++++++--- 1 file changed, 79 insertions(+), 8 deletions(-) diff --git a/src/controller/slice_changes.rs b/src/controller/slice_changes.rs index 2083cda..92d9a2d 100644 --- a/src/controller/slice_changes.rs +++ b/src/controller/slice_changes.rs @@ -25,6 +25,15 @@ fn key(slice: &K) -> SliceKey { (slice.namespace().unwrap_or_default(), slice.name_any()) } +fn owner(slice: &K) -> Option { + slice + .meta() + .labels + .as_ref()? + .get("kubernetes.io/service-name") + .cloned() +} + fn target_nodes(slice: &EndpointSlice, mode: SliceTargetMode) -> BTreeSet { slice .endpoints @@ -44,8 +53,8 @@ fn target_nodes(slice: &EndpointSlice, mode: SliceTargetMode) -> BTreeSet>, - live: HashSet, - relisted: HashSet, + live: HashMap>, + relisted: HashMap>, } /// A shared cache for the controller's `EndpointSlice` mapper and the deletion watch. @@ -82,7 +91,7 @@ impl SliceChanges { false } Event::InitApply(slice) => { - state.relisted.insert(key(slice)); + state.relisted.insert(key(slice), owner(slice)); false } Event::InitDone => { @@ -90,14 +99,31 @@ impl SliceChanges { let removed = state .observed .iter() - .any(|(name, nodes)| !live.contains(name) && !nodes.is_empty()); - state.observed.retain(|name, _| live.contains(name)); + .any(|(name, nodes)| !live.contains_key(name) && !nodes.is_empty()); + // A relist may be the only evidence of a Service label change. + let changed: HashSet<_> = live + .iter() + .filter(|(name, next)| state.live.get(*name).is_some_and(|old| old != *next)) + .map(|(name, _)| name.clone()) + .collect(); + state + .observed + .retain(|name, _| live.contains_key(name) && !changed.contains(name)); state.live = live; - removed + removed || !changed.is_empty() } Event::Apply(slice) => { - state.live.insert(key(slice)); - false + let name = key(slice); + let next = owner(slice); + let changed = state + .live + .insert(name.clone(), next.clone()) + .is_some_and(|old| old != next); + if changed { + // The main watch may not see a node-set change for the new owner. + state.observed.remove(&name); + } + changed } Event::Delete(slice) => { let name = key(slice); @@ -192,6 +218,51 @@ mod tests { assert!(!changes.track(&Event::Delete(initial))); } + fn with_owner(mut slice: EndpointSlice, service: Option<&str>) -> EndpointSlice { + slice.metadata.labels = service.map(|name| { + [("kubernetes.io/service-name".to_string(), name.to_string())].into() + }); + slice + } + + #[test] + fn relabeling_a_slice_wakes_old_and_new_service_even_with_unchanged_nodes() { + let changes = SliceChanges::default(); + let first = with_owner(slice(true, "node-a"), Some("service-a")); + changes.observe(&first, SliceTargetMode::ReadyEndpointNodes); + changes.track(&Event::::Init); + changes.track(&Event::InitApply(first.clone())); + assert!(!changes.track(&Event::::InitDone)); + + let second = with_owner(first.clone(), Some("service-b")); + // Main watch may get the event before the metadata watch. + assert!(!changes.observe(&second, SliceTargetMode::ReadyEndpointNodes)); + assert!(changes.track(&Event::Apply(second.clone()))); + assert!(changes.observe(&second, SliceTargetMode::ReadyEndpointNodes)); + assert!(!changes.track(&Event::Apply(second.clone()))); + + // Losing the owner label must also wake the previous Service. + assert!(changes.track(&Event::Apply(with_owner(second, None)))); + } + + #[test] + fn relist_detects_service_reassignment_without_endpoint_changes() { + let changes = SliceChanges::default(); + let first = with_owner(slice(true, "node-a"), Some("service-a")); + changes.observe(&first, SliceTargetMode::ReadyEndpointNodes); + changes.track(&Event::::Init); + changes.track(&Event::InitApply(first.clone())); + assert!(!changes.track(&Event::::InitDone)); + + changes.track(&Event::::Init); + changes.track(&Event::InitApply(with_owner(first.clone(), Some("service-b")))); + assert!(changes.track(&Event::::InitDone)); + assert!(changes.observe( + &with_owner(first, Some("service-b")), + SliceTargetMode::ReadyEndpointNodes + )); + } + #[test] fn forgetting_a_slice_restores_first_observation() { let changes = SliceChanges::default(); From bb0e17ace2182db73737e4d256f77a08464586f8 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 15:13:17 +0200 Subject: [PATCH 42/47] fix(controller): repair cached ingress and report stalled provisioning with fixed polling --- src/controller/mod.rs | 53 +++++++++++++++++++++++++++++++++++-------- 1 file changed, 43 insertions(+), 10 deletions(-) diff --git a/src/controller/mod.rs b/src/controller/mod.rs index 84b6cf2..724247d 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -99,6 +99,9 @@ impl ReconcileOutcome { } } +const INGRESS_RETRY_INTERVAL: Duration = Duration::from_secs(15); +const INGRESS_PROVISIONING_TIMEOUT: Duration = Duration::from_secs(180); + const fn should_reset_failure_backoff(result: &RobotLBResult) -> bool { matches!( result, @@ -225,10 +228,10 @@ async fn publish_reconciliation_warning( } const fn warning_reason(error: &RobotLBError) -> &'static str { - if matches!(error, RobotLBError::InvalidLoadBalancerConfiguration(_)) { - "InvalidLoadBalancerConfiguration" - } else { - "SyncLoadBalancerFailed" + match error { + RobotLBError::InvalidLoadBalancerConfiguration(_) => "InvalidLoadBalancerConfiguration", + RobotLBError::LoadBalancerProvisioningTimeout => "LoadBalancerProvisioningTimeout", + _ => "SyncLoadBalancerFailed", } } @@ -277,12 +280,14 @@ async fn reconcile_load_balancer( let uid = svc.uid(); let fingerprint = memo::fingerprint(&svc, &lb.targets, &lb.services); - if let Some(wait) = uid.as_deref().and_then(|uid| { + if let Some((wait, ingress)) = uid.as_deref().and_then(|uid| { context .memo .unchanged(uid, fingerprint, std::time::Instant::now()) }) { tracing::debug!("No effective configuration change; skipping Hetzner API"); + // Status is not part of the fingerprint: repair it without Hetzner calls. + patch_ingress_status(&svc, &context, ingress).await?; return Ok(ReconcileOutcome::Applied(Action::requeue(wait))); } @@ -299,21 +304,34 @@ async fn reconcile_load_balancer( // until Kubernetes can advertise an ingress address. In particular, never // memoize an empty result and postpone address discovery for five minutes. if should_retry_missing_ingress(&ingress) { + if uid.as_deref().is_some_and(|uid| { + context.memo.missing_ingress_overdue( + uid, + fingerprint, + std::time::Instant::now(), + INGRESS_PROVISIONING_TIMEOUT, + ) + }) { + return Err(RobotLBError::LoadBalancerProvisioningTimeout); + } tracing::info!( service = %svc.name_any(), "Load balancer has no ingress address yet; retrying shortly" ); return Ok(ReconcileOutcome::Deferred(Action::requeue( - Duration::from_secs(15), + INGRESS_RETRY_INTERVAL, ))); } - patch_ingress_status(&svc, &context, ingress).await?; + patch_ingress_status(&svc, &context, ingress.clone()).await?; + if let Some(uid) = uid.as_deref() { + context.memo.clear_pending(uid); + } let resync = Duration::from_secs(context.config.resync_interval); if let Some(uid) = uid { context .memo - .record(&uid, fingerprint, std::time::Instant::now(), resync); + .record(&uid, fingerprint, std::time::Instant::now(), resync, ingress); } Ok(ReconcileOutcome::Applied(Action::requeue(resync))) } @@ -345,6 +363,8 @@ fn error_action( ) -> Action { match error { RobotLBError::SkipService => Action::await_change(), + // Provisioning keeps a fixed polling interval even after emitting an error. + RobotLBError::LoadBalancerProvisioningTimeout => Action::requeue(INGRESS_RETRY_INTERVAL), error if error.is_rate_limited() => { let delay = rate_limit.on_rate_limit(now); tracing::warn!( @@ -396,8 +416,9 @@ fn slice_target_mode(svc: &Service, dynamic_node_selector: bool) -> Option, @@ -612,6 +633,18 @@ mod tests { assert!(!should_reset_failure_backoff(&Err(error))); } + #[test] + fn provisioning_timeout_emits_a_warning_and_keeps_a_fixed_retry() { + let gate = crate::rate_limit::RateLimitGate::default(); + let backoff = crate::backoff::ErrorBackoff::default(); + let error = RobotLBError::LoadBalancerProvisioningTimeout; + assert_eq!(warning_reason(&error), "LoadBalancerProvisioningTimeout"); + assert_eq!( + error_action(&error, &gate, &backoff, std::time::Instant::now(), "shop/web"), + Action::requeue(INGRESS_RETRY_INTERVAL) + ); + } + #[test] fn an_unassigned_lb_address_is_not_considered_reconciled() { // Neither a new public balancer nor a private network attachment should From 9c1f90c70475728f744f6a57b6f328d5e93b7b00 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 15:13:36 +0200 Subject: [PATCH 43/47] fix(error): classify missing ingress provisioning deadline --- src/error.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/error.rs b/src/error.rs index 90f167b..136c8c7 100644 --- a/src/error.rs +++ b/src/error.rs @@ -84,6 +84,8 @@ pub enum RobotLBError { HCloudError(String), #[error("Invalid load balancer configuration: {0}")] InvalidLoadBalancerConfiguration(String), + #[error("Load balancer has no ingress address after the provisioning deadline")] + LoadBalancerProvisioningTimeout, #[error("Cannot list Hetzner load balancer types: {}", describe_hcloud(.0))] HCloudListLoadBalancerTypesError( #[from] From 510e7823c62caf7927b5b84b226757d246f39637 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 15:13:41 +0200 Subject: [PATCH 44/47] docs: explain cached ingress repair and bounded provisioning warning --- README.md | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 04bdc25..9fede5d 100644 --- a/README.md +++ b/README.md @@ -245,10 +245,11 @@ Node changes that affect targets (readiness, labels, cordon state or addresses) trigger reconciliation immediately; kubelet heartbeats do not. For `Local` Services, EndpointSlice updates reconcile only when relevant target nodes change: Pod-selector Services ignore endpoint readiness/address churn, while selectorless -Services track ready endpoint nodes. EndpointSlice deletions also trigger -reconciliation. On otherwise unchanged Services, the operator compares the -desired Service spec, `robotlb/` annotations, target IPs and listener ports with -the last successful reconcile and skips Hetzner requests. +Services track ready endpoint nodes. EndpointSlice deletions and ownership-label +changes also trigger reconciliation. On otherwise unchanged Services, the operator +compares the desired Service spec, `robotlb/` annotations, target IPs and listener +ports with the last successful reconcile and skips Hetzner requests, but repairs +unexpected ingress status changes from the last successful status snapshot. Every `ROBOTLB_RESYNC_INTERVAL` seconds (default **300**) it performs a full check anyway to recover out-of-band changes made in Hetzner. @@ -256,9 +257,11 @@ A failing Service retries independently with exponential backoff from 5 seconds up to 5 minutes, without suppressing reconciliation triggered by Kubernetes watch events. Project-wide Hetzner API rate limits use the separate shared gate without resetting the per-Service failure backoff. After Hetzner creates an LB, -the operator retries every 15 seconds until an ingress address is available; -an incomplete provisioning response is never cached. Errors remain visible in -logs and Kubernetes Warning Events. +the operator retries every 15 seconds until an ingress address is available. +After 180 seconds without an ingress address, it reports a reconciliation error +and emits throttled Kubernetes Warning Events while continuing the same 15-second +polling interval. Incomplete provisioning is never memoized. Other errors use +the normal failure backoff. When several clusters share one Hetzner Cloud project, set `ROBOTLB_CLUSTER_NAME` (for example `prod`) to a DNS-label prefix From 5522515dcd357eb1bc31f68cd4519cec667df22c Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 15:16:14 +0200 Subject: [PATCH 45/47] fix(memo): simplify cached snapshot and release pending guard before returning --- src/memo.rs | 40 ++++++++++++++++++++++++++++------------ 1 file changed, 28 insertions(+), 12 deletions(-) diff --git a/src/memo.rs b/src/memo.rs index 57d0269..8a6ff31 100644 --- a/src/memo.rs +++ b/src/memo.rs @@ -9,9 +9,16 @@ use std::{ /// What each service looked like after its last successful reconcile, so that a /// watch event that changed nothing robotlb acts on costs no Hetzner requests. +#[derive(Clone)] +struct SuccessfulReconcile { + fingerprint: u64, + until: Instant, + ingress: Vec, +} + #[derive(Default)] pub struct ReconcileMemo { - entries: Mutex)>>, + entries: Mutex>, pending_ingress: Mutex>, } @@ -23,17 +30,17 @@ impl ReconcileMemo { fingerprint: u64, now: Instant, ) -> Option<(Duration, Vec)> { - let (recorded, until, ingress) = self + let snapshot = self .entries .lock() .unwrap_or_else(PoisonError::into_inner) .get(uid)? .clone(); - if recorded != fingerprint { + if snapshot.fingerprint != fingerprint { return None; } - let left = until.saturating_duration_since(now); - (!left.is_zero()).then_some((left, ingress)) + let left = snapshot.until.saturating_duration_since(now); + (!left.is_zero()).then_some((left, snapshot.ingress)) } /// Keep the successful desired state and ingress status for `valid_for`. @@ -48,7 +55,14 @@ impl ReconcileMemo { self.entries .lock() .unwrap_or_else(PoisonError::into_inner) - .insert(uid.to_string(), (fingerprint, now + valid_for, ingress)); + .insert( + uid.to_string(), + SuccessfulReconcile { + fingerprint, + until: now + valid_for, + ingress, + }, + ); } /// An incomplete ingress is normal briefly; escalate only after the deadline. @@ -68,7 +82,9 @@ impl ReconcileMemo { if entry.0 != fingerprint { *entry = (fingerprint, now); } - now.saturating_duration_since(entry.1) >= deadline + let overdue = now.saturating_duration_since(entry.1) >= deadline; + drop(pending); + overdue } pub fn clear_pending(&self, uid: &str) { @@ -236,10 +252,7 @@ mod tests { ..Default::default() }]; memo.record("uid-1", 7, start, RESYNC, ingress.clone()); - assert_eq!( - memo.unchanged("uid-1", 7, start), - Some((RESYNC, ingress)) - ); + assert_eq!(memo.unchanged("uid-1", 7, start), Some((RESYNC, ingress))); } #[test] @@ -280,7 +293,10 @@ mod tests { let start = Instant::now(); memo.record("uid-1", 7, start, RESYNC, vec![]); memo.record("uid-1", 8, start + RESYNC, RESYNC, vec![]); - assert_eq!(memo.unchanged("uid-1", 8, start + RESYNC), Some((RESYNC, vec![]))); + assert_eq!( + memo.unchanged("uid-1", 8, start + RESYNC), + Some((RESYNC, vec![])) + ); } fn settled(svc: &Service, succeeded: bool) -> bool { From e3473d7571217bacd5fb1b6eed88781c348bfdcd Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 15:16:26 +0200 Subject: [PATCH 46/47] style(controller): format memo record and provisioning test --- src/controller/mod.rs | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/src/controller/mod.rs b/src/controller/mod.rs index 724247d..eb0f68c 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -329,9 +329,13 @@ async fn reconcile_load_balancer( let resync = Duration::from_secs(context.config.resync_interval); if let Some(uid) = uid { - context - .memo - .record(&uid, fingerprint, std::time::Instant::now(), resync, ingress); + context.memo.record( + &uid, + fingerprint, + std::time::Instant::now(), + resync, + ingress, + ); } Ok(ReconcileOutcome::Applied(Action::requeue(resync))) } @@ -640,7 +644,13 @@ mod tests { let error = RobotLBError::LoadBalancerProvisioningTimeout; assert_eq!(warning_reason(&error), "LoadBalancerProvisioningTimeout"); assert_eq!( - error_action(&error, &gate, &backoff, std::time::Instant::now(), "shop/web"), + error_action( + &error, + &gate, + &backoff, + std::time::Instant::now(), + "shop/web" + ), Action::requeue(INGRESS_RETRY_INTERVAL) ); } From 1ced02db681d67863f5cc5574e9042f6a9fd5c25 Mon Sep 17 00:00:00 2001 From: Alexander Gil Casas Date: Tue, 13 Oct 2026 15:16:38 +0200 Subject: [PATCH 47/47] fix(slice): remove redundant test clone and apply rustfmt --- src/controller/slice_changes.rs | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/src/controller/slice_changes.rs b/src/controller/slice_changes.rs index 92d9a2d..34a221b 100644 --- a/src/controller/slice_changes.rs +++ b/src/controller/slice_changes.rs @@ -219,9 +219,8 @@ mod tests { } fn with_owner(mut slice: EndpointSlice, service: Option<&str>) -> EndpointSlice { - slice.metadata.labels = service.map(|name| { - [("kubernetes.io/service-name".to_string(), name.to_string())].into() - }); + slice.metadata.labels = service + .map(|name| [("kubernetes.io/service-name".to_string(), name.to_string())].into()); slice } @@ -234,7 +233,7 @@ mod tests { changes.track(&Event::InitApply(first.clone())); assert!(!changes.track(&Event::::InitDone)); - let second = with_owner(first.clone(), Some("service-b")); + let second = with_owner(first, Some("service-b")); // Main watch may get the event before the metadata watch. assert!(!changes.observe(&second, SliceTargetMode::ReadyEndpointNodes)); assert!(changes.track(&Event::Apply(second.clone()))); @@ -255,7 +254,10 @@ mod tests { assert!(!changes.track(&Event::::InitDone)); changes.track(&Event::::Init); - changes.track(&Event::InitApply(with_owner(first.clone(), Some("service-b")))); + changes.track(&Event::InitApply(with_owner( + first.clone(), + Some("service-b"), + ))); assert!(changes.track(&Event::::InitDone)); assert!(changes.observe( &with_owner(first, Some("service-b")),