outbound: configure failure accrual policies from stack targets (#2354)

Depends on #2353.

PR #2353 adds middleware for implementing request-level circuit
breaking. This branch adds the circuit breaking middleware to the
outbound concrete stack, and adds plumbing for configuring a concrete
stack's circuit breaker based on params provided by the target.


A new `FailureAccrual` enum is added in `linkerd2-proxy-client-policy`
to represent the failure accrual policy for a circuit breaker.
Currently, no actual implementations of failure accrual policies exist
in the proxy, so the only available variant of `FailureAccrual` is
`FailureAccrual::None`, which disables circuit breaking. Circuit
breaking middleware (a `Gate`/`BroadcastClassification` pair) is still
constructed, but no failure accural task is spawned to actually open and
shut the gate, so no circuit breaking is actually performed. Subsequent
branches will actually implement failure accrual policies.

Failure accrual policies are configured at the protocol level, rather
than per-route or per-backend. This is because a given policy may
contain multiple routes referencing the same backend, and a single
concrete stack is constructed for that backend that's shared across all
distributions that include it. If failure accrual policies were
configured at the `RouteBackend` level, we would need to build separate
client stacks if the same backend is referenced by `RouteBackend`s that
have different failure accrual policies. Currently, failure accrual
policies are not present in the proxy API, so all `ClientPolicy`
instances have the default policy, `FailureAccrual::None`. Once the
`OutboundPolicy` proxy API actually provides failure accrual
configurations, a subsequent branch will populate this configuration
from discovery.

Author: hawkw

linkerd/app/core/src/classify.rs

@@ -1,4 +1,5 @@
 use crate::profiles;
+pub use classify::gate;
 use linkerd_error::Error;
 use linkerd_proxy_client_policy as client_policy;
 use linkerd_proxy_http::{classify, HasH2Reason, ResponseTimeoutError};

linkerd/app/outbound/src/discover.rs

@@ -238,8 +238,12 @@ pub fn synthesize_forward_policy(
         timeout,
         http1: policy::http::Http1 {
             routes: routes.clone(),
+            failure_accrual: Default::default(),
+        },
+        http2: policy::http::Http2 {
+            routes,
+            failure_accrual: Default::default(),
         },
-        http2: policy::http::Http2 { routes },
         opaque,
     };
 

linkerd/app/outbound/src/http/concrete.rs

@@ -4,7 +4,7 @@
 use super::{balance, client, handle_proxy_error_headers};
 use crate::{http, stack_labels, Outbound};
 use linkerd_app_core::{
-    metrics, profiles,
+    classify, metrics, profiles,
     proxy::{
         api_resolve::{ConcreteAddr, Metadata, ProtocolHint},
         core::Resolve,
@@ -15,6 +15,7 @@ use linkerd_app_core::{
     transport::{self, addrs::*},
     Error, Infallible, NameAddr,
 };
+use linkerd_proxy_client_policy::FailureAccrual;
 use std::{fmt::Debug, net::SocketAddr, sync::Arc};
 use tracing::info_span;
 
@@ -88,6 +89,8 @@ impl<N> Outbound<N> {
         // Concrete target type.
         T: svc::Param<Dispatch>,
         // TODO(ver) T: svc::Param<svc::queue::Capacity> + svc::Param<svc::queue::Timeout>,
+        // Failure accrual policy.
+        T: svc::Param<FailureAccrual>,
         T: Clone + Debug + Send + Sync + 'static,
         // Endpoint resolution.
         R: Resolve<ConcreteAddr, Error = Error, Endpoint = Metadata>,
@@ -125,6 +128,20 @@ impl<N> Outbound<N> {
                 )
                 .instrument(|e: &Endpoint<T>| info_span!("endpoint", addr = %e.addr));
 
+            let mk_breaker = |target: &Balance<T>| {
+                match target.parent.param() {
+                    FailureAccrual::None => |_: &(SocketAddr, Metadata)| {
+                        // Construct a gate channel, dropping the controller
+                        // side of the channel such that response summaries
+                        // are never read. The failure accrual gate never
+                        // closes in this configuration.
+                        tracing::trace!("No failure accrual policy enabled");
+                        let (prms, _, _) = classify::gate::Params::channel(1);
+                        prms
+                    },
+                }
+            };
+
             let balance = endpoint
                 .push_map_target({
                     let inbound_ips = inbound_ips.clone();
@@ -143,7 +160,11 @@ impl<N> Outbound<N> {
                         }
                     }
                 })
+                .push_on_service(svc::MapErr::layer_boxed())
                 .lift_new_with_target()
+                .push(
+                    http::NewClassifyGateSet::<classify::Response, _, _, _>::layer_via(mk_breaker),
+                )
                 .push(http::NewBalancePeakEwma::layer(resolve))
                 .push(svc::NewMapErr::layer_from_target::<ConcreteError, _>())
                 .push_on_service(http::BoxResponse::layer())

linkerd/app/outbound/src/http/logical.rs

@@ -38,6 +38,7 @@ pub struct Concrete<T> {
     target: concrete::Dispatch,
     authority: Option<http::uri::Authority>,
     parent: T,
+    failure_accrual: policy::FailureAccrual,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -169,6 +170,7 @@ where
                                     target: concrete::Dispatch::Forward(remote, meta),
                                     authority: None,
                                     parent,
+                                    failure_accrual: Default::default(),
                                 })
                             }
                             RouterParams::Profile(profile) => {
@@ -254,6 +256,12 @@ impl<T> svc::Param<concrete::Dispatch> for Concrete<T> {
     }
 }
 
+impl<T> svc::Param<policy::FailureAccrual> for Concrete<T> {
+    fn param(&self) -> policy::FailureAccrual {
+        self.failure_accrual
+    }
+}
+
 // === impl CanonicalDstHeader ===
 
 impl From<CanonicalDstHeader> for http::HeaderPair {

linkerd/app/outbound/src/http/logical/policy.rs

@@ -11,7 +11,7 @@ pub use self::{
     route::errors,
     router::{GrpcParams, HttpParams},
 };
-pub use linkerd_proxy_client_policy::ClientPolicy;
+pub use linkerd_proxy_client_policy::{ClientPolicy, FailureAccrual};
 
 /// HTTP or gRPC policy route parameters.
 #[derive(Clone, Debug, PartialEq, Eq)]

linkerd/app/outbound/src/http/logical/policy/router.rs

@@ -13,6 +13,7 @@ pub struct Params<M, F, E> {
     pub addr: Addr,
     pub routes: Arc<[http_route::Route<M, policy::RoutePolicy<F, E>>]>,
     pub backends: Arc<[policy::Backend]>,
+    pub failure_accrual: policy::FailureAccrual,
 }
 
 pub type HttpParams =
@@ -114,6 +115,7 @@ where
             addr,
             routes,
             backends,
+            failure_accrual,
         } = rts;
 
         let mk_concrete = {
@@ -130,6 +132,7 @@ where
                     target,
                     authority,
                     parent: parent.clone(),
+                    failure_accrual,
                 }
             }
         };

linkerd/app/outbound/src/http/logical/policy/tests.rs

@@ -75,6 +75,7 @@ async fn header_based_route() {
                 ],
             }]),
             backends: std::iter::once(default).chain(Some(special)).collect(),
+            failure_accrual: Default::default(),
         }
     });
 
@@ -167,6 +168,7 @@ async fn http_filter_request_headers() {
                 }],
             }]),
             backends: std::iter::once(backend).collect(),
+            failure_accrual: Default::default(),
         }
     });
 

linkerd/app/outbound/src/http/logical/profile.rs

@@ -123,6 +123,7 @@ where
                 target: concrete::Dispatch::Balance(addr.clone(), DEFAULT_EWMA),
                 authority: Some(addr.as_http_authority()),
                 parent: parent.clone(),
+                failure_accrual: Default::default(),
             };
             let backends = std::iter::once(concrete.clone()).collect();
             let distribution = Distribution::first_available(std::iter::once(concrete));
@@ -134,6 +135,7 @@ where
                     target: concrete::Dispatch::Balance(t.addr.clone(), DEFAULT_EWMA),
                     authority: Some(t.addr.as_http_authority()),
                     parent: parent.clone(),
+                    failure_accrual: Default::default(),
                 })
                 .collect();
             let distribution = Distribution::random_available(targets.iter().cloned().map(
@@ -142,6 +144,7 @@ where
                         authority: Some(addr.as_http_authority()),
                         target: concrete::Dispatch::Balance(addr, DEFAULT_EWMA),
                         parent: parent.clone(),
+                        failure_accrual: Default::default(),
                     };
                     (concrete, weight)
                 },

linkerd/app/outbound/src/ingress.rs

@@ -417,40 +417,53 @@ fn policy_routes(
             ref http2,
             ..
         } => {
-            let routes = match version {
-                http::Version::Http1 => http1.routes.clone(),
-                http::Version::H2 => http2.routes.clone(),
+            let (routes, failure_accrual) = match version {
+                http::Version::Http1 => (http1.routes.clone(), http1.failure_accrual),
+                http::Version::H2 => (http2.routes.clone(), http2.failure_accrual),
             };
             Some(http::Routes::Policy(http::policy::Params::Http(
                 http::policy::HttpParams {
                     addr,
                     backends: policy.backends.clone(),
                     routes,
+                    failure_accrual,
                 },
             )))
         }
         // TODO(eliza): what do we do here if the configured
         // protocol doesn't match the actual protocol for the
         // target? probably should make an error route instead?
-        policy::Protocol::Http1(ref http1) => Some(http::Routes::Policy(
-            http::policy::Params::Http(http::policy::HttpParams {
+        policy::Protocol::Http1(policy::http::Http1 {
+            ref routes,
+            failure_accrual,
+        }) => Some(http::Routes::Policy(http::policy::Params::Http(
+            http::policy::HttpParams {
                 addr,
                 backends: policy.backends.clone(),
-                routes: http1.routes.clone(),
-            }),
-        )),
-        policy::Protocol::Http2(ref http2) => Some(http::Routes::Policy(
-            http::policy::Params::Http(http::policy::HttpParams {
+                routes: routes.clone(),
+                failure_accrual,
+            },
+        ))),
+        policy::Protocol::Http2(policy::http::Http2 {
+            ref routes,
+            failure_accrual,
+        }) => Some(http::Routes::Policy(http::policy::Params::Http(
+            http::policy::HttpParams {
                 addr,
                 backends: policy.backends.clone(),
-                routes: http2.routes.clone(),
-            }),
-        )),
-        policy::Protocol::Grpc(ref grpc) => Some(http::Routes::Policy(http::policy::Params::Grpc(
+                routes: routes.clone(),
+                failure_accrual,
+            },
+        ))),
+        policy::Protocol::Grpc(policy::grpc::Grpc {
+            ref routes,
+            failure_accrual,
+        }) => Some(http::Routes::Policy(http::policy::Params::Grpc(
             http::policy::GrpcParams {
                 addr,
                 backends: policy.backends.clone(),
-                routes: grpc.routes.clone(),
+                routes: routes.clone(),
+                failure_accrual,
             },
         ))),
         _ => None,

linkerd/app/outbound/src/sidecar.rs

@@ -212,23 +212,33 @@ impl HttpSidecar {
         // protocol changes but remains HTTP-ish, we propagate those
         // changes. If the protocol flips to an opaque protocol, we ignore
         // the protocol update.
-        let routes = match policy.protocol {
+        let (routes, failure_accrual) = match policy.protocol {
             policy::Protocol::Detect {
                 ref http1,
                 ref http2,
                 ..
             } => match version {
-                http::Version::Http1 => http1.routes.clone(),
-                http::Version::H2 => http2.routes.clone(),
+                http::Version::Http1 => (http1.routes.clone(), http1.failure_accrual),
+                http::Version::H2 => (http2.routes.clone(), http2.failure_accrual),
             },
-            policy::Protocol::Http1(ref http1) => http1.routes.clone(),
-            policy::Protocol::Http2(ref http2) => http2.routes.clone(),
-            policy::Protocol::Grpc(ref grpc) => {
+            policy::Protocol::Http1(policy::http::Http1 {
+                ref routes,
+                failure_accrual,
+            }) => (routes.clone(), failure_accrual),
+            policy::Protocol::Http2(policy::http::Http2 {
+                ref routes,
+                failure_accrual,
+            }) => (routes.clone(), failure_accrual),
+            policy::Protocol::Grpc(policy::grpc::Grpc {
+                ref routes,
+                failure_accrual,
+            }) => {
                 return Some(http::Routes::Policy(http::policy::Params::Grpc(
                     http::policy::GrpcParams {
                         addr: orig_dst.into(),
                         backends: policy.backends.clone(),
-                        routes: grpc.routes.clone(),
+                        routes: routes.clone(),
+                        failure_accrual,
                     },
                 )))
             }
@@ -245,6 +255,7 @@ impl HttpSidecar {
                 addr: orig_dst.into(),
                 routes,
                 backends: policy.backends.clone(),
+                failure_accrual,
             },
         )))
     }