diff --git a/Cargo.lock b/Cargo.lock index e5943895bf14c..759e1e11924e3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1521,9 +1521,9 @@ dependencies = [ [[package]] name = "linkerd2-proxy-api" -version = "0.18.0" +version = "0.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba9e3b341ca4992feaf43a4d2bdbfe2081aa3e2b9a503753544ce55242af6342" +checksum = "e0cd682795e8f91ea36e2131a2f7021da136c74f6d3cd3d9eabbf7842511042d" dependencies = [ "http", "ipnet", diff --git a/Cargo.toml b/Cargo.toml index f54cee5f841ae..aabc662459c61 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -47,5 +47,5 @@ path = "./policy-controller/runtime" default-features = false [workspace.dependencies.linkerd2-proxy-api] -version = "0.18.0" +version = "0.19.0" features = ["inbound", "outbound"] diff --git a/policy-controller/core/src/outbound.rs b/policy-controller/core/src/outbound.rs index 69f1504883e6b..912f1f10e6856 100644 --- a/policy-controller/core/src/outbound.rs +++ b/policy-controller/core/src/outbound.rs @@ -160,6 +160,23 @@ pub struct Backoff { pub jitter: f32, } +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +pub struct LoadBiasConfig { + pub enabled: bool, + pub penalty: time::Duration, + pub penalty_decay: time::Duration, +} + +pub const DEFAULT_LOAD_BIAS_PENALTY: time::Duration = time::Duration::from_secs(5); +pub const DEFAULT_LOAD_BIAS_PENALTY_DECAY: time::Duration = time::Duration::from_secs(10); + +pub const DEFAULT_RETRY_AFTER_MAX_DURATION: time::Duration = time::Duration::from_secs(300); + +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +pub struct RetryAfterConfig { + pub max_duration: time::Duration, +} + #[derive(Clone, Debug, PartialEq, Eq)] pub enum Filter { RequestHeaderModifier(HeaderModifierFilter), diff --git a/policy-controller/core/src/outbound/policy.rs b/policy-controller/core/src/outbound/policy.rs index 273b8a7ffc49d..d52f8ccdf8c90 100644 --- a/policy-controller/core/src/outbound/policy.rs +++ b/policy-controller/core/src/outbound/policy.rs @@ -1,6 +1,7 @@ use super::{ AppProtocol, FailureAccrual, GrpcRetryCondition, GrpcRoute, HttpRetryCondition, HttpRoute, - RouteRetry, RouteSet, RouteTimeouts, TcpRoute, TlsRoute, TrafficPolicy, + LoadBiasConfig, RetryAfterConfig, RouteRetry, RouteSet, RouteTimeouts, TcpRoute, TlsRoute, + TrafficPolicy, }; use std::num::NonZeroU16; @@ -31,6 +32,8 @@ pub struct OutboundPolicy { pub port: NonZeroU16, pub app_protocol: Option, pub accrual: Option, + pub load_bias: Option, + pub retry_after: Option, pub http_retry: Option>, pub grpc_retry: Option>, pub timeouts: RouteTimeouts, diff --git a/policy-controller/grpc/src/outbound.rs b/policy-controller/grpc/src/outbound.rs index da463dcdc545a..230ae66e2f8c0 100644 --- a/policy-controller/grpc/src/outbound.rs +++ b/policy-controller/grpc/src/outbound.rs @@ -402,10 +402,14 @@ fn fallback(original_dst: SocketAddr) -> outbound::OutboundPolicy { http1: Some(outbound::proxy_protocol::Http1 { routes: http_routes.clone(), failure_accrual: None, + load_bias: None, + retry_after: None, }), http2: Some(outbound::proxy_protocol::Http2 { routes: http_routes, failure_accrual: None, + load_bias: None, + retry_after: None, }), }, )), @@ -472,22 +476,32 @@ fn to_proto( ) -> outbound::OutboundPolicy { let backend: outbound::Backend = default_backend(&policy, original_dst); - let accrual = policy.accrual.map(|accrual| outbound::FailureAccrual { - kind: Some(match accrual { - linkerd_policy_controller_core::outbound::FailureAccrual::Consecutive { + let accrual = policy.accrual.map(|accrual| { + let linkerd_policy_controller_core::outbound::FailureAccrual::Consecutive { + max_failures, + backoff, + } = accrual; + outbound::FailureAccrual { + consecutive_failures: Some(outbound::failure_accrual::ConsecutiveFailures { max_failures, - backoff, - } => outbound::failure_accrual::Kind::ConsecutiveFailures( - outbound::failure_accrual::ConsecutiveFailures { - max_failures, - backoff: Some(outbound::ExponentialBackoff { - min_backoff: convert_duration("min_backoff", backoff.min_penalty), - max_backoff: convert_duration("max_backoff", backoff.max_penalty), - jitter_ratio: backoff.jitter, - }), - }, - ), - }), + backoff: Some(outbound::ExponentialBackoff { + min_backoff: convert_duration("min_backoff", backoff.min_penalty), + max_backoff: convert_duration("max_backoff", backoff.max_penalty), + jitter_ratio: backoff.jitter, + }), + }), + success_rate: None, + } + }); + + let load_bias = policy.load_bias.map(|lb| outbound::LoadBiasConfig { + enabled: lb.enabled, + penalty: convert_duration("load_bias_penalty", lb.penalty), + penalty_decay: convert_duration("load_bias_penalty_decay", lb.penalty_decay), + }); + + let retry_after = policy.retry_after.map(|ra| outbound::RetryAfterConfig { + max_duration: convert_duration("retry_after_max_duration", ra.max_duration), }); let mut http_routes = policy.http_routes.clone().into_iter().collect::>(); @@ -499,6 +513,8 @@ fn to_proto( backend, http_routes.into_iter(), accrual, + load_bias, + retry_after, policy.http_retry.clone(), policy.timeouts.clone(), allow_l5d_request_headers, @@ -515,6 +531,8 @@ fn to_proto( backend, grpc_routes.into_iter(), accrual, + load_bias, + retry_after, policy.grpc_retry.clone(), policy.timeouts.clone(), allow_l5d_request_headers, @@ -527,6 +545,8 @@ fn to_proto( backend, http_routes.into_iter(), accrual, + load_bias, + retry_after, policy.http_retry.clone(), policy.timeouts.clone(), allow_l5d_request_headers, @@ -567,6 +587,8 @@ fn to_proto( backend, grpc_routes.into_iter(), accrual, + load_bias, + retry_after, policy.grpc_retry.clone(), policy.timeouts.clone(), allow_l5d_request_headers, @@ -579,6 +601,8 @@ fn to_proto( backend, http_routes.into_iter(), accrual, + load_bias, + retry_after, policy.http_retry.clone(), policy.timeouts.clone(), allow_l5d_request_headers, @@ -607,6 +631,8 @@ fn to_proto( backend, http_routes.into_iter(), accrual, + load_bias, + retry_after, policy.http_retry.clone(), policy.timeouts.clone(), allow_l5d_request_headers, @@ -690,6 +716,7 @@ fn default_backend(policy: &OutboundPolicy, original_dst: Option) -> )), }), load: Some(default_balancer_config()), + ejection: None, }, )), }, diff --git a/policy-controller/grpc/src/outbound/grpc.rs b/policy-controller/grpc/src/outbound/grpc.rs index f6b87076278ad..cd03ac826b8ed 100644 --- a/policy-controller/grpc/src/outbound/grpc.rs +++ b/policy-controller/grpc/src/outbound/grpc.rs @@ -21,6 +21,8 @@ pub(crate) fn protocol( default_backend: outbound::Backend, routes: impl Iterator, failure_accrual: Option, + load_bias: Option, + retry_after: Option, service_retry: Option>, service_timeouts: RouteTimeouts, allow_l5d_request_headers: bool, @@ -54,6 +56,8 @@ pub(crate) fn protocol( outbound::proxy_protocol::Kind::Grpc(outbound::proxy_protocol::Grpc { routes, failure_accrual, + load_bias, + retry_after, }) } @@ -230,6 +234,7 @@ fn convert_backend( )), }), load: Some(default_balancer_config()), + ejection: None, }, )), }), diff --git a/policy-controller/grpc/src/outbound/http.rs b/policy-controller/grpc/src/outbound/http.rs index 34ca459a03206..d711af9e57fed 100644 --- a/policy-controller/grpc/src/outbound/http.rs +++ b/policy-controller/grpc/src/outbound/http.rs @@ -21,6 +21,8 @@ pub(crate) fn protocol( default_backend: outbound::Backend, routes: impl Iterator, accrual: Option, + load_bias: Option, + retry_after: Option, service_retry: Option>, service_timeouts: RouteTimeouts, allow_l5d_request_headers: bool, @@ -76,10 +78,14 @@ pub(crate) fn protocol( http1: Some(outbound::proxy_protocol::Http1 { routes: routes.clone(), failure_accrual: accrual, + load_bias, + retry_after, }), http2: Some(outbound::proxy_protocol::Http2 { routes, failure_accrual: accrual, + load_bias, + retry_after, }), }) } @@ -89,6 +95,8 @@ pub(crate) fn http1_only_protocol( default_backend: outbound::Backend, routes: impl Iterator, accrual: Option, + load_bias: Option, + retry_after: Option, service_retry: Option>, service_timeouts: RouteTimeouts, allow_l5d_request_headers: bool, @@ -106,6 +114,8 @@ pub(crate) fn http1_only_protocol( original_dst, ), failure_accrual: accrual, + load_bias, + retry_after, }) } @@ -114,6 +124,8 @@ pub(crate) fn http2_only_protocol( default_backend: outbound::Backend, routes: impl Iterator, accrual: Option, + load_bias: Option, + retry_after: Option, service_retry: Option>, service_timeouts: RouteTimeouts, allow_l5d_request_headers: bool, @@ -131,6 +143,8 @@ pub(crate) fn http2_only_protocol( original_dst, ), failure_accrual: accrual, + load_bias, + retry_after, }) } @@ -318,6 +332,7 @@ fn convert_backend( )), }), load: Some(default_balancer_config()), + ejection: None, }, )), }), diff --git a/policy-controller/grpc/src/outbound/tcp.rs b/policy-controller/grpc/src/outbound/tcp.rs index afd1222136bdb..f1d2fcafac52c 100644 --- a/policy-controller/grpc/src/outbound/tcp.rs +++ b/policy-controller/grpc/src/outbound/tcp.rs @@ -133,6 +133,7 @@ fn convert_backend( )), }), load: Some(default_balancer_config()), + ejection: None, }, )), }), diff --git a/policy-controller/grpc/src/outbound/tls.rs b/policy-controller/grpc/src/outbound/tls.rs index 86d8a29c8cb7a..73016c2253069 100644 --- a/policy-controller/grpc/src/outbound/tls.rs +++ b/policy-controller/grpc/src/outbound/tls.rs @@ -135,6 +135,7 @@ fn convert_backend( )), }), load: Some(default_balancer_config()), + ejection: None, }, )), }), diff --git a/policy-controller/k8s/index/src/outbound/index.rs b/policy-controller/k8s/index/src/outbound/index.rs index 4e0f5b6a5ed2f..06ffcfb2f6d90 100644 --- a/policy-controller/k8s/index/src/outbound/index.rs +++ b/policy-controller/k8s/index/src/outbound/index.rs @@ -9,8 +9,10 @@ use egress_network::EgressNetwork; use linkerd_policy_controller_core::{ outbound::{ AppProtocol, Backend, Backoff, FailureAccrual, GrpcRetryCondition, GrpcRoute, - HttpRetryCondition, HttpRoute, Kind, OutboundDiscoverTarget, OutboundPolicy, ParentInfo, - ResourceTarget, RouteRetry, RouteSet, RouteTimeouts, TcpRoute, TlsRoute, TrafficPolicy, + HttpRetryCondition, HttpRoute, Kind, LoadBiasConfig, OutboundDiscoverTarget, + OutboundPolicy, ParentInfo, ResourceTarget, RetryAfterConfig, RouteRetry, RouteSet, + RouteTimeouts, TcpRoute, TlsRoute, TrafficPolicy, DEFAULT_LOAD_BIAS_PENALTY, + DEFAULT_LOAD_BIAS_PENALTY_DECAY, DEFAULT_RETRY_AFTER_MAX_DURATION, }, routes::GroupKindNamespaceName, }; @@ -101,6 +103,8 @@ struct Namespace { struct ResourceInfo { app_protocols: PortMap, accrual: Option, + load_bias: Option, + retry_after: Option, http_retry: Option>, grpc_retry: Option>, timeouts: RouteTimeouts, @@ -122,6 +126,8 @@ struct ResourceRoutes { watches_by_ns: HashMap, app_protocol: Option, accrual: Option, + load_bias: Option, + retry_after: Option, http_retry: Option>, grpc_retry: Option>, timeouts: RouteTimeouts, @@ -132,6 +138,8 @@ struct RoutesWatch { parent_info: ParentInfo, app_protocol: Option, accrual: Option, + load_bias: Option, + retry_after: Option, http_retry: Option>, grpc_retry: Option>, timeouts: RouteTimeouts, @@ -215,9 +223,17 @@ impl kubert::index::IndexNamespacedResource for Index { let name = service.name_unchecked(); let ns = service.namespace().expect("Service must have a namespace"); tracing::debug!(name, ns, "indexing service"); + // NB: The admission webhook only intercepts Route resources, not Services, so + // annotation parse errors here surface only as controller log warnings (not API rejections). let accrual = parse_accrual_config(service.annotations()) .map_err(|error| tracing::warn!(%error, service=name, namespace=ns, "Failed to parse accrual config")) .unwrap_or_default(); + let load_bias = parse_load_bias_config(service.annotations()) + .map_err(|error| tracing::warn!(%error, service=name, namespace=ns, "Failed to parse load bias config")) + .unwrap_or_default(); + let retry_after = parse_retry_after_config(service.annotations()) + .map_err(|error| tracing::warn!(%error, service=name, namespace=ns, "Failed to parse retry after config")) + .unwrap_or_default(); let mut app_protocols = service .spec @@ -321,6 +337,8 @@ impl kubert::index::IndexNamespacedResource for Index { let service_info = ResourceInfo { app_protocols, accrual, + load_bias, + retry_after, http_retry, grpc_retry, timeouts, @@ -380,6 +398,30 @@ impl kubert::index::IndexNamespacedResource for let accrual = parse_accrual_config(egress_network.annotations()) .map_err(|error| tracing::warn!(%error, service=name, namespace=ns, "Failed to parse accrual config")) .unwrap_or_default(); + // EgressNetwork uses Forward instead of Balancer, so load bias and + // retry-after have no effect. Warn if someone set them anyway. + let load_bias = None; + let retry_after = None; + if egress_network + .annotations() + .contains_key("balancer.alpha.linkerd.io/load-bias") + { + tracing::warn!( + service = name, + namespace = ns, + "load-bias annotation has no effect on EgressNetwork (no balancer)" + ); + } + if egress_network + .annotations() + .contains_key("balancer.alpha.linkerd.io/retry-after") + { + tracing::warn!( + service = name, + namespace = ns, + "retry-after annotation has no effect on EgressNetwork (no balancer)" + ); + } let opaque_ports = ports_annotation( egress_network.annotations(), "config.linkerd.io/opaque-ports", @@ -421,6 +463,8 @@ impl kubert::index::IndexNamespacedResource for let egress_network_info = ResourceInfo { app_protocols, accrual, + load_bias, + retry_after, http_retry, grpc_retry, timeouts, @@ -1252,6 +1296,8 @@ impl Namespace { resource_routes.update_resource( app_protocol, resource.accrual, + resource.load_bias, + resource.retry_after, resource.http_retry.clone(), resource.grpc_retry.clone(), resource.timeouts.clone(), @@ -1337,12 +1383,16 @@ impl Namespace { }; let mut app_protocol = None; let mut accrual = None; + let mut load_bias = None; + let mut retry_after = None; let mut http_retry = None; let mut grpc_retry = None; let mut timeouts = Default::default(); if let Some(resource) = resource_info.get(&resource_ref) { app_protocol = resource.app_protocols.get(&rp.port).cloned(); accrual = resource.accrual; + load_bias = resource.load_bias; + retry_after = resource.retry_after; http_retry = resource.http_retry.clone(); grpc_retry = resource.grpc_retry.clone(); timeouts = resource.timeouts.clone(); @@ -1383,6 +1433,8 @@ impl Namespace { parent_info, app_protocol, accrual, + load_bias, + retry_after, http_retry, grpc_retry, timeouts, @@ -1604,6 +1656,8 @@ impl ResourceRoutes { port: self.port, app_protocol: self.app_protocol.clone(), accrual: self.accrual, + load_bias: self.load_bias, + retry_after: self.retry_after, http_retry: self.http_retry.clone(), grpc_retry: self.grpc_retry.clone(), timeouts: self.timeouts.clone(), @@ -1622,6 +1676,8 @@ impl ResourceRoutes { watch: sender, app_protocol: self.app_protocol.clone(), accrual: self.accrual, + load_bias: self.load_bias, + retry_after: self.retry_after, http_retry: self.http_retry.clone(), grpc_retry: self.grpc_retry.clone(), timeouts: self.timeouts.clone(), @@ -1717,10 +1773,13 @@ impl ResourceRoutes { } } + #[allow(clippy::too_many_arguments)] fn update_resource( &mut self, app_protocol: Option, accrual: Option, + load_bias: Option, + retry_after: Option, http_retry: Option>, grpc_retry: Option>, timeouts: RouteTimeouts, @@ -1728,6 +1787,8 @@ impl ResourceRoutes { ) { self.app_protocol = app_protocol.clone(); self.accrual = accrual; + self.load_bias = load_bias; + self.retry_after = retry_after; self.http_retry = http_retry.clone(); self.grpc_retry = grpc_retry.clone(); self.timeouts = timeouts.clone(); @@ -1735,6 +1796,8 @@ impl ResourceRoutes { for watch in self.watches_by_ns.values_mut() { watch.app_protocol = app_protocol.clone(); watch.accrual = accrual; + watch.load_bias = load_bias; + watch.retry_after = retry_after; watch.http_retry = http_retry.clone(); watch.grpc_retry = grpc_retry.clone(); watch.timeouts = timeouts.clone(); @@ -1834,6 +1897,16 @@ impl RoutesWatch { modified = true; } + if self.load_bias != policy.load_bias { + policy.load_bias = self.load_bias; + modified = true; + } + + if self.retry_after != policy.retry_after { + policy.retry_after = self.retry_after; + modified = true; + } + if self.http_retry != policy.http_retry { policy.http_retry = self.http_retry.clone(); modified = true; @@ -1975,21 +2048,124 @@ pub fn parse_timeouts( }) } +pub fn parse_load_bias_config( + annotations: &std::collections::BTreeMap, +) -> Result> { + annotations + .get("balancer.alpha.linkerd.io/load-bias") + .and_then(|s| match s.trim() { + "false" => None, + mode => Some(mode), + }) + .map(|mode| { + if mode != "true" { + bail!("unsupported load-bias mode: '{mode}' (expected 'true' or 'false')"); + } + + let penalty = annotations + .get("balancer.alpha.linkerd.io/load-bias-penalty") + .map(|s| parse_duration(s)) + .transpose()? + .unwrap_or(DEFAULT_LOAD_BIAS_PENALTY); + + let penalty_decay = annotations + .get("balancer.alpha.linkerd.io/load-bias-penalty-decay") + .map(|s| parse_duration(s)) + .transpose()? + .unwrap_or(DEFAULT_LOAD_BIAS_PENALTY_DECAY); + + ensure!( + penalty > time::Duration::ZERO, + "load-bias penalty must be greater than zero" + ); + ensure!( + penalty_decay > time::Duration::ZERO, + "load-bias penalty_decay must be greater than zero" + ); + + Ok(LoadBiasConfig { + enabled: true, + penalty, + penalty_decay, + }) + }) + .transpose() +} + +pub fn parse_retry_after_config( + annotations: &std::collections::BTreeMap, +) -> Result> { + annotations + .get("balancer.alpha.linkerd.io/retry-after") + .and_then(|s| match s.trim() { + "false" => None, + mode => Some(mode), + }) + .map(|mode| { + if mode != "true" { + bail!("unsupported retry-after mode: '{mode}' (expected 'true' or 'false')"); + } + + let max_duration = annotations + .get("balancer.alpha.linkerd.io/retry-after-max-duration") + .map(|s| parse_duration(s)) + .transpose()? + .unwrap_or(DEFAULT_RETRY_AFTER_MAX_DURATION); + + ensure!( + max_duration > time::Duration::ZERO, + "retry-after-max-duration must be greater than zero" + ); + + Ok(RetryAfterConfig { max_duration }) + }) + .transpose() +} + fn parse_duration(s: &str) -> Result { let s = s.trim(); + if s.starts_with('-') { + bail!("duration value cannot be negative: '{s}'"); + } let offset = s .rfind(|c: char| c.is_ascii_digit()) .ok_or_else(|| anyhow::anyhow!("{s} does not contain a timeout duration value"))?; let (magnitude, unit) = s.split_at(offset + 1); + + if magnitude.contains('.') { + if unit == "s" { + let frac: f64 = magnitude + .parse() + .map_err(|_| anyhow::anyhow!("invalid fractional value {magnitude}"))?; + if frac == 0.0 { + bail!("fractional seconds not supported; use '0' for zero duration"); + } + if !frac.is_finite() || frac > (u64::MAX / 1000) as f64 { + bail!("duration value {s} overflows when converted to milliseconds"); + } + let ms = (frac * 1000.0).round() as u64; + if ms >= 1 { + bail!("fractional seconds not supported; try '{ms}ms' instead of '{s}'"); + } else { + bail!("{s} value is sub-millisecond; minimum resolution is 1ms"); + } + } else { + bail!("fractional values not supported for duration unit '{unit}'"); + } + } + let magnitude = magnitude.parse::()?; let mul = match unit { + // Special case: "0" is valid as zero duration without requiring a unit suffix. + // Non-zero bare numbers (ie. "5") require a unit. "" if magnitude == 0 => 0, "ms" => 1, "s" => 1000, "m" => 1000 * 60, "h" => 1000 * 60 * 60, "d" => 1000 * 60 * 60 * 24, + "" => bail!("missing duration unit; did you mean '{magnitude}s' or '{magnitude}ms'?"), _ => bail!("invalid duration unit {unit} (expected one of 'ms', 's', 'm', 'h', or 'd')"), }; @@ -1998,3 +2174,503 @@ fn parse_duration(s: &str) -> Result { .ok_or_else(|| anyhow::anyhow!("Timeout value {s} overflows when converted to 'ms'"))?; Ok(time::Duration::from_millis(ms)) } + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::BTreeMap; + + #[test] + fn parse_duration_integer_seconds() { + let d = parse_duration("10s").expect("should parse"); + assert_eq!(d, time::Duration::from_secs(10)); + } + + #[test] + fn parse_duration_milliseconds() { + let d = parse_duration("500ms").expect("should parse"); + assert_eq!(d, time::Duration::from_millis(500)); + } + + #[test] + fn parse_duration_minutes() { + let d = parse_duration("5m").expect("should parse"); + assert_eq!(d, time::Duration::from_secs(300)); + } + + #[test] + fn parse_duration_hours() { + let d = parse_duration("2h").expect("should parse"); + assert_eq!(d, time::Duration::from_secs(7200)); + } + + #[test] + fn parse_duration_days() { + let d = parse_duration("1d").expect("should parse"); + assert_eq!(d, time::Duration::from_secs(86400)); + } + + #[test] + fn parse_duration_zero() { + let d = parse_duration("0").expect("zero without unit should parse"); + assert_eq!(d, time::Duration::ZERO); + } + + #[test] + fn parse_duration_zero_seconds() { + let d = parse_duration("0s").expect("0s should parse"); + assert_eq!(d, time::Duration::ZERO); + } + + #[test] + fn parse_duration_zero_milliseconds() { + let d = parse_duration("0ms").expect("0ms should parse"); + assert_eq!(d, time::Duration::ZERO); + } + + #[test] + fn parse_duration_fractional_seconds_rejected() { + let err = parse_duration("0.5s").expect_err("fractional seconds should fail"); + assert!( + err.to_string().contains("500ms"), + "should suggest ms equivalent: {err}" + ); + assert!( + err.to_string().contains("fractional seconds not supported"), + "should explain the issue: {err}" + ); + } + + #[test] + fn parse_duration_fractional_zero_seconds_rejected() { + let err = parse_duration("0.0s").expect_err("fractional zero should fail"); + assert!( + err.to_string().contains("use '0' for zero duration"), + "should suggest bare 0: {err}" + ); + } + + #[test] + fn parse_duration_fractional_non_seconds_rejected() { + let err = parse_duration("0.5m").expect_err("fractional minutes should fail"); + assert!( + err.to_string().contains("fractional values not supported"), + "should reject fractional: {err}" + ); + } + + #[test] + fn parse_duration_bare_number_rejected() { + let err = parse_duration("5").expect_err("bare number should fail"); + assert!( + err.to_string().contains("missing duration unit"), + "should mention missing unit: {err}" + ); + } + + #[test] + fn parse_duration_bare_number_error_suggests_units() { + let err = parse_duration("100").expect_err("bare number should fail"); + assert!( + err.to_string().contains("'100s' or '100ms'"), + "should suggest likely units: {err}" + ); + } + + #[test] + fn parse_duration_overflow_rejected() { + let err = parse_duration("999999999999999999d").expect_err("huge value should overflow"); + assert!( + err.to_string().contains("overflows"), + "should mention overflow: {err}" + ); + } + + #[test] + fn parse_duration_invalid_unit_rejected() { + let err = parse_duration("10x").expect_err("invalid unit should fail"); + assert!( + err.to_string().contains("invalid duration unit"), + "should mention invalid unit: {err}" + ); + } + + #[test] + fn parse_duration_empty_rejected() { + let result = parse_duration(""); + assert!(result.is_err(), "empty string should fail"); + } + + #[test] + fn parse_duration_negative_seconds_rejected() { + let err = parse_duration("-5s").expect_err("negative should fail"); + assert!( + err.to_string().contains("cannot be negative"), + "should mention negative: {err}" + ); + } + + #[test] + fn parse_duration_negative_fractional_rejected() { + let err = parse_duration("-0.5s").expect_err("negative fractional should fail"); + assert!( + err.to_string().contains("cannot be negative"), + "should mention negative: {err}" + ); + } + + #[test] + fn load_bias_true_returns_defaults() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "true".to_string(), + ); + let config = parse_load_bias_config(&annotations) + .expect("mode=true should succeed") + .expect("should return Some"); + assert!(config.enabled); + assert_eq!(config.penalty, DEFAULT_LOAD_BIAS_PENALTY); + assert_eq!(config.penalty_decay, DEFAULT_LOAD_BIAS_PENALTY_DECAY); + } + + #[test] + fn load_bias_false_returns_none() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "false".to_string(), + ); + let result = parse_load_bias_config(&annotations).expect("mode=false should succeed"); + assert!(result.is_none(), "mode=false should return None"); + } + + #[test] + fn load_bias_absent_returns_none() { + let annotations = BTreeMap::new(); + let result = + parse_load_bias_config(&annotations).expect("absent annotation should succeed"); + assert!(result.is_none(), "absent annotation should return None"); + } + + #[test] + fn load_bias_invalid_mode_rejected() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "maybe".to_string(), + ); + let err = parse_load_bias_config(&annotations).expect_err("mode=maybe should be rejected"); + assert!( + err.to_string().contains("'maybe'"), + "error should quote the invalid mode: {err}" + ); + } + + #[test] + fn load_bias_empty_mode_rejected_with_visible_quotes() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "".to_string(), + ); + let err = parse_load_bias_config(&annotations).expect_err("empty mode should be rejected"); + let msg = err.to_string(); + assert!( + msg.contains("''"), + "error should show empty value in quotes: {msg}" + ); + } + + #[test] + fn load_bias_custom_penalty_and_decay() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias-penalty".to_string(), + "3s".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias-penalty-decay".to_string(), + "7s".to_string(), + ); + let config = parse_load_bias_config(&annotations) + .expect("custom values should succeed") + .expect("should return Some"); + assert_eq!(config.penalty, time::Duration::from_secs(3)); + assert_eq!(config.penalty_decay, time::Duration::from_secs(7)); + } + + #[test] + fn load_bias_invalid_penalty_duration_rejected() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias-penalty".to_string(), + "notaduration".to_string(), + ); + let result = parse_load_bias_config(&annotations); + assert!(result.is_err(), "invalid penalty duration should fail"); + } + + #[test] + fn load_bias_zero_penalty_rejected() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias-penalty".to_string(), + "0".to_string(), + ); + let err = + parse_load_bias_config(&annotations).expect_err("zero penalty should be rejected"); + assert!( + err.to_string().contains("greater than zero"), + "error should mention 'greater than zero': {err}" + ); + } + + #[test] + fn load_bias_zero_penalty_decay_rejected() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias-penalty-decay".to_string(), + "0".to_string(), + ); + let err = parse_load_bias_config(&annotations) + .expect_err("zero penalty_decay should be rejected"); + assert!( + err.to_string().contains("greater than zero"), + "error should mention 'greater than zero': {err}" + ); + } + + #[test] + fn load_bias_whitespace_true_accepted() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + " true ".to_string(), + ); + let config = parse_load_bias_config(&annotations) + .expect("whitespace-padded 'true' should succeed") + .expect("should return Some"); + assert!(config.enabled); + } + + #[test] + fn load_bias_whitespace_false_returns_none() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + " false ".to_string(), + ); + let result = + parse_load_bias_config(&annotations).expect("whitespace-padded 'false' should succeed"); + assert!(result.is_none()); + } + + #[test] + fn retry_after_true_returns_defaults() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after".to_string(), + "true".to_string(), + ); + let config = parse_retry_after_config(&annotations) + .expect("mode=true should succeed") + .expect("should return Some"); + assert_eq!(config.max_duration, DEFAULT_RETRY_AFTER_MAX_DURATION); + } + + #[test] + fn retry_after_false_returns_none() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after".to_string(), + "false".to_string(), + ); + let result = parse_retry_after_config(&annotations).expect("mode=false should succeed"); + assert!(result.is_none(), "mode=false should return None"); + } + + #[test] + fn retry_after_absent_returns_none() { + let annotations = BTreeMap::new(); + let result = + parse_retry_after_config(&annotations).expect("absent annotation should succeed"); + assert!(result.is_none(), "absent annotation should return None"); + } + + #[test] + fn retry_after_whitespace_true_accepted() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after".to_string(), + " true ".to_string(), + ); + let config = parse_retry_after_config(&annotations) + .expect("whitespace-padded 'true' should succeed") + .expect("should return Some"); + assert_eq!(config.max_duration, DEFAULT_RETRY_AFTER_MAX_DURATION); + } + + #[test] + fn retry_after_invalid_mode_rejected() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after".to_string(), + "sometimes".to_string(), + ); + let err = + parse_retry_after_config(&annotations).expect_err("mode=sometimes should be rejected"); + assert!( + err.to_string().contains("'sometimes'"), + "error should quote the invalid mode: {err}" + ); + } + + #[test] + fn retry_after_custom_max_duration() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after-max-duration".to_string(), + "60s".to_string(), + ); + let config = parse_retry_after_config(&annotations) + .expect("custom max should succeed") + .expect("should return Some"); + assert_eq!(config.max_duration, time::Duration::from_secs(60)); + } + + #[test] + fn retry_after_zero_max_duration_rejected() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after-max-duration".to_string(), + "0s".to_string(), + ); + let err = parse_retry_after_config(&annotations) + .expect_err("zero max_duration should be rejected"); + assert!( + err.to_string().contains("greater than zero"), + "error should mention 'greater than zero': {err}" + ); + } + + #[test] + fn load_bias_custom_decay_default_penalty() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias-penalty-decay".to_string(), + "20s".to_string(), + ); + let config = parse_load_bias_config(&annotations) + .expect("custom decay with default penalty should succeed") + .expect("should return Some"); + assert_eq!(config.penalty, DEFAULT_LOAD_BIAS_PENALTY); + assert_eq!(config.penalty_decay, time::Duration::from_secs(20)); + } + + #[test] + fn load_bias_false_ignores_sub_annotations() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "false".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias-penalty".to_string(), + "3s".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias-penalty-decay".to_string(), + "7s".to_string(), + ); + let result = parse_load_bias_config(&annotations).expect("mode=false should succeed"); + assert!( + result.is_none(), + "mode=false should return None even with sub-annotations" + ); + } + + #[test] + fn retry_after_false_ignores_sub_annotations() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after".to_string(), + "false".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after-max-duration".to_string(), + "60s".to_string(), + ); + let result = parse_retry_after_config(&annotations).expect("mode=false should succeed"); + assert!( + result.is_none(), + "mode=false should return None even with sub-annotations" + ); + } + + #[test] + fn load_bias_negative_penalty_rejected() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias-penalty".to_string(), + "-5s".to_string(), + ); + let err = + parse_load_bias_config(&annotations).expect_err("negative penalty should be rejected"); + assert!( + err.to_string().contains("cannot be negative"), + "error should mention negative: {err}" + ); + } + + #[test] + fn retry_after_negative_max_duration_rejected() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after-max-duration".to_string(), + "-10s".to_string(), + ); + let err = parse_retry_after_config(&annotations) + .expect_err("negative max_duration should be rejected"); + assert!( + err.to_string().contains("cannot be negative"), + "error should mention negative: {err}" + ); + } +} diff --git a/policy-controller/runtime/src/admission.rs b/policy-controller/runtime/src/admission.rs index 7c552c274cc8e..f7acbfd91d230 100644 --- a/policy-controller/runtime/src/admission.rs +++ b/policy-controller/runtime/src/admission.rs @@ -495,11 +495,16 @@ impl Validate for Admission { } } + // Validate balancer annotations on the Route itself. Annotations set directly + // on Services are only parsed at indexer time (not admission-validated) because + // the ValidatingWebhookConfiguration does not intercept core/v1 Services. if spec.parent_refs.iter().flatten().any(|parent| { outbound_index::is_parent_service_or_egress_network(&parent.kind, &parent.group) }) { outbound_index::http::parse_http_retry(annotations)?; outbound_index::parse_accrual_config(annotations)?; + outbound_index::parse_load_bias_config(annotations)?; + outbound_index::parse_retry_after_config(annotations)?; outbound_index::parse_timeouts(annotations)?; } @@ -619,11 +624,14 @@ impl Validate for Admission { } } + // See note in Validate: only Route annotations are validated here if spec.parent_refs.iter().flatten().any(|parent| { outbound_index::is_parent_service_or_egress_network(&parent.kind, &parent.group) }) { outbound_index::http::parse_http_retry(annotations)?; outbound_index::parse_accrual_config(annotations)?; + outbound_index::parse_load_bias_config(annotations)?; + outbound_index::parse_retry_after_config(annotations)?; outbound_index::parse_timeouts(annotations)?; } @@ -689,11 +697,14 @@ impl Validate for Admission { } } + // See note in Validate: only Route annotations are validated here if spec.parent_refs.iter().flatten().any(|parent| { outbound_index::is_parent_service_or_egress_network(&parent.kind, &parent.group) }) { outbound_index::grpc::parse_grpc_retry(annotations)?; outbound_index::parse_accrual_config(annotations)?; + outbound_index::parse_load_bias_config(annotations)?; + outbound_index::parse_retry_after_config(annotations)?; outbound_index::parse_timeouts(annotations)?; } @@ -853,3 +864,331 @@ impl Validate for Admission { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + + /// Constructs an HttpRouteSpec with a single Service parent ref, + /// which causes validate() to parse annotations. + fn service_parent_spec() -> HttpRouteSpec { + HttpRouteSpec { + parent_refs: Some(vec![gateway::HTTPRouteParentRefs { + name: "test-svc".to_string(), + namespace: Some("test-ns".to_string()), + kind: Some("Service".to_string()), + group: Some("core".to_string()), + section_name: None, + port: None, + }]), + hostnames: None, + rules: None, + } + } + + fn gateway_http_service_parent_spec() -> gateway::HTTPRouteSpec { + gateway::HTTPRouteSpec { + parent_refs: Some(vec![gateway::HTTPRouteParentRefs { + name: "test-svc".to_string(), + namespace: Some("test-ns".to_string()), + kind: Some("Service".to_string()), + group: Some("core".to_string()), + section_name: None, + port: None, + }]), + hostnames: None, + rules: None, + } + } + + fn grpc_service_parent_spec() -> gateway::GRPCRouteSpec { + gateway::GRPCRouteSpec { + parent_refs: Some(vec![gateway::GRPCRouteParentRefs { + name: "test-svc".to_string(), + namespace: Some("test-ns".to_string()), + kind: Some("Service".to_string()), + group: Some("core".to_string()), + section_name: None, + port: None, + }]), + hostnames: None, + rules: None, + } + } + + #[tokio::test(flavor = "current_thread")] + async fn admission_valid_load_bias() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias-penalty".to_string(), + "10s".to_string(), + ); + + let admission = Admission::new(); + let result = admission + .validate("test-ns", "test-route", &annotations, service_parent_spec()) + .await; + result.expect("valid load-bias should be accepted"); + } + + #[tokio::test(flavor = "current_thread")] + async fn admission_invalid_load_bias_mode() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "invalid".to_string(), + ); + + let admission = Admission::new(); + let result = admission + .validate("test-ns", "test-route", &annotations, service_parent_spec()) + .await; + let err = result.expect_err("invalid load-bias mode should be rejected"); + assert!( + err.to_string().contains("load-bias"), + "error should mention load-bias: {err}" + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn admission_valid_retry_after() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after-max-duration".to_string(), + "120s".to_string(), + ); + + let admission = Admission::new(); + let result = admission + .validate("test-ns", "test-route", &annotations, service_parent_spec()) + .await; + result.expect("valid retry-after should be accepted"); + } + + #[tokio::test(flavor = "current_thread")] + async fn admission_invalid_retry_after_duration() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after-max-duration".to_string(), + "5".to_string(), + ); + + let admission = Admission::new(); + let result = admission + .validate("test-ns", "test-route", &annotations, service_parent_spec()) + .await; + let err = result.expect_err("bare number duration should be rejected"); + assert!( + err.to_string().contains("missing duration unit"), + "error should mention missing unit: {err}" + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn admission_gateway_httproute_valid_load_bias() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "true".to_string(), + ); + + let admission = Admission::new(); + let result = admission + .validate( + "test-ns", + "test-route", + &annotations, + gateway_http_service_parent_spec(), + ) + .await; + result.expect("valid load-bias on gateway HTTPRoute should be accepted"); + } + + #[tokio::test(flavor = "current_thread")] + async fn admission_gateway_httproute_invalid_load_bias() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "invalid".to_string(), + ); + + let admission = Admission::new(); + let result = admission + .validate( + "test-ns", + "test-route", + &annotations, + gateway_http_service_parent_spec(), + ) + .await; + let err = result.expect_err("invalid load-bias on gateway HTTPRoute should be rejected"); + assert!( + err.to_string().contains("load-bias"), + "error should mention load-bias: {err}" + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn admission_gateway_httproute_valid_retry_after() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after-max-duration".to_string(), + "120s".to_string(), + ); + + let admission = Admission::new(); + let result = admission + .validate( + "test-ns", + "test-route", + &annotations, + gateway_http_service_parent_spec(), + ) + .await; + result.expect("valid retry-after on gateway HTTPRoute should be accepted"); + } + + #[tokio::test(flavor = "current_thread")] + async fn admission_gateway_httproute_invalid_retry_after() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after-max-duration".to_string(), + "5".to_string(), + ); + + let admission = Admission::new(); + let result = admission + .validate( + "test-ns", + "test-route", + &annotations, + gateway_http_service_parent_spec(), + ) + .await; + let err = result.expect_err("bare number duration on gateway HTTPRoute should be rejected"); + assert!( + err.to_string().contains("missing duration unit"), + "error should mention missing unit: {err}" + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn admission_grpcroute_valid_load_bias() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias-penalty".to_string(), + "10s".to_string(), + ); + + let admission = Admission::new(); + let result = admission + .validate( + "test-ns", + "test-route", + &annotations, + grpc_service_parent_spec(), + ) + .await; + result.expect("valid load-bias on GRPCRoute should be accepted"); + } + + #[tokio::test(flavor = "current_thread")] + async fn admission_grpcroute_invalid_load_bias() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/load-bias".to_string(), + "invalid".to_string(), + ); + + let admission = Admission::new(); + let result = admission + .validate( + "test-ns", + "test-route", + &annotations, + grpc_service_parent_spec(), + ) + .await; + let err = result.expect_err("invalid load-bias on GRPCRoute should be rejected"); + assert!( + err.to_string().contains("load-bias"), + "error should mention load-bias: {err}" + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn admission_grpcroute_valid_retry_after() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after-max-duration".to_string(), + "60s".to_string(), + ); + + let admission = Admission::new(); + let result = admission + .validate( + "test-ns", + "test-route", + &annotations, + grpc_service_parent_spec(), + ) + .await; + result.expect("valid retry-after on GRPCRoute should be accepted"); + } + + #[tokio::test(flavor = "current_thread")] + async fn admission_grpcroute_invalid_retry_after() { + let mut annotations = BTreeMap::new(); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after".to_string(), + "true".to_string(), + ); + annotations.insert( + "balancer.alpha.linkerd.io/retry-after-max-duration".to_string(), + "0s".to_string(), + ); + + let admission = Admission::new(); + let result = admission + .validate( + "test-ns", + "test-route", + &annotations, + grpc_service_parent_spec(), + ) + .await; + let err = result.expect_err("zero max_duration on GRPCRoute should be rejected"); + assert!( + err.to_string().contains("greater than zero"), + "error should mention 'greater than zero': {err}" + ); + } +} diff --git a/policy-test/src/outbound_api.rs b/policy-test/src/outbound_api.rs index e4db46bb986c3..581d3f3b5db20 100644 --- a/policy-test/src/outbound_api.rs +++ b/policy-test/src/outbound_api.rs @@ -70,6 +70,8 @@ pub fn http1_routes(config: &grpc::outbound::OutboundPolicy) -> &[grpc::outbound if let grpc::outbound::proxy_protocol::Kind::Http1(grpc::outbound::proxy_protocol::Http1 { routes, failure_accrual: _, + load_bias: _, + retry_after: _, }) = kind { routes @@ -90,6 +92,8 @@ pub fn http2_routes(config: &grpc::outbound::OutboundPolicy) -> &[grpc::outbound if let grpc::outbound::proxy_protocol::Kind::Http2(grpc::outbound::proxy_protocol::Http2 { routes, failure_accrual: _, + load_bias: _, + retry_after: _, }) = kind { routes @@ -110,6 +114,8 @@ pub fn grpc_routes(config: &grpc::outbound::OutboundPolicy) -> &[grpc::outbound: if let grpc::outbound::proxy_protocol::Kind::Grpc(grpc::outbound::proxy_protocol::Grpc { routes, failure_accrual: _, + load_bias: _, + retry_after: _, }) = kind { routes @@ -194,17 +200,52 @@ where pub fn failure_accrual_consecutive( accrual: Option<&grpc::outbound::FailureAccrual>, ) -> &grpc::outbound::failure_accrual::ConsecutiveFailures { - assert!( - accrual.is_some(), - "failure accrual must be configured for service" - ); - let kind = accrual - .unwrap() + accrual + .expect("failure accrual must be configured for service") + .consecutive_failures + .as_ref() + .expect("failure accrual must have consecutive failures") +} + +/// Extract load_bias and retry_after from a Detect protocol config. +#[track_caller] +pub fn detect_load_bias_and_retry_after( + config: &grpc::outbound::OutboundPolicy, +) -> ( + Option<&grpc::outbound::LoadBiasConfig>, + Option<&grpc::outbound::RetryAfterConfig>, +) { + let kind = config + .protocol + .as_ref() + .expect("must have proxy protocol") .kind .as_ref() - .expect("failure accrual must have kind"); - let grpc::outbound::failure_accrual::Kind::ConsecutiveFailures(accrual) = kind; - accrual + .expect("must have kind"); + if let grpc::outbound::proxy_protocol::Kind::Detect(grpc::outbound::proxy_protocol::Detect { + http1, + http2, + .. + }) = kind + { + let http1 = http1 + .as_ref() + .expect("proxy protocol must have http1 field"); + let http2 = http2 + .as_ref() + .expect("proxy protocol must have http2 field"); + assert_eq!( + http1.load_bias, http2.load_bias, + "http1 and http2 load_bias configs must match" + ); + assert_eq!( + http1.retry_after, http2.retry_after, + "http1 and http2 retry_after configs must match" + ); + (http1.load_bias.as_ref(), http1.retry_after.as_ref()) + } else { + panic!("proxy protocol must be Detect; actually got:\n{kind:#?}") + } } #[track_caller] diff --git a/policy-test/tests/outbound_api_failure_accrual.rs b/policy-test/tests/outbound_api_failure_accrual.rs index 26ea11c33bb0e..ed2283246011b 100644 --- a/policy-test/tests/outbound_api_failure_accrual.rs +++ b/policy-test/tests/outbound_api_failure_accrual.rs @@ -1,14 +1,16 @@ use std::time::Duration; use futures::StreamExt; +use kube::Resource; use linkerd_policy_controller_k8s_api::{self as k8s, policy}; use linkerd_policy_test::{ assert_default_accrual_backoff, assert_resource_meta, create, grpc, outbound_api::{ - detect_failure_accrual, failure_accrual_consecutive, retry_watch_outbound_policy, + detect_failure_accrual, detect_load_bias_and_retry_after, failure_accrual_consecutive, + retry_watch_outbound_policy, }, test_route::TestParent, - with_temp_ns, + update, with_temp_ns, }; use maplit::btreemap; @@ -242,3 +244,481 @@ async fn default_failure_accrual() { test::().await; test::().await; } + +#[tokio::test(flavor = "current_thread")] +async fn load_bias_with_custom_penalty() { + async fn test() { + with_temp_ns(|client, ns| async move { + let port = 4191; + let mut parent = P::make_parent(&ns); + parent.meta_mut().annotations = Some(btreemap! { + "balancer.alpha.linkerd.io/load-bias".to_string() => "true".to_string(), + "balancer.alpha.linkerd.io/load-bias-penalty".to_string() => "3s".to_string(), + "balancer.alpha.linkerd.io/load-bias-penalty-decay".to_string() => "6s".to_string(), + }); + let parent = create(&client, parent).await; + + let mut rx = retry_watch_outbound_policy(&client, &ns, parent.ip(), port).await; + let config = rx + .next() + .await + .expect("watch must not fail") + .expect("watch must return an initial config"); + tracing::trace!(?config); + + let dt = P::DynamicType::default(); + let (load_bias, _) = detect_load_bias_and_retry_after(&config); + if P::kind(&dt) == "EgressNetwork" { + assert!( + load_bias.is_none(), + "EgressNetwork should not have load_bias" + ); + } else { + let lb = load_bias.expect("load_bias must be configured"); + assert!(lb.enabled, "load_bias must be enabled"); + assert_eq!( + lb.penalty, + Some(Duration::from_secs(3).try_into().unwrap()), + "penalty should be 3s" + ); + assert_eq!( + lb.penalty_decay, + Some(Duration::from_secs(6).try_into().unwrap()), + "penalty_decay should be 6s" + ); + } + }) + .await; + } + + test::().await; + test::().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn retry_after_with_custom_max() { + async fn test() { + with_temp_ns(|client, ns| async move { + let port = 4191; + let mut parent = P::make_parent(&ns); + parent.meta_mut().annotations = Some(btreemap! { + "balancer.alpha.linkerd.io/retry-after".to_string() => "true".to_string(), + "balancer.alpha.linkerd.io/retry-after-max-duration".to_string() => "120s".to_string(), + }); + let parent = create(&client, parent).await; + + let mut rx = retry_watch_outbound_policy(&client, &ns, parent.ip(), port).await; + let config = rx + .next() + .await + .expect("watch must not fail") + .expect("watch must return an initial config"); + tracing::trace!(?config); + + let dt = P::DynamicType::default(); + let (_, retry_after) = detect_load_bias_and_retry_after(&config); + if P::kind(&dt) == "EgressNetwork" { + assert!(retry_after.is_none(), "EgressNetwork should not have retry_after"); + } else { + let ra = retry_after.expect("retry_after must be configured"); + assert_eq!( + ra.max_duration, + Some(Duration::from_secs(120).try_into().unwrap()), + "max_duration should be 120s" + ); + } + }) + .await; + } + + test::().await; + test::().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn combined_load_bias_and_retry_after() { + async fn test() { + with_temp_ns(|client, ns| async move { + let port = 4191; + let mut parent = P::make_parent(&ns); + parent.meta_mut().annotations = Some(btreemap! { + "balancer.alpha.linkerd.io/load-bias".to_string() => "true".to_string(), + "balancer.alpha.linkerd.io/retry-after".to_string() => "true".to_string(), + }); + let parent = create(&client, parent).await; + + let mut rx = retry_watch_outbound_policy(&client, &ns, parent.ip(), port).await; + let config = rx + .next() + .await + .expect("watch must not fail") + .expect("watch must return an initial config"); + tracing::trace!(?config); + + let dt = P::DynamicType::default(); + let (load_bias, retry_after) = detect_load_bias_and_retry_after(&config); + if P::kind(&dt) == "EgressNetwork" { + assert!( + load_bias.is_none(), + "EgressNetwork should not have load_bias" + ); + assert!( + retry_after.is_none(), + "EgressNetwork should not have retry_after" + ); + } else { + let lb = load_bias.expect("load_bias must be configured"); + assert!(lb.enabled, "load_bias must be enabled"); + assert_eq!( + lb.penalty, + Some(Duration::from_secs(5).try_into().unwrap()), + "default penalty should be 5s" + ); + assert_eq!( + lb.penalty_decay, + Some(Duration::from_secs(10).try_into().unwrap()), + "default penalty_decay should be 10s" + ); + + let ra = retry_after.expect("retry_after must be configured"); + assert_eq!( + ra.max_duration, + Some(Duration::from_secs(300).try_into().unwrap()), + "default max_duration should be 300s" + ); + } + }) + .await; + } + + test::().await; + test::().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn accrual_with_load_bias_and_retry_after() { + async fn test() { + with_temp_ns(|client, ns| async move { + let port = 4191; + let mut parent = P::make_parent(&ns); + parent.meta_mut().annotations = Some(btreemap! { + "balancer.linkerd.io/failure-accrual".to_string() => "consecutive".to_string(), + "balancer.alpha.linkerd.io/load-bias".to_string() => "true".to_string(), + "balancer.alpha.linkerd.io/retry-after".to_string() => "true".to_string(), + }); + let parent = create(&client, parent).await; + + let mut rx = retry_watch_outbound_policy(&client, &ns, parent.ip(), port).await; + let config = rx + .next() + .await + .expect("watch must not fail") + .expect("watch must return an initial config"); + tracing::trace!(?config); + + detect_failure_accrual(&config, |accrual| { + let _consecutive = failure_accrual_consecutive(accrual); + }); + + let dt = P::DynamicType::default(); + let (load_bias, retry_after) = detect_load_bias_and_retry_after(&config); + if P::kind(&dt) == "EgressNetwork" { + assert!( + load_bias.is_none(), + "EgressNetwork should not have load_bias" + ); + assert!( + retry_after.is_none(), + "EgressNetwork should not have retry_after" + ); + } else { + let lb = load_bias.expect("load_bias must be configured"); + assert!(lb.enabled, "load_bias must be enabled"); + assert_eq!( + lb.penalty, + Some(Duration::from_secs(5).try_into().unwrap()), + "default penalty should be 5s" + ); + assert_eq!( + lb.penalty_decay, + Some(Duration::from_secs(10).try_into().unwrap()), + "default penalty_decay should be 10s" + ); + + let ra = retry_after.expect("retry_after must be configured"); + assert_eq!( + ra.max_duration, + Some(Duration::from_secs(300).try_into().unwrap()), + "default max_duration should be 300s" + ); + } + }) + .await; + } + + test::().await; + test::().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn invalid_load_bias_mode_produces_default() { + async fn test() { + with_temp_ns(|client, ns| async move { + let port = 4191; + let mut parent = P::make_parent(&ns); + parent.meta_mut().annotations = Some(btreemap! { + "balancer.alpha.linkerd.io/load-bias".to_string() => "invalid".to_string(), + }); + let parent = create(&client, parent).await; + + let mut rx = retry_watch_outbound_policy(&client, &ns, parent.ip(), port).await; + let config = rx + .next() + .await + .expect("watch must not fail") + .expect("watch must return an initial config"); + tracing::trace!(?config); + + // Invalid mode value causes a parse error. The indexer logs + // a warning and falls through to the default (no load bias). + let (load_bias, _) = detect_load_bias_and_retry_after(&config); + assert!( + load_bias.is_none(), + "invalid load-bias mode should produce no load_bias config" + ); + }) + .await; + } + + test::().await; + test::().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn invalid_retry_after_duration_produces_default() { + async fn test() { + with_temp_ns(|client, ns| async move { + let port = 4191; + let mut parent = P::make_parent(&ns); + parent.meta_mut().annotations = Some(btreemap! { + "balancer.alpha.linkerd.io/retry-after".to_string() => "true".to_string(), + "balancer.alpha.linkerd.io/retry-after-max-duration".to_string() => "5".to_string(), + }); + let parent = create(&client, parent).await; + + let mut rx = retry_watch_outbound_policy(&client, &ns, parent.ip(), port).await; + let config = rx + .next() + .await + .expect("watch must not fail") + .expect("watch must return an initial config"); + tracing::trace!(?config); + + // Bare number "5" lacks a duration unit, causing a parse error. + // The indexer logs a warning and falls through to the default. + let (_, retry_after) = detect_load_bias_and_retry_after(&config); + assert!( + retry_after.is_none(), + "bare-number duration should produce no retry_after config" + ); + }) + .await; + } + + test::().await; + test::().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn unannotated_service_has_no_new_config() { + async fn test() { + with_temp_ns(|client, ns| async move { + let port = 4191; + let parent = P::make_parent(&ns); + // No balancer annotations at all -- backwards compatibility test. + let parent = create(&client, parent).await; + + let mut rx = retry_watch_outbound_policy(&client, &ns, parent.ip(), port).await; + let config = rx + .next() + .await + .expect("watch must not fail") + .expect("watch must return an initial config"); + tracing::trace!(?config); + + // No annotations means no failure accrual, load bias, or retry + // after config -- zero behavior change for unannotated resources. + detect_failure_accrual(&config, |accrual| { + assert!( + accrual.is_none(), + "unannotated resource should have no failure accrual" + ); + }); + let (load_bias, retry_after) = detect_load_bias_and_retry_after(&config); + assert!( + load_bias.is_none(), + "unannotated resource should have no load_bias" + ); + assert!( + retry_after.is_none(), + "unannotated resource should have no retry_after" + ); + }) + .await; + } + + test::().await; + test::().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn egress_network_ignores_load_bias_and_retry_after() { + async fn test() { + with_temp_ns(|client, ns| async move { + let port = 4191; + let mut parent = P::make_parent(&ns); + parent.meta_mut().annotations = Some(btreemap! { + "balancer.alpha.linkerd.io/load-bias".to_string() => "true".to_string(), + "balancer.alpha.linkerd.io/load-bias-penalty".to_string() => "3s".to_string(), + "balancer.alpha.linkerd.io/retry-after".to_string() => "true".to_string(), + "balancer.alpha.linkerd.io/retry-after-max-duration".to_string() => "60s".to_string(), + }); + let parent = create(&client, parent).await; + + let mut rx = retry_watch_outbound_policy(&client, &ns, parent.ip(), port).await; + let config = rx + .next() + .await + .expect("watch must not fail") + .expect("watch must return an initial config"); + tracing::trace!(?config); + + // EgressNetworks use Forward instead of Balancer, so the + // indexer skips load-bias and retry-after even when annotated. + let dt = P::DynamicType::default(); + let kind = P::kind(&dt); + let (load_bias, retry_after) = detect_load_bias_and_retry_after(&config); + assert!( + load_bias.is_none(), + "{kind} should not have load_bias" + ); + assert!( + retry_after.is_none(), + "{kind} should not have retry_after" + ); + }) + .await; + } + + test::().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn consecutive_accrual_pipeline_unchanged() { + async fn test() { + with_temp_ns(|client, ns| async move { + let port = 4191; + let mut parent = P::make_parent(&ns); + parent.meta_mut().annotations = Some(btreemap! { + "balancer.linkerd.io/failure-accrual".to_string() => "consecutive".to_string(), + }); + let parent = create(&client, parent).await; + + let mut rx = retry_watch_outbound_policy(&client, &ns, parent.ip(), port).await; + let config = rx + .next() + .await + .expect("watch must not fail") + .expect("watch must return an initial config"); + tracing::trace!(?config); + + // Consecutive mode should produce failure accrual with + // consecutive_failures but no success_rate. + detect_failure_accrual(&config, |accrual| { + let accrual = accrual.expect("failure accrual must be configured"); + assert!( + accrual.consecutive_failures.is_some(), + "consecutive_failures must be present" + ); + assert!( + accrual.success_rate.is_none(), + "success_rate must NOT be set in consecutive mode" + ); + }); + + // Consecutive mode should not enable load bias or retry after. + let (load_bias, retry_after) = detect_load_bias_and_retry_after(&config); + assert!( + load_bias.is_none(), + "consecutive mode should not enable load_bias" + ); + assert!( + retry_after.is_none(), + "consecutive mode should not enable retry_after" + ); + }) + .await; + } + + test::().await; + test::().await; +} + +#[tokio::test(flavor = "current_thread")] +async fn load_bias_watch_update() { + with_temp_ns(|client, ns| async move { + let port = 4191; + let parent = k8s::Service::make_parent(&ns); + let mut parent = create(&client, parent).await; + + let mut rx = retry_watch_outbound_policy(&client, &ns, parent.ip(), port).await; + let config = rx + .next() + .await + .expect("watch must not fail") + .expect("watch must return an initial config"); + tracing::trace!(?config); + + let (load_bias, retry_after) = detect_load_bias_and_retry_after(&config); + assert!( + load_bias.is_none(), + "unannotated service should have no load_bias" + ); + assert!( + retry_after.is_none(), + "unannotated service should have no retry_after" + ); + + parent.meta_mut().annotations = Some(btreemap! { + "balancer.alpha.linkerd.io/load-bias".to_string() => "true".to_string(), + "balancer.alpha.linkerd.io/load-bias-penalty".to_string() => "3s".to_string(), + "balancer.alpha.linkerd.io/retry-after".to_string() => "true".to_string(), + "balancer.alpha.linkerd.io/retry-after-max-duration".to_string() => "60s".to_string(), + }); + update(&client, parent).await; + + let config = rx + .next() + .await + .expect("watch must not fail") + .expect("watch must return an updated config"); + tracing::trace!(?config); + + let (load_bias, retry_after) = detect_load_bias_and_retry_after(&config); + let lb = load_bias.expect("load_bias must be present after update"); + assert!(lb.enabled, "load_bias must be enabled"); + assert_eq!( + lb.penalty, + Some(Duration::from_secs(3).try_into().unwrap()), + "penalty should be 3s" + ); + + let ra = retry_after.expect("retry_after must be present after update"); + assert_eq!( + ra.max_duration, + Some(Duration::from_secs(60).try_into().unwrap()), + "max_duration should be 60s" + ); + }) + .await; +}