fix: redact Hetzner errors and retry transient balancer deletion

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<HashMap<EventKey, Instant>>,
+}
+
+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);
+    }
+}

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<CurrentContext>,
 ) -> RobotLBResult<Action> {
     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<Service>,
+    context: Arc<CurrentContext>,
+) -> RobotLBResult<Action> {
     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::<Service>::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);