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: diff --git a/Dockerfile b/Dockerfile index 9ef2dca..c8fb94d 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 @@ -8,11 +8,13 @@ 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 -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 diff --git a/README.md b/README.md index c5dbfd5..9fede5d 100644 --- a/README.md +++ b/README.md @@ -231,3 +231,43 @@ 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. 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 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. + +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. +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 +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. 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: [] 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); + } +} diff --git a/src/config.rs b/src/config.rs index 858f6da..da20379 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)] @@ -23,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, @@ -140,8 +153,10 @@ 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.resync_interval, 300); assert_eq!(config.default_lb_retries, 3); assert_eq!(config.default_lb_timeout, 10); assert_eq!(config.default_lb_interval, 15); @@ -167,6 +182,10 @@ mod tests { "robotlb", "--hcloud-token", "token", + "--cluster-name", + "prod", + "--resync-interval", + "120", "--default-network", "prod-net", "--default-lb-retries", @@ -204,6 +223,8 @@ 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); assert_eq!(config.default_lb_timeout, 7); diff --git a/src/controller/mod.rs b/src/controller/mod.rs index 317203c..eb0f68c 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -4,23 +4,27 @@ //! with the `robotlb` load balancer class. 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}, time::Duration, }; +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, @@ -41,8 +45,6 @@ pub(crate) use nodes::{ }; pub(crate) use status::{build_ingress, patch_ingress_status}; -const SUCCESS_REQUEUE_SECS: u64 = 180; - /// Check if a service is supported by robotlb. /// /// Returns an error if the service is not a `LoadBalancer` type or @@ -81,6 +83,32 @@ 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, + } + } +} + +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, + Ok(ReconcileOutcome::Applied(_)) | Err(RobotLBError::SkipService) + ) +} + /// Reconcile the service. /// /// This function is called by the controller for each service. @@ -97,19 +125,25 @@ 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, matches!(result, Ok(ReconcileOutcome::Applied(_)))); + if should_reset_failure_backoff(&result) { + context.error_backoff.forget(&service_key(&svc)); + } if let Err(error) = &result && !matches!(error, RobotLBError::SkipService) { 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); } @@ -118,7 +152,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 @@ -128,7 +162,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)?; @@ -144,6 +178,11 @@ 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( @@ -189,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", } } @@ -218,11 +257,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?; @@ -239,6 +278,19 @@ 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, 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))); + } + let hcloud_lb = lb.reconcile().await?; let ingress = build_ingress( @@ -247,32 +299,85 @@ pub async fn reconcile_load_balancer( lb.proxy_mode, lb.public_interface, ); - patch_ingress_status(&svc, &context, ingress).await?; + // 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) { + 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( + INGRESS_RETRY_INTERVAL, + ))); + } + 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, + ingress, + ); + } + Ok(ReconcileOutcome::Applied(Action::requeue(resync))) +} - Ok(Action::requeue(Duration::from_secs(SUCCESS_REQUEUE_SECS))) +const fn should_retry_missing_ingress( + ingress: &[k8s_openapi::api::core::v1::LoadBalancerIngress], +) -> bool { + ingress.is_empty() } -/// 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(_: 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); - } - action_for_error(error) +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), + ) } -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)) + // 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!( + ?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)), } } @@ -292,6 +397,85 @@ 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. +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 when target-relevant Nodes change or an observed +/// `EndpointSlice` disappears or changes its Service label. +/// Ordinary endpoint 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(node_client), + watcher::Config::default(), + ) + .default_backoff() + ); + while let Some(event) = events.next().await { + match event { + Ok(event) => { + if changes.observe(&event) && node_sender.unbounded_send(()).is_err() { + return; + } + } + Err(error) => tracing::warn!(%error, "Node watch failed"), + } + } + }); + + // 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 +} + /// Create and run the controller. pub async fn run( kube_client: kube::Client, @@ -300,48 +484,69 @@ 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(), - ) - .watches( - kube::Api::::all(context.client.clone()), - watcher::Config::default(), - |endpoint_slice| map_endpoint_slice_to_service(&endpoint_slice), - ) - .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" - ); + ); + 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(), + 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)] @@ -403,24 +608,105 @@ mod tests { } #[test] - fn on_error_awaits_change_for_skipped_service() { - let action = action_for_error(&RobotLBError::SkipService); - assert_eq!(action, Action::await_change()); + 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 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 + // 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 invalid_configuration_requeues_less_frequently() { + 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!( - action_for_error(&RobotLBError::InvalidLoadBalancerConfiguration( - "test".to_string() - )), - Action::requeue(Duration::from_secs(180)) + 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 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() + ); } } diff --git a/src/controller/node_changes.rs b/src/controller/node_changes.rs new file mode 100644 index 0000000..d690728 --- /dev/null +++ b/src/controller/node_changes.rs @@ -0,0 +1,146 @@ +//! Detect node changes that actually affect Hetzner target selection. +//! Kubelet heartbeats must not trigger reconciling every `LoadBalancer` Service. + +use std::{ + collections::{HashMap, hash_map::DefaultHasher}, + hash::{Hash, Hasher}, +}; + +use k8s_openapi::api::core::v1::Node; +use kube::{ResourceExt, runtime::watcher::Event}; + +#[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")))); + } +} diff --git a/src/controller/slice_changes.rs b/src/controller/slice_changes.rs new file mode 100644 index 0000000..34a221b --- /dev/null +++ b/src/controller/slice_changes.rs @@ -0,0 +1,276 @@ +//! 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 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 + .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: HashMap>, + relisted: HashMap>, +} + +/// 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), owner(slice)); + false + } + Event::InitDone => { + let live = std::mem::take(&mut state.relisted); + let removed = state + .observed + .iter() + .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 || !changed.is_empty() + } + Event::Apply(slice) => { + 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); + 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))); + } + + 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, 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(); + let initial = slice(true, "node-a"); + assert!(changes.observe(&initial, SliceTargetMode::PodNodes)); + changes.forget(&initial); + assert!(changes.observe(&initial, SliceTargetMode::PodNodes)); + } +} diff --git a/src/error.rs b/src/error.rs index 66b086e..136c8c7 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,9 @@ pub enum RobotLBError { HCloudError(String), #[error("Invalid load balancer configuration: {0}")] InvalidLoadBalancerConfiguration(String), - #[error("Cannot list Hetzner load balancer types: {0}")] + #[error("Load balancer has no ingress address after the provisioning deadline")] + LoadBalancerProvisioningTimeout, + #[error("Cannot list Hetzner load balancer types: {}", describe_hcloud(.0))] HCloudListLoadBalancerTypesError( #[from] hcloud::apis::Error, @@ -60,62 +99,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, ), @@ -224,6 +263,33 @@ 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!( diff --git a/src/lb/config.rs b/src/lb/config.rs index 289785b..c721afc 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,47 @@ 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. +/// +/// # 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()) + && 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,8 +192,10 @@ 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, + resync_interval: 300, default_lb_retries: 3, default_lb_timeout: 10, default_lb_interval: 15, @@ -209,6 +252,58 @@ 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(); diff --git a/src/lb/mod.rs b/src/lb/mod.rs index 68926d8..91dd882 100644 --- a/src/lb/mod.rs +++ b/src/lb/mod.rs @@ -30,6 +30,7 @@ use crate::{ }; pub(crate) use api::{HcloudLoadBalancerApi, LiveHcloudLoadBalancerApi}; +pub use config::parse_cluster_name; pub(crate) use config::parse_load_balancer_config; use types::{ServiceReconcileAction, TargetReconcileAction, normalize_ip, service_matches_desired}; 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(), } } diff --git a/src/memo.rs b/src/memo.rs new file mode 100644 index 0000000..8a6ff31 --- /dev/null +++ b/src/memo.rs @@ -0,0 +1,399 @@ +use k8s_openapi::api::core::v1::{LoadBalancerIngress, 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(Clone)] +struct SuccessfulReconcile { + fingerprint: u64, + until: Instant, + ingress: Vec, +} + +#[derive(Default)] +pub struct ReconcileMemo { + 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<(Duration, Vec)> { + let snapshot = self + .entries + .lock() + .unwrap_or_else(PoisonError::into_inner) + .get(uid)? + .clone(); + if snapshot.fingerprint != fingerprint { + return None; + } + 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`. + 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(), + SuccessfulReconcile { + fingerprint, + until: 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); + } + let overdue = now.saturating_duration_since(entry.1) >= deadline; + drop(pending); + overdue + } + + 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 + /// 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 let Some(uid) = svc.uid() { + if !(succeeded && serves) { + self.forget(&uid); + } + if !serves { + self.clear_pending(&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::{ReconcileMemo, fingerprint}; + use crate::consts; + use k8s_openapi::{ + api::core::v1::{ + LoadBalancerIngress, LoadBalancerStatus, Service, ServicePort, ServiceSpec, + ServiceStatus, + }, + apimachinery::pkg::apis::meta::v1::ObjectMeta, + }; + 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, vec![]); + assert_eq!( + memo.unchanged("uid-1", 7, start + Duration::from_secs(100)), + Some((Duration::from_secs(200), vec![])) + ); + 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, vec![]); + assert_eq!( + memo.unchanged("uid-1", 7, start + Duration::from_secs(10)), + Some((Duration::from_secs(20), vec![])) + ); + 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, 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, vec![]); + 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, 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, vec![]); + 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(k8s_openapi::serde_json::from_str(r#""2026-01-01T00:00:00Z""#).unwrap()); + 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_ANNOTATION.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())); + } +}