feat(policy): support load-bias and retry-after balancer annotations

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",

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"]

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),

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<AppProtocol>,
     pub accrual: Option<FailureAccrual>,
+    pub load_bias: Option<LoadBiasConfig>,
+    pub retry_after: Option<RetryAfterConfig>,
     pub http_retry: Option<RouteRetry<HttpRetryCondition>>,
     pub grpc_retry: Option<RouteRetry<GrpcRetryCondition>>,
     pub timeouts: RouteTimeouts,

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::<Vec<_>>();
@@ -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<SocketAddr>) ->
                         )),
                     }),
                     load: Some(default_balancer_config()),
+                    ejection: None,
                 },
             )),
         },

policy-controller/grpc/src/outbound/grpc.rs

@@ -21,6 +21,8 @@ pub(crate) fn protocol(
     default_backend: outbound::Backend,
     routes: impl Iterator<Item = (GroupKindNamespaceName, GrpcRoute)>,
     failure_accrual: Option<outbound::FailureAccrual>,
+    load_bias: Option<outbound::LoadBiasConfig>,
+    retry_after: Option<outbound::RetryAfterConfig>,
     service_retry: Option<RouteRetry<GrpcRetryCondition>>,
     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,
                             },
                         )),
                     }),

policy-controller/grpc/src/outbound/http.rs

@@ -21,6 +21,8 @@ pub(crate) fn protocol(
     default_backend: outbound::Backend,
     routes: impl Iterator<Item = (GroupKindNamespaceName, HttpRoute)>,
     accrual: Option<outbound::FailureAccrual>,
+    load_bias: Option<outbound::LoadBiasConfig>,
+    retry_after: Option<outbound::RetryAfterConfig>,
     service_retry: Option<RouteRetry<HttpRetryCondition>>,
     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<Item = (GroupKindNamespaceName, HttpRoute)>,
     accrual: Option<outbound::FailureAccrual>,
+    load_bias: Option<outbound::LoadBiasConfig>,
+    retry_after: Option<outbound::RetryAfterConfig>,
     service_retry: Option<RouteRetry<HttpRetryCondition>>,
     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<Item = (GroupKindNamespaceName, HttpRoute)>,
     accrual: Option<outbound::FailureAccrual>,
+    load_bias: Option<outbound::LoadBiasConfig>,
+    retry_after: Option<outbound::RetryAfterConfig>,
     service_retry: Option<RouteRetry<HttpRetryCondition>>,
     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,
                             },
                         )),
                     }),

policy-controller/grpc/src/outbound/tcp.rs

@@ -133,6 +133,7 @@ fn convert_backend(
                                 )),
                             }),
                             load: Some(default_balancer_config()),
+                            ejection: None,
                         },
                     )),
                 }),

policy-controller/grpc/src/outbound/tls.rs

@@ -135,6 +135,7 @@ fn convert_backend(
                                 )),
                             }),
                             load: Some(default_balancer_config()),
+                            ejection: None,
                         },
                     )),
                 }),