diff --git a/README.md b/README.md index 02c9aae..7f3b221 100644 --- a/README.md +++ b/README.md @@ -46,10 +46,18 @@ Setting `ROBOTLB_DYNAMIC_NODE_SELECTOR` to `false` replaces both with the node s A balancer type caps how many targets it holds: `lb11`, the default type, holds 25. When more nodes are selected than the type holds, the extra ones are dropped in a stable order and a warning names the limit. Pick a bigger type through `ROBOTLB_DEFAULT_LB_TYPE` or the `robotlb/balancer-type` annotation to use the whole cluster. -Every port of the service needs an allocated `nodePort`. A Hetzner load balancer forwards traffic to the IP of a node, so a port is reachable only through its `nodePort`: ports without one are skipped, and `allocateLoadBalancerNodePorts: false` is not supported. When no port of a service can be exposed, no balancer is created for it, and a service that already advertises an external IP loses it. +robotlb deletes the balancer when the service is deleted or stops being a `LoadBalancer`, and finds it by the name the service has at that moment. When the `robotlb/balancer` annotation was changed or removed before that, the balancer under the old name stays in the project, and the balancer under the current name is deleted, even when another service uses it. Services that share a balancer name share one balancer: releasing either of them deletes it. Without the `robotlb/balancer` annotation the name is the service name without its namespace. + +Every port of the service needs an allocated `nodePort`. A Hetzner load balancer forwards traffic to the IP of a node, so a port is reachable only through its `nodePort`: ports without one are skipped, and `allocateLoadBalancerNodePorts: false` is not supported. When no port of a service can be exposed, no balancer is created for it, an existing balancer and the service's external IP are kept as they are, and a warning event on the service reports the problem. > Earlier releases treated every service as if it had the `Local` policy. Services that leave `externalTrafficPolicy` unset therefore get the full node list on upgrade, which changes the targets of their existing balancers. +> Earlier releases kept the balancer of a service whose type was changed from `LoadBalancer`, or that moved to another load balancer class. Such a service still carries the `robotlb/finalizer` finalizer, and its Hetzner balancer is deleted on the first start after the upgrade. Its balancer is found by name. Without the `robotlb/balancer` annotation the name is the service name without its namespace, so a `LoadBalancer` service with the same name in another namespace may be using that balancer. List the affected services before upgrading and check their balancers: +> +> ```bash +> kubectl get services --all-namespaces --output json | jq --raw-output '.items[] | select(((.metadata.finalizers // []) | index("robotlb/finalizer")) and (.spec.type != "LoadBalancer" or (.spec.loadBalancerClass // "robotlb") != "robotlb")) | "\(.metadata.namespace)/\(.metadata.name)"' +> ``` + ## Configuration diff --git a/helm/values.yaml b/helm/values.yaml index 2524ad9..865062a 100644 --- a/helm/values.yaml +++ b/helm/values.yaml @@ -28,7 +28,7 @@ serviceAccount: # If not set and create is true, a name is generated using the fullname template name: "" # This is a list of cluster permissions to apply to the service account. - # By default it grants all permissions. + # The default is what robotlb needs; a custom list replaces it entirely. permissions: - apiGroups: [""] resources: [services, services/status] @@ -39,6 +39,9 @@ serviceAccount: - apiGroups: [discovery.k8s.io] resources: [endpointslices] verbs: [get, list, watch] + - apiGroups: [events.k8s.io] + resources: [events] + verbs: [create] podAnnotations: {} podLabels: {} diff --git a/src/error.rs b/src/error.rs index dae2e3f..573719b 100644 --- a/src/error.rs +++ b/src/error.rs @@ -22,65 +22,234 @@ pub enum RobotLBError { UnknownLBAlgorithm, #[error("Cannot get target nodes, because the service has no selector")] ServiceWithoutSelector, + #[error( + "No TCP port of the service has a nodePort, so the load balancer has nothing to forward" + )] + NoExposablePorts, + #[error("Hetzner Cloud API rate limit reached, the pause ends in {}s", .0.as_millis().div_ceil(1000))] + RateLimited(std::time::Duration), // HCloud API errors - #[error("Cannot attach load balancer to a network. Reason: {0}")] + #[error("Cannot attach load balancer to a network. Reason: {}", describe(.0))] HCloudLBAttachToNetworkError( #[from] hcloud::apis::Error, ), - #[error("Cannot detach load balancer from network. Reason: {0}")] + #[error("Cannot detach load balancer from network. Reason: {}", describe(.0))] HcloudLBDetachFromNetworkError( #[from] hcloud::apis::Error, ), - #[error("Cannot add load balancer target. Reason: {0}")] + #[error("Cannot add load balancer target. Reason: {}", describe(.0))] HcloudLBAddTargetError( #[from] hcloud::apis::Error, ), - #[error("Cannot remove load balancer target. Reason: {0}")] + #[error("Cannot remove load balancer target. Reason: {}", describe(.0))] HcloudLBRemoveTargetError( #[from] hcloud::apis::Error, ), - #[error("Cannot add service to load balancer. Reason: {0}")] + #[error("Cannot add service to load balancer. Reason: {}", describe(.0))] HcloudLBAddServiceError( #[from] hcloud::apis::Error, ), - #[error("Cannot remove service from load balancer. Reason: {0}")] + #[error("Cannot remove service from load balancer. Reason: {}", describe(.0))] HcloudLBRemoveServiceError( #[from] hcloud::apis::Error, ), - #[error("Cannot create load balancer. Reason: {0}")] + #[error("Cannot create load balancer. Reason: {}", describe(.0))] HcloudLBCreateError( #[from] hcloud::apis::Error, ), - #[error("Cannot delete load balancer. Reason: {0}")] + #[error("Cannot delete load balancer. Reason: {}", describe(.0))] HcloudLBDeleteError( #[from] hcloud::apis::Error, ), - #[error("Cannot get load balancer. Reason: {0}")] + #[error("Cannot get load balancer. Reason: {}", describe(.0))] HcloudLBGetError( #[from] hcloud::apis::Error, ), - #[error("Cannot update service. Reason: {0}")] + #[error("Cannot update service. Reason: {}", describe(.0))] HcloudLBUpdateServiceError( #[from] hcloud::apis::Error, ), - #[error("Cannot change type of load balancer. Reason: {0}")] + #[error("Cannot change type of load balancer. Reason: {}", describe(.0))] HcloudLBChangeType( #[from] hcloud::apis::Error, ), - #[error("Cannot change algorithm of load balancer. Reason: {0}")] + #[error("Cannot change algorithm of load balancer. Reason: {}", describe(.0))] HcloudLBChangeAlgorithm( #[from] hcloud::apis::Error, ), - #[error("Cannot list networks. Reason: {0}")] + #[error("Cannot list networks. Reason: {}", describe(.0))] HcloudListNetworksError( #[from] hcloud::apis::Error, ), - #[error("Cannot list load balancers. Reason: {0}")] + #[error("Cannot list load balancers. Reason: {}", describe(.0))] HcloudListLoadBalancersError( #[from] hcloud::apis::Error, ), } + +impl RobotLBError { + /// Whether Hetzner rejected the call because the project ran out of API requests. + #[must_use] + pub fn is_rate_limited(&self) -> bool { + // No wildcard arm: a new variant must be sorted into one of the two groups. + match self { + Self::HCloudLBAttachToNetworkError(error) => is_rate_limit_response(error), + Self::HcloudLBDetachFromNetworkError(error) => is_rate_limit_response(error), + Self::HcloudLBAddTargetError(error) => is_rate_limit_response(error), + Self::HcloudLBRemoveTargetError(error) => is_rate_limit_response(error), + Self::HcloudLBAddServiceError(error) => is_rate_limit_response(error), + Self::HcloudLBRemoveServiceError(error) => is_rate_limit_response(error), + Self::HcloudLBCreateError(error) => is_rate_limit_response(error), + Self::HcloudLBDeleteError(error) => is_rate_limit_response(error), + Self::HcloudLBGetError(error) => is_rate_limit_response(error), + Self::HcloudLBUpdateServiceError(error) => is_rate_limit_response(error), + Self::HcloudLBChangeType(error) => is_rate_limit_response(error), + Self::HcloudLBChangeAlgorithm(error) => is_rate_limit_response(error), + Self::HcloudListNetworksError(error) => is_rate_limit_response(error), + Self::HcloudListLoadBalancersError(error) => is_rate_limit_response(error), + Self::InvalidNodeFilter(_) + | Self::UnsupportedServiceType + | Self::SkipService + | Self::PaseIntError(_) + | Self::PaseBoolError(_) + | Self::HCloudError(_) + | Self::KubeError(_) + | Self::UnknownLBAlgorithm + | Self::ServiceWithoutSelector + | Self::NoExposablePorts + | Self::RateLimited(_) => false, + } + } +} + +/// Whether Hetzner answered 429 because the project ran out of API requests. +#[must_use] +pub fn is_rate_limit_response(error: &hcloud::apis::Error) -> bool { + matches!(error, hcloud::apis::Error::ResponseError(response) if response.status.as_u16() == 429) +} + +/// One line with the HTTP status and the error Hetzner reported, if the body carries one. +#[must_use] +pub fn describe(error: &hcloud::apis::Error) -> String { + let hcloud::apis::Error::ResponseError(response) = error else { + return error.to_string(); + }; + let body = + k8s_openapi::serde_json::from_str::(&response.content).ok(); + let reported = body.as_ref().and_then(|body| { + let error = body.get("error")?; + Some(format!( + "{}: {}", + error.get("code")?.as_str()?, + error.get("message")?.as_str()? + )) + }); + reported.map_or_else( + || response.status.to_string(), + |reported| { + let one_line = reported.split_whitespace().collect::>().join(" "); + format!("{}: {one_line}", response.status) + }, + ) +} + +/// Replace the API token, or any prefix of it long enough to identify it, with a marker. +/// Hetzner quotes the start of the token in some error messages. +#[must_use] +pub fn redact(message: &str, token: &str) -> String { + const SHORTEST_PREFIX: usize = 8; + let mut redacted = message.to_string(); + for len in (SHORTEST_PREFIX..=token.len()).rev() { + if let Some(prefix) = token.get(..len) { + redacted = redacted.replace(prefix, "[REDACTED]"); + } + } + redacted +} + +#[cfg(test)] +mod tests { + use super::{describe, is_rate_limit_response, redact, RobotLBError}; + use hcloud::apis::{load_balancers_api::ListLoadBalancersError, Error, ResponseContent}; + + const TOKEN: &str = "abcdefghijklmnopqrstuvwxyz0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ01"; + + fn response_error(status: u16, content: &str) -> Error { + Error::ResponseError(ResponseContent { + status: status.try_into().unwrap(), + content: content.to_string(), + entity: None, + }) + } + + #[test] + fn a_token_prefix_is_redacted() { + let message = format!("limit reached for token {}", &TOKEN[..32]); + assert_eq!( + redact(&message, TOKEN), + "limit reached for token [REDACTED]" + ); + } + + #[test] + fn the_whole_token_is_redacted() { + assert_eq!(redact(&format!("x{TOKEN}y"), TOKEN), "x[REDACTED]y"); + } + + #[test] + fn a_message_without_the_token_is_kept() { + assert_eq!(redact("status 500", TOKEN), "status 500"); + } + + #[test] + fn an_empty_token_redacts_nothing() { + assert_eq!(redact("status 500", ""), "status 500"); + } + + #[test] + fn the_hetzner_message_is_described_on_one_line() { + let error = response_error( + 429, + r#"{"error": {"code": "rate_limit_exceeded", "message": "limit\n reached"}}"#, + ); + assert_eq!( + describe(&error), + "429 Too Many Requests: rate_limit_exceeded: limit reached" + ); + } + + #[test] + fn a_body_that_is_not_json_falls_back_to_the_status() { + let error = response_error(502, "bad gateway"); + assert_eq!(describe(&error), "502 Bad Gateway"); + } + + #[test] + fn a_pause_shorter_than_a_second_is_not_reported_as_over() { + let error = RobotLBError::RateLimited(std::time::Duration::from_millis(300)); + assert!(error.to_string().ends_with("in 1s")); + } + + #[test] + fn a_429_is_a_rate_limit() { + let error = RobotLBError::from(response_error(429, "")); + assert!(error.is_rate_limited()); + } + + #[test] + fn only_a_429_response_is_a_rate_limit() { + assert!(is_rate_limit_response(&response_error(429, ""))); + // Hetzner answers 422 for a target outside the vSwitch subnet. + assert!(!is_rate_limit_response(&response_error(422, ""))); + } + + #[test] + fn other_statuses_are_not_a_rate_limit() { + assert!(!RobotLBError::from(response_error(500, "")).is_rate_limited()); + assert!(!RobotLBError::SkipService.is_rate_limited()); + } +} diff --git a/src/lb.rs b/src/lb.rs index 8725314..2da9c40 100644 --- a/src/lb.rs +++ b/src/lb.rs @@ -366,9 +366,13 @@ impl LoadBalancer { // which must not keep the remaining nodes out of the load balancer. match added { Ok(_) => live += 1, + // Every further call would be rejected too and only drain the budget. + Err(error) if crate::error::is_rate_limit_response(&error) => { + return Err(error.into()); + } Err(error) => { tracing::warn!("Cannot add target {ip}: {error}"); - last_error = Some(error.to_string()); + last_error = Some(crate::error::describe(&error)); } } } @@ -513,46 +517,11 @@ impl LoadBalancer { Ok(()) } - /// Cleanup the load balancer. - /// This method will remove all the services and targets from the - /// load balancer. + /// Delete the balancer of the service, found by its name. pub async fn cleanup(&self) -> RobotLBResult<()> { let Some(hcloud_balancer) = self.get_hcloud_lb().await? else { return Ok(()); }; - for service in &hcloud_balancer.services { - tracing::info!( - "Deleting service that listens for port {} from load-balancer {}", - service.listen_port, - hcloud_balancer.name, - ); - hcloud::apis::load_balancers_api::delete_service( - &self.hcloud_config, - DeleteServiceParams { - id: hcloud_balancer.id, - delete_service_request: Some(DeleteServiceRequest { - listen_port: service.listen_port, - }), - }, - ) - .await?; - } - for target in &hcloud_balancer.targets { - if let Some(target_ip) = target.ip.clone() { - tracing::info!("Removing target {}", target_ip.ip); - hcloud::apis::load_balancers_api::remove_target( - &self.hcloud_config, - RemoveTargetParams { - id: hcloud_balancer.id, - remove_target_request: Some(RemoveTargetRequest { - ip: Some(target_ip), - ..Default::default() - }), - }, - ) - .await?; - } - } hcloud::apis::load_balancers_api::delete_load_balancer( &self.hcloud_config, DeleteLoadBalancerParams { @@ -619,16 +588,9 @@ impl LoadBalancer { }), }, ) - .await; - if let Err(e) = response { - tracing::error!("Failed to create load balancer: {:?}", e); - return Err(RobotLBError::HCloudError(format!( - "Failed to create load balancer: {:?}", - e - ))); - } + .await?; - Ok(*response.unwrap().load_balancer) + Ok(*response.load_balancer) } /// Get the network from Hetzner Cloud. diff --git a/src/main.rs b/src/main.rs index 164cc6f..e3c4475 100644 --- a/src/main.rs +++ b/src/main.rs @@ -18,7 +18,7 @@ use clap::Parser; use config::OperatorConfig; -use error::{RobotLBError, RobotLBResult}; +use error::{redact, RobotLBError, RobotLBResult}; use futures::StreamExt; use hcloud::apis::configuration::Configuration as HCloudConfig; use k8s_openapi::{ @@ -30,12 +30,22 @@ use k8s_openapi::{ }; use kube::{ api::{ListParams, PatchParams}, - runtime::{controller::Action, watcher, Controller}, + runtime::{ + controller::{self, Action}, + events::{Event, EventType, Recorder, Reporter}, + watcher, Controller, + }, Resource, ResourceExt, }; use label_filter::LabelFilter; use lb::{LBService, LoadBalancer}; -use std::{collections::HashSet, str::FromStr, sync::Arc, time::Duration}; +use rate_limit::{spread, RateLimitGate}; +use std::{ + collections::HashSet, + str::FromStr, + sync::Arc, + time::{Duration, Instant}, +}; pub mod config; pub mod consts; @@ -43,6 +53,7 @@ pub mod error; pub mod finalizers; pub mod label_filter; pub mod lb; +pub mod rate_limit; #[cfg(not(target_env = "msvc"))] #[global_allocator] @@ -69,27 +80,38 @@ async fn main() -> RobotLBResult<()> { hcloud_conf, )); tracing::info!("Starting the controller"); + let token = operator_config.hcloud_token; Controller::new( kube::Api::::all(kube_client), watcher::Config::default(), ) .run(reconcile_service, on_error, context) - .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 { + .for_each(|reconcilation_result| { + let token = token.clone(); + async move { + match reconcilation_result { + Ok((service, _action)) => { + tracing::info!("Reconcilation of a service {} was successful", service.name); + } // During reconcilation process, // the controller has decided to skip the service. - kube::runtime::controller::Error::ReconcilerFailed( - RobotLBError::SkipService, - _, - ) => {} - _ => { - tracing::error!("Error reconciling service: {:#?}", err); + Err(controller::Error::ReconcilerFailed(RobotLBError::SkipService, _)) => {} + Err(controller::Error::ReconcilerFailed( + error @ RobotLBError::RateLimited(_), + service, + )) => { + tracing::info!("Service {service}: {error}"); } - }, + Err(controller::Error::ReconcilerFailed(error, service)) => { + tracing::error!( + "Reconcilation of service {service} failed: {}", + redact(&error.to_string(), &token) + ); + } + Err(error) => { + tracing::error!("Controller error: {}", redact(&error.to_string(), &token)); + } + } } }) .await; @@ -101,18 +123,16 @@ pub struct CurrentContext { pub client: kube::Client, pub config: OperatorConfig, pub hcloud_config: HCloudConfig, + pub rate_limit: Arc, } impl CurrentContext { #[must_use] - pub const fn new( - client: kube::Client, - config: OperatorConfig, - hcloud_config: HCloudConfig, - ) -> Self { + pub fn new(client: kube::Client, config: OperatorConfig, hcloud_config: HCloudConfig) -> Self { Self { client, config, hcloud_config, + rate_limit: Arc::default(), } } } @@ -126,35 +146,111 @@ pub async fn reconcile_service( svc: Arc, context: Arc, ) -> RobotLBResult { - let svc_type = svc - .spec - .as_ref() - .and_then(|s| s.type_.as_ref()) - .map(String::as_str) - .unwrap_or("ClusterIP"); - if svc_type != "LoadBalancer" { - tracing::debug!("Service type is not LoadBalancer. Skipping..."); - return Err(RobotLBError::SkipService); + let result = sync_service(svc.clone(), context.clone()).await; + if let Err(error) = &result { + if publishes_event(error) { + report_failure(&svc, &context, error).await; + } } + result +} + +/// 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 { + !matches!(error, RobotLBError::SkipService) +} - let lb_type = svc +/// Put the error on the service as a warning event, where `kubectl describe` shows it. +async fn report_failure(svc: &Service, context: &CurrentContext, error: &RobotLBError) { + let recorder = Recorder::new( + context.client.clone(), + Reporter { + controller: "robotlb".to_string(), + instance: None, + }, + svc.object_ref(&()), + ); + let published = recorder + .publish(Event { + type_: EventType::Warning, + reason: "SyncLoadBalancerFailed".to_string(), + note: Some(event_note(error, &context.config.hcloud_token)), + action: "Reconcile".to_string(), + secondary: None, + }) + .await; + if let Err(publish_error) = published { + tracing::warn!("Cannot publish an event for the service: {publish_error}"); + } +} + +/// The error as an event note: token redacted and cut to the size the API accepts. +fn event_note(error: &RobotLBError, token: &str) -> String { + const MAX_NOTE_BYTES: usize = 1024; + let mut note = 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 +} + +#[derive(Debug, PartialEq, Eq)] +enum ServiceRole { + /// The service is not robotlb's. + Skip, + /// The service needs its load balancer created or updated. + Reconcile, + /// The service had a load balancer and no longer needs it. + Release, +} + +/// The API server wipes `loadBalancerClass` together with the `LoadBalancer` type, +/// so once the type changes only the finalizer still marks the service as robotlb's, +/// and a service with the finalizer that robotlb no longer serves gets released. +fn service_role(svc: &Service) -> ServiceRole { + let is_load_balancer = + svc.spec.as_ref().and_then(|s| s.type_.as_deref()) == Some("LoadBalancer"); + let class = svc .spec .as_ref() - .and_then(|s| s.load_balancer_class.as_ref()) - .map(String::as_str) + .and_then(|s| s.load_balancer_class.as_deref()) .unwrap_or(consts::ROBOTLB_LB_CLASS); - if lb_type != consts::ROBOTLB_LB_CLASS { - tracing::debug!("Load balancer class is not robotlb. Skipping..."); + let is_robotlb = is_load_balancer && class == consts::ROBOTLB_LB_CLASS; + let owned = finalizers::check(svc); + let deleting = svc.meta().deletion_timestamp.is_some(); + if (deleting && is_robotlb) || (owned && !is_robotlb) { + ServiceRole::Release + } else if is_robotlb && !deleting { + ServiceRole::Reconcile + } else { + ServiceRole::Skip + } +} + +async fn sync_service(svc: Arc, context: Arc) -> RobotLBResult { + let role = service_role(&svc); + if role == ServiceRole::Skip { + tracing::debug!("Service is not a robotlb load balancer. Skipping..."); return Err(RobotLBError::SkipService); } + // Hetzner counts requests per project, so while one service is rate limited + // every other one waits too instead of spending the budget being waited for. + if let Some(wait) = context.rate_limit.remaining(Instant::now()) { + return Err(RobotLBError::RateLimited(wait)); + } + tracing::info!("Starting service reconcilation"); let lb = LoadBalancer::try_from_svc(&svc, &context)?; - // If the service is being deleted, we need to clean up the resources. - if svc.meta().deletion_timestamp.is_some() { - tracing::info!("Service deletion detected. Cleaning up resources."); + if role == ServiceRole::Release { + tracing::info!("Service no longer needs a load balancer. Cleaning up resources."); lb.cleanup().await?; finalizers::remove(context.client.clone(), &svc).await?; return Ok(Action::await_change()); @@ -436,23 +532,23 @@ pub async fn reconcile_load_balancer( lb.add_service(service.listen_port, service.target_port); } - let svc_api = kube::Api::::namespaced( - context.client.clone(), - svc.namespace() - .unwrap_or_else(|| context.client.default_namespace().to_string()) - .as_str(), - ); - // A balancer without a single service forwards nothing while still being billed, // so none is created until the service has a port that can be exposed. + // An existing balancer and the address it gave the service are kept: a missing + // nodePort is far more likely a mistake than a request to delete the balancer. if lb.services.is_empty() { - tracing::warn!("Service has no port that can be exposed. Skipping the load balancer."); - clear_ingress_status(&svc_api, &svc).await?; - return Ok(Action::requeue(Duration::from_secs(30))); + return Err(RobotLBError::NoExposablePorts); } let hcloud_lb = lb.reconcile().await?; + let svc_api = kube::Api::::namespaced( + context.client.clone(), + svc.namespace() + .unwrap_or_else(|| context.client.default_namespace().to_string()) + .as_str(), + ); + let mut ingress = vec![]; let dns_ipv4 = hcloud_lb.public_net.ipv4.dns_ptr.flatten(); @@ -495,40 +591,25 @@ pub async fn reconcile_load_balancer( Ok(Action::requeue(Duration::from_secs(30))) } -/// Drop the external IP a service advertises, so that nothing keeps sending -/// traffic to a load balancer that no longer forwards it. -async fn clear_ingress_status(svc_api: &kube::Api, svc: &Service) -> RobotLBResult<()> { - let advertises_ingress = svc - .status - .as_ref() - .and_then(|status| status.load_balancer.as_ref()) - .and_then(|lb| lb.ingress.as_ref()) - .is_some_and(|ingress| !ingress.is_empty()); - if !advertises_ingress { - return Ok(()); - } - tracing::info!("Removing the external IP from the service status"); - svc_api - .patch_status( - svc.name_any().as_str(), - &PatchParams::default(), - &kube::api::Patch::Merge(json!({ - "status": { - "loadBalancer": { - "ingress": null - } - } - })), - ) - .await?; - Ok(()) -} - /// Handle the error during reconcilation. #[allow(clippy::needless_pass_by_value)] -fn on_error(_: Arc, error: &RobotLBError, _context: Arc) -> Action { +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) +} + +fn error_action( + error: &RobotLBError, + rate_limit: &RateLimitGate, + now: Instant, + service: &str, +) -> Action { match error { RobotLBError::SkipService => Action::await_change(), + RobotLBError::RateLimited(wait) => Action::requeue(spread(*wait, service)), + error if error.is_rate_limited() => { + Action::requeue(spread(rate_limit.on_rate_limited(now), service)) + } _ => Action::requeue(Duration::from_secs(30)), } } @@ -536,8 +617,9 @@ fn on_error(_: Arc, error: &RobotLBError, _context: Arc #[cfg(test)] mod tests { use super::{ - collect_lb_services, consts, is_excluded_from_lb, is_lb_eligible_node, - is_local_traffic_policy, node_source, NodeSource, + collect_lb_services, consts, error_action, event_note, is_excluded_from_lb, + is_lb_eligible_node, is_local_traffic_policy, node_source, publishes_event, service_role, + NodeSource, ServiceRole, }; use k8s_openapi::{ api::core::v1::{ @@ -733,4 +815,182 @@ mod tests { }); assert_eq!(node_source(&svc, false), NodeSource::Annotation); } + + #[test] + fn an_event_note_is_redacted_and_bounded() { + let token = "0123456789abcdef"; + let error = crate::error::RobotLBError::HCloudError(format!( + "rejected token {} {}", + &token[..10], + "é".repeat(600) + )); + let note = event_note(&error, token); + assert!(!note.contains(&token[..10])); + assert!(note.contains("[REDACTED]")); + assert!(note.len() <= 1024); + } + + #[test] + fn a_gated_service_waits_out_the_pause() { + let gate = crate::rate_limit::RateLimitGate::default(); + 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"), + kube::runtime::controller::Action::requeue(crate::rate_limit::spread(wait, "shop/web")) + ); + } + + #[test] + fn a_rate_limited_call_closes_the_gate() { + let gate = crate::rate_limit::RateLimitGate::default(); + let now = std::time::Instant::now(); + let error = 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, + }, + )); + assert_eq!( + error_action(&error, &gate, now, "shop/web"), + kube::runtime::controller::Action::requeue(crate::rate_limit::spread( + std::time::Duration::from_secs(60), + "shop/web" + )) + ); + assert!(gate.remaining(now).is_some()); + } + + #[test] + fn other_errors_retry_in_30_seconds() { + let gate = crate::rate_limit::RateLimitGate::default(); + 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)) + ); + } + + #[test] + fn every_failure_except_a_skip_publishes_an_event() { + use crate::error::RobotLBError; + assert!(!publishes_event(&RobotLBError::SkipService)); + // Jitter wakes the same service first after every pause, so the others only + // ever see the closed gate, and without an event of their own they go silent. + assert!(publishes_event(&RobotLBError::RateLimited( + std::time::Duration::from_secs(1) + ))); + assert!(publishes_event(&RobotLBError::from(hcloud::apis::Error::< + hcloud::apis::load_balancers_api::ListLoadBalancersError, + >::ResponseError( + hcloud::apis::ResponseContent { + status: 429_u16.try_into().unwrap(), + content: String::new(), + entity: None, + } + )))); + assert!(publishes_event(&RobotLBError::HCloudError( + "boom".to_string() + ))); + } + + fn owned_service(type_: &str, class: Option<&str>, deleting: bool) -> Service { + Service { + metadata: ObjectMeta { + finalizers: Some(vec![consts::FINALIZER_NAME.to_string()]), + deletion_timestamp: deleting.then(|| { + k8s_openapi::apimachinery::pkg::apis::meta::v1::Time( + k8s_openapi::chrono::Utc::now(), + ) + }), + ..Default::default() + }, + spec: Some(ServiceSpec { + type_: Some(type_.to_string()), + load_balancer_class: class.map(str::to_string), + ..Default::default() + }), + ..Default::default() + } + } + + #[test] + fn a_robotlb_load_balancer_is_reconciled() { + let svc = service(ServiceSpec { + type_: Some("LoadBalancer".to_string()), + ..Default::default() + }); + assert_eq!(service_role(&svc), ServiceRole::Reconcile); + let svc = owned_service("LoadBalancer", Some(consts::ROBOTLB_LB_CLASS), false); + assert_eq!(service_role(&svc), ServiceRole::Reconcile); + } + + #[test] + fn services_robotlb_does_not_own_are_skipped() { + let other_class = service(ServiceSpec { + type_: Some("LoadBalancer".to_string()), + load_balancer_class: Some("example.com/other".to_string()), + ..Default::default() + }); + assert_eq!(service_role(&other_class), ServiceRole::Skip); + let cluster_ip = service(ServiceSpec { + type_: Some("ClusterIP".to_string()), + ..Default::default() + }); + assert_eq!(service_role(&cluster_ip), ServiceRole::Skip); + let mut deleting_cluster_ip = owned_service("ClusterIP", None, true); + deleting_cluster_ip.metadata.finalizers = None; + assert_eq!(service_role(&deleting_cluster_ip), ServiceRole::Skip); + let mut deleting_other_class = + owned_service("LoadBalancer", Some("example.com/other"), true); + deleting_other_class.metadata.finalizers = None; + assert_eq!(service_role(&deleting_other_class), ServiceRole::Skip); + } + + // A type change and a change back with another class can both land before robotlb + // reconciles, for example while the rate limit gate is closed. + #[test] + fn an_owned_service_taken_over_by_another_class_is_released() { + let svc = owned_service("LoadBalancer", Some("example.com/other"), false); + assert_eq!(service_role(&svc), ServiceRole::Release); + } + + // Reachable when the finalizer was removed by hand while another controller's + // finalizer still holds the object. + #[test] + fn a_deleted_robotlb_load_balancer_without_the_finalizer_is_released() { + let mut svc = owned_service("LoadBalancer", None, true); + svc.metadata.finalizers = None; + assert_eq!(service_role(&svc), ServiceRole::Release); + } + + #[test] + fn an_owned_service_that_stopped_being_a_load_balancer_is_released() { + assert_eq!( + service_role(&owned_service("ClusterIP", None, false)), + ServiceRole::Release + ); + } + + #[test] + fn a_deleted_owned_service_is_released() { + assert_eq!( + service_role(&owned_service("LoadBalancer", None, true)), + ServiceRole::Release + ); + assert_eq!( + service_role(&owned_service("ClusterIP", None, true)), + ServiceRole::Release + ); + } + + #[test] + fn a_service_without_exposable_ports_reports_an_event() { + let error = crate::error::RobotLBError::NoExposablePorts; + assert!(publishes_event(&error)); + assert!(error.to_string().starts_with("No TCP port")); + } } diff --git a/src/rate_limit.rs b/src/rate_limit.rs new file mode 100644 index 0000000..2792e08 --- /dev/null +++ b/src/rate_limit.rs @@ -0,0 +1,176 @@ +use std::{ + collections::hash_map::DefaultHasher, + hash::{Hash, Hasher}, + sync::Mutex, + time::{Duration, Instant}, +}; + +const FIRST_DELAY: Duration = Duration::from_secs(60); +const MAX_DOUBLINGS: u32 = 4; + +/// Pause shared by every service. Hetzner counts API requests per project, so once one +/// reconciliation is rate limited, any other call would only spend the budget being waited for. +/// +/// The generated `hcloud` client drops response headers, so `RateLimit-Reset` is not +/// available and the pause grows exponentially instead. +#[derive(Debug, Default)] +pub struct RateLimitGate { + state: Mutex, +} + +#[derive(Debug, Default)] +struct State { + closed_until: Option, + last_delay: Duration, + consecutive: u32, +} + +impl State { + fn remaining(&self, now: Instant) -> Option { + self.closed_until + .and_then(|until| until.checked_duration_since(now)) + .filter(|remaining| !remaining.is_zero()) + } +} + +impl RateLimitGate { + /// How long calls to the API must still wait, if they must. + pub fn remaining(&self, now: Instant) -> Option { + let state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + state.remaining(now) + } + + /// Record a rate-limited call and return how long to wait before the next one. + /// A call rejected while the gate is already closed was sent before it closed, + /// so it keeps the current pause instead of lengthening it. The pause starts short + /// again only after the API went without a 429 for as long as the last pause: + /// a reconciliation that succeeds may not have called the API at all. + pub fn on_rate_limited(&self, now: Instant) -> Duration { + let mut state = self + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(remaining) = state.remaining(now) { + return remaining; + } + if let Some(reopened) = state.closed_until { + if now >= reopened + state.last_delay { + state.consecutive = 0; + } + } + let delay = FIRST_DELAY * 2_u32.pow(state.consecutive.min(MAX_DOUBLINGS)); + state.consecutive += 1; + state.closed_until = Some(now + delay); + state.last_delay = delay; + delay + } +} + +/// Push a wait out by up to a quarter, differently for each service, so services +/// paused together do not all call the API the moment the pause ends. +#[must_use] +pub fn spread(wait: Duration, service: &str) -> Duration { + let mut hasher = DefaultHasher::new(); + service.hash(&mut hasher); + let window = u64::try_from((wait / 4).as_millis()) + .unwrap_or(u64::MAX) + .max(1); + wait + Duration::from_millis(hasher.finish() % window) +} + +#[cfg(test)] +mod tests { + use super::{spread, RateLimitGate}; + use std::time::{Duration, Instant}; + + #[test] + fn an_untouched_gate_is_open() { + assert_eq!(RateLimitGate::default().remaining(Instant::now()), None); + } + + #[test] + fn a_rate_limit_closes_the_gate_for_the_returned_delay() { + let gate = RateLimitGate::default(); + let now = Instant::now(); + let delay = gate.on_rate_limited(now); + assert_eq!(delay, Duration::from_secs(60)); + assert_eq!(gate.remaining(now), Some(delay)); + assert_eq!(gate.remaining(now + delay), None); + } + + #[test] + fn repeated_rate_limits_double_the_delay_up_to_the_cap() { + let gate = RateLimitGate::default(); + let mut now = Instant::now(); + let mut delays = Vec::new(); + for _ in 0..8 { + let delay = gate.on_rate_limited(now); + delays.push(delay.as_secs()); + now += delay; + } + assert_eq!(delays, vec![60, 120, 240, 480, 960, 960, 960, 960]); + } + + #[test] + fn a_quiet_period_as_long_as_the_last_pause_resets_the_delay() { + let gate = RateLimitGate::default(); + let now = Instant::now(); + let first = gate.on_rate_limited(now); + let second = gate.on_rate_limited(now + first); + let reopened = now + first + second; + assert_eq!( + gate.on_rate_limited(reopened + second), + Duration::from_secs(60) + ); + } + + #[test] + fn a_shorter_quiet_period_keeps_doubling() { + let gate = RateLimitGate::default(); + let now = Instant::now(); + let first = gate.on_rate_limited(now); + let second = gate.on_rate_limited(now + first); + let reopened = now + first + second; + assert_eq!( + gate.on_rate_limited(reopened + second.checked_sub(Duration::from_secs(1)).unwrap()), + Duration::from_secs(240) + ); + } + + #[test] + fn calls_already_in_flight_do_not_lengthen_the_pause() { + let gate = RateLimitGate::default(); + let now = Instant::now(); + gate.on_rate_limited(now); + let later = now + Duration::from_secs(10); + assert_eq!(gate.on_rate_limited(later), Duration::from_secs(50)); + assert_eq!( + gate.on_rate_limited(later + Duration::from_secs(50)), + Duration::from_secs(120) + ); + } + + #[test] + fn spread_adds_up_to_a_quarter_of_the_pause() { + let wait = Duration::from_secs(60); + for name in ["web", "api", "dns", "ingress-nginx-controller"] { + let spread_wait = spread(wait, name); + assert!(spread_wait >= wait); + assert!(spread_wait < wait + wait / 4); + } + } + + #[test] + fn spread_is_stable_per_service_and_differs_between_services() { + let wait = Duration::from_secs(60); + assert_eq!(spread(wait, "web"), spread(wait, "web")); + let wakeups = ["a", "b", "c", "d", "e", "f"] + .map(|name| spread(wait, name)) + .into_iter() + .collect::>(); + assert!(wakeups.len() > 1); + } +}