diff --git a/src/controller/event_throttle.rs b/src/controller/event_throttle.rs new file mode 100644 index 0000000..5d4137d --- /dev/null +++ b/src/controller/event_throttle.rs @@ -0,0 +1,124 @@ +//! Bound repeated Kubernetes Warning Events during persistent reconciliation failures. +//! +//! Reconciliation and error logging continue on every attempt. Only identical +//! Event publications are throttled, so a changed error remains visible. + +use std::{ + collections::HashMap, + sync::{Mutex, PoisonError}, + time::{Duration, Instant}, +}; + +const REPEAT_INTERVAL: Duration = Duration::from_secs(300); + +/// Identifies the exact error shown on a particular Service. +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +pub struct EventKey { + pub service_uid: String, + pub reason: String, + pub note: String, +} + +/// Records recent Event publication attempts for one controller instance. +#[derive(Debug, Default)] +pub struct EventThrottle { + recent: Mutex>, +} + +impl EventThrottle { + /// Reserve an Event. Concurrent reconciliations cannot reserve the same key. + /// Expired entries are removed to avoid retaining deleted Services indefinitely. + pub fn reserve(&self, key: &EventKey, now: Instant) -> bool { + let mut recent = self.recent.lock().unwrap_or_else(PoisonError::into_inner); + recent.retain(|_, last| now.saturating_duration_since(*last) < REPEAT_INTERVAL); + if recent.contains_key(key) { + return false; + } + recent.insert(key.clone(), now); + true + } + + /// Permit another attempt if publishing failed. Do not clear a newer reservation. + pub fn release(&self, key: &EventKey, reserved_at: Instant) { + let mut recent = self.recent.lock().unwrap_or_else(PoisonError::into_inner); + if recent.get(key) == Some(&reserved_at) { + recent.remove(key); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn key(uid: &str, reason: &str, note: &str) -> EventKey { + EventKey { + service_uid: uid.to_string(), + reason: reason.to_string(), + note: note.to_string(), + } + } + + #[test] + fn identical_event_is_suppressed_for_five_minutes_then_resent() { + let throttle = EventThrottle::default(); + let first = Instant::now(); + let event = key("service-uid", "SyncLoadBalancerFailed", "Hetzner 503"); + assert!(throttle.reserve(&event, first)); + assert!(!throttle.reserve(&event, first + Duration::from_secs(299))); + assert!(throttle.reserve(&event, first + REPEAT_INTERVAL)); + assert!(!throttle.reserve(&event, first + REPEAT_INTERVAL)); + } + + #[test] + fn different_service_reason_or_message_is_not_suppressed() { + let throttle = EventThrottle::default(); + let now = Instant::now(); + assert!(throttle.reserve(&key("uid-a", "SyncLoadBalancerFailed", "503"), now)); + assert!(throttle.reserve(&key("uid-b", "SyncLoadBalancerFailed", "503"), now)); + assert!(throttle.reserve( + &key("uid-a", "InvalidLoadBalancerConfiguration", "503"), + now + )); + assert!(throttle.reserve(&key("uid-a", "SyncLoadBalancerFailed", "502"), now)); + } + + #[test] + fn failed_event_publication_does_not_suppress_retry() { + let throttle = EventThrottle::default(); + let now = Instant::now(); + let event = key("uid", "SyncLoadBalancerFailed", "503"); + assert!(throttle.reserve(&event, now)); + throttle.release(&event, now); + assert!(throttle.reserve(&event, now + Duration::from_secs(30))); + } + + #[test] + fn stale_failure_cannot_cancel_a_new_reservation() { + let throttle = EventThrottle::default(); + let now = Instant::now(); + let event = key("uid", "SyncLoadBalancerFailed", "503"); + assert!(throttle.reserve(&event, now)); + assert!(throttle.reserve(&event, now + REPEAT_INTERVAL)); + throttle.release(&event, now); + assert!(!throttle.reserve(&event, now + REPEAT_INTERVAL)); + } + + #[test] + fn only_one_concurrent_reconcile_may_publish_an_event() { + let throttle = std::sync::Arc::new(EventThrottle::default()); + let now = Instant::now(); + let event = key("uid", "SyncLoadBalancerFailed", "503"); + let reservations: usize = std::thread::scope(|scope| { + // Spawn all workers before joining to exercise concurrent reservations. + let mut tasks = Vec::new(); + for _ in 0..8 { + let throttle = throttle.clone(); + let event = event.clone(); + tasks.push(scope.spawn(move || usize::from(throttle.reserve(&event, now)))); + } + tasks.into_iter().map(|task| task.join().unwrap()).sum() + }); + assert_eq!(reservations, 1); + } +} diff --git a/src/controller/mod.rs b/src/controller/mod.rs index 8192583..317203c 100644 --- a/src/controller/mod.rs +++ b/src/controller/mod.rs @@ -3,9 +3,13 @@ //! This module contains the reconciliation logic for Kubernetes services //! with the `robotlb` load balancer class. +mod event_throttle; mod nodes; mod status; +use event_throttle::EventKey; +pub(crate) use event_throttle::EventThrottle; + use std::{ sync::{Arc, atomic::Ordering}, time::Duration, @@ -92,7 +96,20 @@ pub async fn reconcile_service( context: Arc, ) -> RobotLBResult { let mut timer = ReconcileTimer::new(context.metrics.clone()); + let result = reconcile_service_inner(svc.clone(), context.clone()).await; + if let Err(error) = &result + && !matches!(error, RobotLBError::SkipService) + { + timer.set_failed(); + publish_reconciliation_warning(&svc, &context, error).await; + } + result +} +async fn reconcile_service_inner( + svc: Arc, + context: Arc, +) -> RobotLBResult { if !context.is_leader.load(Ordering::Relaxed) { return Err(RobotLBError::SkipService); } @@ -116,43 +133,38 @@ pub async fn reconcile_service( ensure_service_is_supported(&svc)?; - tracing::info!("Starting service reconcilation"); + tracing::info!("Starting service reconciliation"); - let lb = match LoadBalancer::try_from_svc(&svc, &context) { - Ok(lb) => lb, - Err(e) => { - timer.set_failed(); - return Err(e); - } - }; + let lb = LoadBalancer::try_from_svc(&svc, &context)?; - // Add finalizer if it's not there yet. - if !finalizers::check(&svc) - && let Err(e) = finalizers::add(context.client.clone(), &svc).await - { - timer.set_failed(); - return Err(e); + if !finalizers::check(&svc) { + finalizers::add(context.client.clone(), &svc).await?; } - // Based on the service type, we will reconcile the load balancer. - match reconcile_load_balancer(lb, svc.clone(), context.clone()).await { - Ok(action) => Ok(action), - Err(e) => { - timer.set_failed(); - if matches!(e, RobotLBError::InvalidLoadBalancerConfiguration(_)) { - publish_invalid_configuration_warning(&svc, &context, &e).await; - } - Err(e) - } - } + reconcile_load_balancer(lb, svc, context).await } -/// Emit a Warning Event while keeping reconciliation errors independent of Event RBAC. -async fn publish_invalid_configuration_warning( +/// 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( svc: &Service, context: &CurrentContext, error: &RobotLBError, ) { + let reason = warning_reason(error); + let note = event_note(error, &context.config.hcloud_token); + let event_key = EventKey { + service_uid: svc.metadata.uid.clone().unwrap_or_else(|| { + format!("{}/{}", svc.namespace().unwrap_or_default(), svc.name_any()) + }), + reason: reason.to_string(), + note: note.clone(), + }; + let reserved_at = std::time::Instant::now(); + if !context.event_throttle.reserve(&event_key, reserved_at) { + return; + } + let recorder = Recorder::new( context.client.clone(), Reporter { @@ -163,18 +175,41 @@ async fn publish_invalid_configuration_warning( let event = Event { type_: EventType::Warning, action: "Reconcile".to_string(), - reason: "InvalidLoadBalancerConfiguration".to_string(), - note: Some(error.to_string()), + reason: reason.to_string(), + note: Some(note), secondary: None, }; if let Err(publish_error) = recorder.publish(&event, &svc.object_ref(&())).await { + context.event_throttle.release(&event_key, reserved_at); tracing::warn!( error = %publish_error, - "Cannot publish invalid configuration Warning Event" + "Cannot publish reconciliation Warning Event" ); } } +const fn warning_reason(error: &RobotLBError) -> &'static str { + if matches!(error, RobotLBError::InvalidLoadBalancerConfiguration(_)) { + "InvalidLoadBalancerConfiguration" + } else { + "SyncLoadBalancerFailed" + } +} + +/// Kubernetes Events cap notes at 1024 bytes, not characters. +fn event_note(error: &RobotLBError, token: &str) -> String { + const MAX_NOTE_BYTES: usize = 1024; + let mut note = crate::error::redact(&error.to_string(), token); + if note.len() > MAX_NOTE_BYTES { + let mut end = MAX_NOTE_BYTES; + while !note.is_char_boundary(end) { + end -= 1; + } + note.truncate(end); + } + note +} + /// Reconcile the `LoadBalancer` type of service. /// /// This function will find the nodes based on the node selector @@ -264,6 +299,7 @@ pub async fn run( shutdown: tokio_util::sync::CancellationToken, ) { let controller_shutdown = shutdown.clone(); + let hcloud_token = context.config.hcloud_token.clone(); Controller::new( kube::Api::::all(kube_client), watcher::Config::default(), @@ -275,20 +311,34 @@ pub async fn run( ) .run(reconcile_service, on_error, context) .take_until(controller_shutdown.cancelled()) - .for_each(|reconcilation_result| async move { - match reconcilation_result { - Ok((service, _action)) => { - tracing::info!("Reconcilation of a service {} was successful", service.name); - } - Err(err) => match err { - kube::runtime::controller::Error::ReconcilerFailed( + .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, _, - ) => {} - _ => { - tracing::error!("Error reconciling service: {:#?}", err); + )) => {} + 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; @@ -323,6 +373,35 @@ mod tests { assert!(ensure_service_is_supported(&svc).is_ok()); } + #[test] + fn all_errors_are_reported_except_skipped_services() { + assert_eq!( + warning_reason(&RobotLBError::InvalidLoadBalancerConfiguration( + "too many targets".to_string() + )), + "InvalidLoadBalancerConfiguration" + ); + assert_eq!( + warning_reason(&RobotLBError::HCloudError("error".to_string())), + "SyncLoadBalancerFailed" + ); + } + + #[test] + fn warning_notes_redact_hetzner_token_and_fit_event_limit() { + let token = "0123456789abcdef"; + let error = RobotLBError::HCloudError(format!( + "invalid token {} {}", + &token[..12], + "é".repeat(600) + )); + let note = event_note(&error, token); + assert!(note.contains("[REDACTED]")); + assert!(!note.contains(&token[..12])); + assert!(note.len() <= 1024); + assert!(note.is_char_boundary(note.len())); + } + #[test] fn on_error_awaits_change_for_skipped_service() { let action = action_for_error(&RobotLBError::SkipService); diff --git a/src/error.rs b/src/error.rs index 3dfe619..66b086e 100644 --- a/src/error.rs +++ b/src/error.rs @@ -13,6 +13,23 @@ fn is_rate_limit(err: &hcloud::apis::Error) -> bool { matches!(err, hcloud::apis::Error::ResponseError(response) if response.status.as_u16() == 429) } +/// Remove the Hetzner API token (including a quoted prefix) from diagnostics. +/// +/// Hetzner error bodies can contain a token prefix, not only the complete token. +/// Eight bytes is the shortest prefix we treat as secret so short unrelated +/// text does not get unnecessarily replaced. +#[must_use] +pub fn redact(message: &str, token: &str) -> String { + const MIN_PREFIX: usize = 8; + let mut safe = message.to_string(); + for len in (MIN_PREFIX..=token.len()).rev() { + if let Some(prefix) = token.get(..len) { + safe = safe.replace(prefix, "[REDACTED]"); + } + } + safe +} + /// All possible errors that can occur in the robotlb operator. #[derive(Debug, Error)] pub enum RobotLBError { @@ -181,3 +198,37 @@ impl RobotLBError { .is_some_and(|code| code == "target_already_defined") } } + +#[cfg(test)] +mod redaction_tests { + use super::redact; + + const TOKEN: &str = "abcdefghijklmnopqrstuvwxyz0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ01"; + + #[test] + fn redacts_complete_token_and_quoted_prefix() { + assert_eq!( + redact(&format!("bad token {TOKEN}"), TOKEN), + "bad token [REDACTED]" + ); + assert_eq!( + redact(&format!("invalid token {}", &TOKEN[..24]), TOKEN), + "invalid token [REDACTED]" + ); + } + + #[test] + fn leaves_unrelated_messages_and_short_prefixes_intact() { + assert_eq!(redact("Hetzner 500", TOKEN), "Hetzner 500"); + assert_eq!(redact("short abcdefg", TOKEN), "short abcdefg"); + assert_eq!(redact("empty token", ""), "empty token"); + } + + #[test] + fn redacts_non_ascii_surrounding_text() { + assert_eq!( + redact(&format!("échec {} 🦀", &TOKEN[..8]), TOKEN), + "échec [REDACTED] 🦀" + ); + } +} diff --git a/src/lb/mod.rs b/src/lb/mod.rs index 9414447..68926d8 100644 --- a/src/lb/mod.rs +++ b/src/lb/mod.rs @@ -500,7 +500,7 @@ impl LoadBalancer { else { return Ok(()); }; - api::delete_load_balancer(&self.hcloud_config, lb.id).await + retry_temporary(|| api::delete_load_balancer(&self.hcloud_config, lb.id)).await } /// Release without parsing annotations: malformed config cannot block deletion. @@ -516,7 +516,7 @@ impl LoadBalancer { let Some(lb) = api::get_owned_load_balancer(&context.hcloud_config, uid).await? else { return Ok(()); }; - api::delete_load_balancer(&context.hcloud_config, lb.id).await + retry_temporary(|| api::delete_load_balancer(&context.hcloud_config, lb.id)).await } /// Existing resources must have the matching Service UID label. @@ -683,6 +683,48 @@ mod tests { use hcloud::apis::configuration::Configuration as HcloudConfig; use std::sync::Mutex; + fn delete_error(status: u16) -> RobotLBError { + RobotLBError::HcloudLBDeleteError(hcloud::apis::Error::ResponseError( + hcloud::apis::ResponseContent { + status: status.try_into().expect("valid HTTP status"), + content: String::new(), + entity: None, + }, + )) + } + + #[tokio::test] + async fn delete_retries_locked_then_succeeds() { + let attempts = std::sync::atomic::AtomicUsize::new(0); + retry_temporary(|| { + let attempt = attempts.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + async move { + if attempt == 0 { + Err(delete_error(423)) + } else { + Ok(()) + } + } + }) + .await + .expect("second deletion succeeds"); + assert_eq!(attempts.load(std::sync::atomic::Ordering::Relaxed), 2); + } + + #[tokio::test] + async fn delete_does_not_retry_permanent_or_rate_limit_errors() { + for status in [403, 429] { + let attempts = std::sync::atomic::AtomicUsize::new(0); + let result: RobotLBResult<()> = retry_temporary(|| { + attempts.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + async move { Err(delete_error(status)) } + }) + .await; + assert!(result.is_err()); + assert_eq!(attempts.load(std::sync::atomic::Ordering::Relaxed), 1); + } + } + #[test] fn capacity_validation_rejects_oversized_configuration() { let mut lb = test_load_balancer(); diff --git a/src/main.rs b/src/main.rs index 85f272d..e0ea12e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -72,6 +72,8 @@ pub struct CurrentContext { pub metrics: Arc, /// Hetzner project-wide API throttle shared across all Services. pub rate_limit: Arc, + /// Suppress repeated identical Kubernetes Warning Events. + pub(crate) event_throttle: Arc, } impl CurrentContext { @@ -91,6 +93,7 @@ impl CurrentContext { is_leader, metrics, rate_limit: Arc::default(), + event_throttle: Arc::default(), } } }