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/main.rs b/src/main.rs index edcfa73..3b9a83e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -16,6 +16,7 @@ ) ] +use backoff::ErrorBackoff; use clap::Parser; use config::OperatorConfig; use error::{redact, RobotLBError, RobotLBResult}; @@ -47,6 +48,7 @@ use std::{ time::{Duration, Instant}, }; +pub mod backoff; pub mod config; pub mod consts; pub mod error; @@ -124,6 +126,7 @@ pub struct CurrentContext { pub config: OperatorConfig, pub hcloud_config: HCloudConfig, pub rate_limit: Arc, + pub error_backoff: Arc, } impl CurrentContext { #[must_use] @@ -133,6 +136,7 @@ impl CurrentContext { config, hcloud_config, rate_limit: Arc::default(), + error_backoff: Arc::default(), } } } @@ -147,6 +151,9 @@ pub async fn reconcile_service( context: Arc, ) -> RobotLBResult { let result = sync_service(svc.clone(), context.clone()).await; + if ends_backoff(&result) { + context.error_backoff.forget(&service_key(&svc)); + } if let Err(error) = &result { if publishes_event(error) { report_failure(&svc, &context, error).await; @@ -155,6 +162,16 @@ pub async fn reconcile_service( result } +/// A skipped service may have failed back when it was still a robotlb load balancer. +/// A service waiting at the rate limit gate has not got any closer to succeeding. +const fn ends_backoff(result: &RobotLBResult) -> bool { + matches!(result, Ok(_) | Err(RobotLBError::SkipService)) +} + +fn service_key(svc: &Service) -> String { + format!("{}/{}", svc.namespace().unwrap_or_default(), svc.name_any()) +} + /// Skipped services are every service robotlb does not own. A service waiting at the /// rate limit gate still gets an event each time it wakes up to a closed gate. const fn publishes_event(error: &RobotLBError) -> bool { @@ -607,13 +624,19 @@ async fn clear_ingress_status(svc_api: &kube::Api, svc: &Service) -> Ro /// Handle the error during reconcilation. #[allow(clippy::needless_pass_by_value)] fn on_error(svc: Arc, error: &RobotLBError, context: Arc) -> Action { - let service = format!("{}/{}", svc.namespace().unwrap_or_default(), svc.name_any()); - error_action(error, &context.rate_limit, Instant::now(), &service) + error_action( + error, + &context.rate_limit, + &context.error_backoff, + Instant::now(), + &service_key(&svc), + ) } fn error_action( error: &RobotLBError, rate_limit: &RateLimitGate, + backoff: &ErrorBackoff, now: Instant, service: &str, ) -> Action { @@ -623,7 +646,7 @@ fn error_action( error if error.is_rate_limited() => { Action::requeue(spread(rate_limit.on_rate_limited(now), service)) } - _ => Action::requeue(Duration::from_secs(30)), + _ => Action::requeue(backoff.on_failure(service, now)), } } @@ -848,7 +871,13 @@ mod tests { let wait = std::time::Duration::from_secs(42); let error = crate::error::RobotLBError::RateLimited(wait); assert_eq!( - error_action(&error, &gate, std::time::Instant::now(), "shop/web"), + error_action( + &error, + &gate, + &crate::backoff::ErrorBackoff::default(), + std::time::Instant::now(), + "shop/web" + ), kube::runtime::controller::Action::requeue(crate::rate_limit::spread(wait, "shop/web")) ); } @@ -867,7 +896,13 @@ mod tests { }, )); assert_eq!( - error_action(&error, &gate, now, "shop/web"), + error_action( + &error, + &gate, + &crate::backoff::ErrorBackoff::default(), + now, + "shop/web" + ), kube::runtime::controller::Action::requeue(crate::rate_limit::spread( std::time::Duration::from_secs(60), "shop/web" @@ -877,15 +912,57 @@ mod tests { } #[test] - fn other_errors_retry_in_30_seconds() { + fn other_errors_back_off_per_service() { let gate = crate::rate_limit::RateLimitGate::default(); + let backoff = crate::backoff::ErrorBackoff::default(); + let now = std::time::Instant::now(); + let error = crate::error::RobotLBError::HCloudError("boom".to_string()); + let retry = |service| error_action(&error, &gate, &backoff, now, service); + let after = + |secs| kube::runtime::controller::Action::requeue(std::time::Duration::from_secs(secs)); + assert_eq!(retry("shop/web"), after(5)); + assert_eq!(retry("shop/web"), after(10)); + assert_eq!(retry("shop/api"), after(5)); + } + + #[test] + fn rate_limits_do_not_lengthen_the_error_backoff() { + let gate = crate::rate_limit::RateLimitGate::default(); + let backoff = crate::backoff::ErrorBackoff::default(); + let now = std::time::Instant::now(); + let gated = crate::error::RobotLBError::RateLimited(std::time::Duration::from_secs(42)); + error_action(&gated, &gate, &backoff, now, "shop/web"); + let rejected = crate::error::RobotLBError::from(hcloud::apis::Error::< + hcloud::apis::load_balancers_api::AddTargetError, + >::ResponseError( + hcloud::apis::ResponseContent { + status: 429_u16.try_into().unwrap(), + content: String::new(), + entity: None, + }, + )); + error_action(&rejected, &gate, &backoff, now, "shop/web"); let error = crate::error::RobotLBError::HCloudError("boom".to_string()); assert_eq!( - error_action(&error, &gate, std::time::Instant::now(), "shop/web"), - kube::runtime::controller::Action::requeue(std::time::Duration::from_secs(30)) + error_action(&error, &gate, &backoff, now, "shop/web"), + kube::runtime::controller::Action::requeue(std::time::Duration::from_secs(5)) ); } + #[test] + fn success_and_skips_end_the_backoff_but_a_closed_gate_does_not() { + use crate::error::RobotLBError; + use kube::runtime::controller::Action; + assert!(super::ends_backoff(&Ok(Action::await_change()))); + assert!(super::ends_backoff(&Err(RobotLBError::SkipService))); + assert!(!super::ends_backoff(&Err(RobotLBError::RateLimited( + std::time::Duration::from_secs(1) + )))); + assert!(!super::ends_backoff(&Err(RobotLBError::HCloudError( + "boom".to_string() + )))); + } + #[test] fn every_failure_except_a_skip_publishes_an_event() { use crate::error::RobotLBError;