diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 69ed78b..3c57201 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -27,6 +27,17 @@ repos: - clippy - --all + - id: test + types: + - rust + name: cargo test + language: system + pass_filenames: false + entry: cargo + args: + - test + - --all + - id: check types: - rust diff --git a/README.md b/README.md index b3b0585..fb3afc7 100644 --- a/README.md +++ b/README.md @@ -35,7 +35,20 @@ After the chart is installed, you should be able to create `LoadBalancer` servic The operator listens to the Kubernetes API for services of type `LoadBalancer` and creates Hetzner load balancers that point to nodes based on `node-ip`. -Nodes are selected based on where the service's target pods are deployed, which is determined by searching for pods with the service's selector. This behavior can be configured. +Target nodes are selected according to the service's `externalTrafficPolicy`: + +- `Cluster`, the Kubernetes default: every node of the cluster becomes a target, since kube-proxy forwards the traffic to a node that hosts a pod. Cordoned and not-ready nodes are left out, as they would only take up target slots. +- `Local`: only the nodes where the service's target pods run, found through the service selector, or through the service's `EndpointSlice` resources when it has no selector. + +Nodes labelled `node.kubernetes.io/exclude-from-external-load-balancers`, which kubeadm puts on control-plane nodes, stay out of every balancer under either policy. + +Setting `ROBOTLB_DYNAMIC_NODE_SELECTOR` to `false` replaces both with the node selector from the `robotlb/node-selector` annotation. + +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. + +> 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. ## Configuration diff --git a/src/consts.rs b/src/consts.rs index 820fa02..0cedb34 100644 --- a/src/consts.rs +++ b/src/consts.rs @@ -22,5 +22,10 @@ pub const DEFAULT_LB_LOCATION: &str = "hel1"; pub const DEFAULT_LB_ALGORITHM: &str = "least-connections"; pub const DEFAULT_LB_BALANCER_TYPE: &str = "lb11"; +/// Well-known Kubernetes label marking a node that must never be a load balancer target. +/// kubeadm applies it to every control-plane node. +pub const EXCLUDE_FROM_LB_LABEL_NAME: &str = + "node.kubernetes.io/exclude-from-external-load-balancers"; + pub const FINALIZER_NAME: &str = "robotlb/finalizer"; pub const ROBOTLB_LB_CLASS: &str = "robotlb"; diff --git a/src/lb.rs b/src/lb.rs index bac19f1..8725314 100644 --- a/src/lb.rs +++ b/src/lb.rs @@ -300,11 +300,28 @@ impl LoadBalancer { &self, hcloud_balancer: &hcloud::models::LoadBalancer, ) -> RobotLBResult<()> { + let max_targets = + usize::try_from(hcloud_balancer.load_balancer_type.max_targets).unwrap_or(usize::MAX); + let planned = plan_targets(&self.targets, max_targets); + if planned.len() < self.targets.len() + // While a type change is in flight the balancer still reports the old type, + // whose limit says nothing about the type the service asked for. + && hcloud_balancer.load_balancer_type.name == self.balancer_type + { + tracing::warn!( + "Selected {} node(s), but a {} balancer holds at most {}. \ + Use a bigger balancer type or externalTrafficPolicy: Local.", + self.targets.len(), + hcloud_balancer.load_balancer_type.name, + max_targets, + ); + } + for target in &hcloud_balancer.targets { let Some(target_ip) = target.ip.clone() else { continue; }; - if !self.targets.contains(&target_ip.ip) { + if !planned.contains(&target_ip.ip.as_str()) { tracing::info!("Removing target {}", target_ip.ip); hcloud::apis::load_balancers_api::remove_target( &self.hcloud_config, @@ -320,27 +337,50 @@ impl LoadBalancer { } } - for ip in &self.targets { - if !hcloud_balancer + let mut live = 0_usize; + let mut last_error = None; + for ip in &planned { + if hcloud_balancer .targets .iter() - .any(|t| t.ip.as_ref().map(|i| i.ip.as_str()) == Some(ip)) + .any(|t| t.ip.as_ref().map(|i| i.ip.as_str()) == Some(*ip)) { - tracing::info!("Adding target {}", ip); - hcloud::apis::load_balancers_api::add_target( - &self.hcloud_config, - AddTargetParams { - id: hcloud_balancer.id, - body: Some(LoadBalancerAddTarget { - ip: Some(Box::new(hcloud::models::LoadBalancerTargetIp { - ip: ip.clone(), - })), - ..Default::default() - }), - }, - ) - .await?; + live += 1; + continue; } + tracing::info!("Adding target {}", ip); + let added = hcloud::apis::load_balancers_api::add_target( + &self.hcloud_config, + AddTargetParams { + id: hcloud_balancer.id, + body: Some(LoadBalancerAddTarget { + ip: Some(Box::new(hcloud::models::LoadBalancerTargetIp { + ip: (*ip).to_string(), + })), + ..Default::default() + }), + }, + ) + .await; + // Hetzner rejects IPs outside the vSwitch subnet of the attached network, + // which must not keep the remaining nodes out of the load balancer. + match added { + Ok(_) => live += 1, + Err(error) => { + tracing::warn!("Cannot add target {ip}: {error}"); + last_error = Some(error.to_string()); + } + } + } + // A balancer left without a single target forwards nothing, so the service + // must not be reported as ready. Counting what is live rather than what this + // run attempted keeps a permanently rejected node from failing every run. + if !planned.is_empty() && live == 0 { + return Err(RobotLBError::HCloudError(format!( + "No target could be added to load balancer {}: {}", + self.name, + last_error.unwrap_or_else(|| "no reason reported".to_string()), + ))); } Ok(()) } @@ -631,6 +671,17 @@ impl LoadBalancer { } } +/// The targets a balancer should end up with: deduplicated, and trimmed to what the +/// balancer type holds. Sorted, so that a cluster larger than the limit keeps the same +/// targets from one reconciliation to the next instead of trading them back and forth. +fn plan_targets(desired: &[String], max_targets: usize) -> Vec<&str> { + let mut planned = desired.iter().map(String::as_str).collect::>(); + planned.sort_unstable(); + planned.dedup(); + planned.truncate(max_targets); + planned +} + impl FromStr for LBAlgorithm { type Err = RobotLBError; fn from_str(s: &str) -> Result { @@ -653,3 +704,38 @@ impl From for LoadBalancerAlgorithm { Self { r#type } } } + +#[cfg(test)] +mod tests { + use super::plan_targets; + + #[test] + fn targets_are_sorted_and_deduplicated() { + let desired = vec![ + "192.168.100.4".to_string(), + "192.168.100.2".to_string(), + "192.168.100.4".to_string(), + ]; + assert_eq!( + plan_targets(&desired, 25), + vec!["192.168.100.2", "192.168.100.4"] + ); + } + + #[test] + fn targets_beyond_the_balancer_limit_are_dropped() { + let desired = (1..=30) + .map(|host| format!("192.168.100.{host:03}")) + .collect::>(); + let planned = plan_targets(&desired, 25); + assert_eq!(planned.len(), 25); + assert_eq!(planned[0], "192.168.100.001"); + assert_eq!(planned[24], "192.168.100.025"); + } + + #[test] + fn a_plan_within_the_limit_keeps_every_target() { + let desired = vec!["192.168.100.2".to_string(), "192.168.100.3".to_string()]; + assert_eq!(plan_targets(&desired, 25).len(), 2); + } +} diff --git a/src/main.rs b/src/main.rs index b237572..164cc6f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -34,7 +34,7 @@ use kube::{ Resource, ResourceExt, }; use label_filter::LabelFilter; -use lb::LoadBalancer; +use lb::{LBService, LoadBalancer}; use std::{collections::HashSet, str::FromStr, sync::Arc, time::Duration}; pub mod config; @@ -220,7 +220,7 @@ async fn get_nodes_dynamically( .list(&ListParams::default()) .await? .into_iter() - .filter(|node| target_nodes.contains(&node.name_any())) + .filter(|node| target_nodes.contains(&node.name_any()) && !is_excluded_from_lb(node)) .collect::>(); Ok(nodes) @@ -268,7 +268,7 @@ async fn get_nodes_from_endpointslices( .list(&ListParams::default()) .await? .into_iter() - .filter(|node| target_nodes.contains(&node.name_any())) + .filter(|node| target_nodes.contains(&node.name_any()) && !is_excluded_from_lb(node)) .collect::>(); Ok(nodes) @@ -292,11 +292,113 @@ async fn get_nodes_by_selector( .list(&ListParams::default()) .await? .into_iter() - .filter(|node| label_filter.check(node.labels())) + .filter(|node| label_filter.check(node.labels()) && !is_excluded_from_lb(node)) .collect::>(); Ok(nodes) } +/// Get every node of the cluster that may serve load balancer traffic. +async fn get_all_nodes(context: &Arc) -> RobotLBResult> { + let nodes_api = kube::Api::::all(context.client.clone()); + let nodes = nodes_api + .list(&ListParams::default()) + .await? + .into_iter() + .filter(is_lb_eligible_node) + .collect::>(); + Ok(nodes) +} + +/// Whether the cluster declares that a node must stay out of external load balancers. +/// The label holds under every traffic policy, the way upstream cloud providers treat it. +fn is_excluded_from_lb(node: &Node) -> bool { + node.labels() + .contains_key(consts::EXCLUDE_FROM_LB_LABEL_NAME) +} + +/// Whether a node may be used as a load balancer target under the `Cluster` policy, +/// where any node can carry the traffic and a draining or unhealthy one only takes +/// up a target slot. +fn is_lb_eligible_node(node: &Node) -> bool { + if is_excluded_from_lb(node) { + tracing::debug!("Node {} is excluded from load balancers", node.name_any()); + return false; + } + if node.spec.as_ref().and_then(|spec| spec.unschedulable) == Some(true) { + tracing::debug!("Node {} is unschedulable", node.name_any()); + return false; + } + let ready = node + .status + .as_ref() + .and_then(|status| status.conditions.as_ref()) + .and_then(|conditions| conditions.iter().find(|cond| cond.type_ == "Ready")) + .map(|cond| cond.status.as_str()); + if matches!(ready, Some(status) if status != "True") { + tracing::debug!("Node {} is not ready", node.name_any()); + return false; + } + true +} + +/// Where the target nodes of a service come from. +#[derive(Debug, PartialEq, Eq)] +enum NodeSource { + /// The `robotlb/node-selector` annotation of the service. + Annotation, + /// The nodes hosting the endpoints of the service. + ServiceEndpoints, + /// Every node that may serve load balancer traffic. + AllNodes, +} + +fn node_source(svc: &Service, dynamic_node_selector: bool) -> NodeSource { + if !dynamic_node_selector { + return NodeSource::Annotation; + } + if is_local_traffic_policy(svc) { + return NodeSource::ServiceEndpoints; + } + NodeSource::AllNodes +} + +/// Whether the service asks for traffic to reach only the nodes that host its endpoints. +/// Under the default `Cluster` policy every node is a valid target, because kube-proxy +/// forwards the traffic to a node that actually hosts a pod. +/// +/// +fn is_local_traffic_policy(svc: &Service) -> bool { + svc.spec + .as_ref() + .and_then(|spec| spec.external_traffic_policy.as_deref()) + == Some("Local") +} + +/// Map the ports of a service onto load balancer services. +fn collect_lb_services(svc: &Service) -> Vec { + let mut services = Vec::new(); + for port in svc.spec.iter().flat_map(|spec| spec.ports.iter().flatten()) { + let protocol = port.protocol.as_deref().unwrap_or("TCP"); + if protocol != "TCP" { + tracing::warn!("Protocol {} is not supported. Skipping...", protocol); + continue; + } + let Some(node_port) = port.node_port else { + tracing::warn!( + "Service port {} has no nodePort allocated. Hetzner load balancers forward \ + traffic to node IPs, so such a port cannot be exposed. Skipping...", + port.port + ); + continue; + }; + services.push(LBService { + listen_port: port.port, + target_port: node_port, + }); + } + services +} + /// Reconcile the `LoadBalancer` type of service. /// This function will find the nodes based on the node selector /// and create or update the load balancer. @@ -310,10 +412,10 @@ pub async fn reconcile_load_balancer( node_ip_type = "ExternalIP"; } - let nodes = if context.config.dynamic_node_selector { - get_nodes_dynamically(&svc, &context).await? - } else { - get_nodes_by_selector(&svc, &context).await? + let nodes = match node_source(&svc, context.config.dynamic_node_selector) { + NodeSource::Annotation => get_nodes_by_selector(&svc, &context).await?, + NodeSource::ServiceEndpoints => get_nodes_dynamically(&svc, &context).await?, + NodeSource::AllNodes => get_all_nodes(&context).await?, }; for node in nodes { @@ -330,26 +432,8 @@ pub async fn reconcile_load_balancer( } } - for port in svc - .spec - .clone() - .unwrap_or_default() - .ports - .unwrap_or_default() - { - let protocol = port.protocol.unwrap_or_else(|| "TCP".to_string()); - if protocol != "TCP" { - tracing::warn!("Protocol {} is not supported. Skipping...", protocol); - continue; - } - let Some(node_port) = port.node_port else { - tracing::warn!( - "Node port is not set for target_port {}. Skipping...", - port.port - ); - continue; - }; - lb.add_service(port.port, node_port); + for service in collect_lb_services(&svc) { + lb.add_service(service.listen_port, service.target_port); } let svc_api = kube::Api::::namespaced( @@ -359,6 +443,14 @@ pub async fn reconcile_load_balancer( .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. + 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))); + } + let hcloud_lb = lb.reconcile().await?; let mut ingress = vec![]; @@ -403,6 +495,35 @@ 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 { @@ -411,3 +532,205 @@ fn on_error(_: Arc, error: &RobotLBError, _context: Arc _ => Action::requeue(Duration::from_secs(30)), } } + +#[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, + }; + use k8s_openapi::{ + api::core::v1::{ + Node, NodeCondition, NodeSpec, NodeStatus, Service, ServicePort, ServiceSpec, + }, + apimachinery::pkg::apis::meta::v1::ObjectMeta, + }; + use std::collections::BTreeMap; + + fn service(spec: ServiceSpec) -> Service { + Service { + spec: Some(spec), + ..Default::default() + } + } + + fn node(labels: &[(&str, &str)], unschedulable: bool) -> Node { + Node { + metadata: ObjectMeta { + labels: Some( + labels + .iter() + .map(|(key, value)| ((*key).to_string(), (*value).to_string())) + .collect::>(), + ), + ..Default::default() + }, + spec: Some(NodeSpec { + unschedulable: Some(unschedulable), + ..Default::default() + }), + ..Default::default() + } + } + + fn node_with_ready_condition(status: &str) -> Node { + Node { + status: Some(NodeStatus { + conditions: Some(vec![NodeCondition { + type_: "Ready".to_string(), + status: status.to_string(), + ..Default::default() + }]), + ..Default::default() + }), + ..node(&[], false) + } + } + + #[test] + fn cluster_policy_is_not_local() { + let svc = service(ServiceSpec { + external_traffic_policy: Some("Cluster".into()), + ..Default::default() + }); + assert!(!is_local_traffic_policy(&svc)); + } + + #[test] + fn unset_policy_defaults_to_cluster() { + assert!(!is_local_traffic_policy(&service(ServiceSpec::default()))); + } + + #[test] + fn local_policy_is_local() { + let svc = service(ServiceSpec { + external_traffic_policy: Some("Local".into()), + ..Default::default() + }); + assert!(is_local_traffic_policy(&svc)); + } + + #[test] + fn ports_map_listen_port_to_node_port() { + let svc = service(ServiceSpec { + ports: Some(vec![ServicePort { + port: 22, + node_port: Some(30821), + protocol: Some("TCP".into()), + ..Default::default() + }]), + ..Default::default() + }); + let services = collect_lb_services(&svc); + assert_eq!(services.len(), 1); + assert_eq!(services[0].listen_port, 22); + assert_eq!(services[0].target_port, 30821); + } + + #[test] + fn ports_without_protocol_are_treated_as_tcp() { + let svc = service(ServiceSpec { + ports: Some(vec![ServicePort { + port: 80, + node_port: Some(31571), + ..Default::default() + }]), + ..Default::default() + }); + let services = collect_lb_services(&svc); + assert_eq!(services.len(), 1); + assert_eq!(services[0].listen_port, 80); + assert_eq!(services[0].target_port, 31571); + } + + #[test] + fn udp_ports_are_skipped() { + let svc = service(ServiceSpec { + ports: Some(vec![ServicePort { + port: 53, + node_port: Some(30053), + protocol: Some("UDP".into()), + ..Default::default() + }]), + ..Default::default() + }); + assert!(collect_lb_services(&svc).is_empty()); + } + + #[test] + fn ports_without_node_port_are_skipped() { + let svc = service(ServiceSpec { + ports: Some(vec![ServicePort { + port: 22, + protocol: Some("TCP".into()), + ..Default::default() + }]), + ..Default::default() + }); + assert!(collect_lb_services(&svc).is_empty()); + } + + #[test] + fn plain_node_is_eligible() { + assert!(is_lb_eligible_node(&node( + &[("kubernetes.io/hostname", "ws1")], + false + ))); + } + + #[test] + fn excluded_node_is_not_eligible() { + assert!(!is_lb_eligible_node(&node( + &[(consts::EXCLUDE_FROM_LB_LABEL_NAME, "")], + false + ))); + } + + #[test] + fn cordoned_node_is_not_eligible() { + assert!(!is_lb_eligible_node(&node(&[], true))); + } + + #[test] + fn ready_node_is_eligible() { + assert!(is_lb_eligible_node(&node_with_ready_condition("True"))); + } + + #[test] + fn not_ready_node_is_not_eligible() { + assert!(!is_lb_eligible_node(&node_with_ready_condition("False"))); + } + + #[test] + fn exclusion_label_is_recognised_on_its_own() { + assert!(is_excluded_from_lb(&node( + &[(consts::EXCLUDE_FROM_LB_LABEL_NAME, "")], + false + ))); + assert!(!is_excluded_from_lb(&node(&[], true))); + } + + #[test] + fn cluster_policy_takes_every_node() { + let svc = service(ServiceSpec::default()); + assert_eq!(node_source(&svc, true), NodeSource::AllNodes); + } + + #[test] + fn local_policy_takes_endpoint_nodes() { + let svc = service(ServiceSpec { + external_traffic_policy: Some("Local".into()), + ..Default::default() + }); + assert_eq!(node_source(&svc, true), NodeSource::ServiceEndpoints); + } + + #[test] + fn static_selector_wins_over_the_policy() { + let svc = service(ServiceSpec { + external_traffic_policy: Some("Local".into()), + ..Default::default() + }); + assert_eq!(node_source(&svc, false), NodeSource::Annotation); + } +}