cache: Let services self-evict

Cargo.lock

@@ -783,7 +783,6 @@ name = "linkerd2-cache"
 version = "0.1.0"
 dependencies = [
  "futures",
- "indexmap",
  "linkerd2-error",
  "linkerd2-lock",
  "linkerd2-stack",

linkerd/app/core/src/classify.rs

@@ -3,7 +3,8 @@ use http;
 use linkerd2_error::Error;
 use linkerd2_http_classify as classify;
 pub use linkerd2_http_classify::{CanClassify, Layer};
-use linkerd2_proxy_http::{timeout, HasH2Reason};
+use linkerd2_proxy_http::HasH2Reason;
+use linkerd2_timeout::error::ResponseTimeout;
 use std::borrow::Cow;
 use tower_grpc::{self as grpc};
 use tracing::trace;
@@ -145,7 +146,7 @@ impl classify::ClassifyResponse for Response {
     }
 
     fn error(self, err: &Error) -> Self::Class {
-        let msg = if err.is::<timeout::error::ResponseTimeout>() {
+        let msg = if err.is::<ResponseTimeout>() {
             "timeout".into()
         } else {
             h2_error(err).into()

linkerd/app/core/src/config.rs

@@ -24,7 +24,7 @@ pub struct ConnectConfig {
 pub struct ProxyConfig<A: OrigDstAddr = SysOrigDstAddr> {
     pub server: ServerConfig<A>,
     pub connect: ConnectConfig,
-    pub cache_capacity: usize,
+    pub buffer_capacity: usize,
     pub cache_max_idle_age: Duration,
     pub disable_protocol_detection_for_ports: Arc<IndexSet<u16>>,
     pub dispatch_timeout: Duration,
@@ -56,7 +56,7 @@ impl<A: OrigDstAddr> ProxyConfig<A> {
         ProxyConfig {
             server: self.server.with_orig_dst_addr(orig_dst_addrs),
             connect: self.connect,
-            cache_capacity: self.cache_capacity,
+            buffer_capacity: self.buffer_capacity,
             cache_max_idle_age: self.cache_max_idle_age,
             disable_protocol_detection_for_ports: self.disable_protocol_detection_for_ports,
             max_in_flight_requests: self.max_in_flight_requests,

linkerd/app/core/src/errors.rs

@@ -1,14 +1,13 @@
-use crate::proxy::http::timeout::error as timeout;
 use crate::proxy::identity;
 use http::{header::HeaderValue, StatusCode};
-use linkerd2_cache::error as cache;
+use linkerd2_buffer as buffer;
 use linkerd2_error::Error;
 use linkerd2_error_metrics as metrics;
 use linkerd2_error_respond as respond;
 pub use linkerd2_error_respond::RespondLayer;
 use linkerd2_lock as lock;
 use linkerd2_proxy_http::HasH2Reason;
-use tower::load_shed::error as shed;
+use linkerd2_timeout::{error::ResponseTimeout, FailFastError};
 use tower_grpc::{self as grpc, Code};
 use tracing::debug;
 
@@ -29,11 +28,10 @@ pub type Label = (super::metric_labels::Direction, Reason);
 
 #[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
 pub enum Reason {
-    CacheFull,
     DispatchTimeout,
     ResponseTimeout,
     IdentityRequired,
-    LoadShed,
+    FailFast,
     Unexpected,
 }
 
@@ -112,18 +110,18 @@ impl<B: Default> respond::Respond for Respond<B> {
 }
 
 fn http_status(error: &Error) -> StatusCode {
-    if error.is::<timeout::ResponseTimeout>() {
+    if error.is::<ResponseTimeout>() {
         http::StatusCode::GATEWAY_TIMEOUT
-    } else if error.is::<cache::NoCapacity>() {
-        http::StatusCode::SERVICE_UNAVAILABLE
-    } else if error.is::<shed::Overloaded>() {
+    } else if error.is::<FailFastError>() {
         http::StatusCode::SERVICE_UNAVAILABLE
     } else if error.is::<tower::timeout::error::Elapsed>() {
         http::StatusCode::SERVICE_UNAVAILABLE
     } else if error.is::<IdentityRequired>() {
         http::StatusCode::FORBIDDEN
     } else if let Some(e) = error.downcast_ref::<lock::error::ServiceError>() {
         http_status(e.inner())
+    } else if let Some(e) = error.downcast_ref::<buffer::error::ServiceError>() {
+        http_status(e.inner())
     } else {
         http::StatusCode::BAD_GATEWAY
     }
@@ -133,20 +131,12 @@ fn set_grpc_status(error: &Error, headers: &mut http::HeaderMap) -> grpc::Code {
     const GRPC_STATUS: &'static str = "grpc-status";
     const GRPC_MESSAGE: &'static str = "grpc-message";
 
-    if error.is::<timeout::ResponseTimeout>() {
+    if error.is::<ResponseTimeout>() {
         let code = Code::DeadlineExceeded;
         headers.insert(GRPC_STATUS, code_header(code));
         headers.insert(GRPC_MESSAGE, HeaderValue::from_static("request timed out"));
         code
-    } else if error.is::<cache::NoCapacity>() {
-        let code = Code::Unavailable;
-        headers.insert(GRPC_STATUS, code_header(code));
-        headers.insert(
-            GRPC_MESSAGE,
-            HeaderValue::from_static("proxy router cache exhausted"),
-        );
-        code
-    } else if error.is::<shed::Overloaded>() {
+    } else if error.is::<FailFastError>() {
         let code = Code::Unavailable;
         headers.insert(GRPC_STATUS, code_header(code));
         headers.insert(
@@ -171,6 +161,8 @@ fn set_grpc_status(error: &Error, headers: &mut http::HeaderMap) -> grpc::Code {
         code
     } else if let Some(e) = error.downcast_ref::<lock::error::ServiceError>() {
         set_grpc_status(e.inner(), headers)
+    } else if let Some(e) = error.downcast_ref::<buffer::error::ServiceError>() {
+        set_grpc_status(e.inner(), headers)
     } else {
         let code = Code::Internal;
         headers.insert(GRPC_STATUS, code_header(code));
@@ -234,12 +226,10 @@ impl metrics::LabelError<Error> for LabelError {
     type Labels = Label;
 
     fn label_error(&self, err: &Error) -> Self::Labels {
-        let reason = if err.is::<timeout::ResponseTimeout>() {
+        let reason = if err.is::<ResponseTimeout>() {
             Reason::ResponseTimeout
-        } else if err.is::<cache::NoCapacity>() {
-            Reason::CacheFull
-        } else if err.is::<shed::Overloaded>() {
-            Reason::LoadShed
+        } else if err.is::<FailFastError>() {
+            Reason::FailFast
         } else if err.is::<tower::timeout::error::Elapsed>() {
             Reason::DispatchTimeout
         } else if err.is::<IdentityRequired>() {
@@ -258,8 +248,7 @@ impl metrics::FmtLabels for Reason {
             f,
             "message=\"{}\"",
             match self {
-                Reason::CacheFull => "cache full",
-                Reason::LoadShed => "load shed",
+                Reason::FailFast => "failfast",
                 Reason::DispatchTimeout => "dispatch timeout",
                 Reason::ResponseTimeout => "response timeout",
                 Reason::IdentityRequired => "identity required",

linkerd/app/core/src/svc.rs

@@ -112,10 +112,6 @@ impl<L> Layers<L> {
         self.push(concurrency_limit::Layer::new(max))
     }
 
-    pub fn push_load_shed(self) -> Layers<Pair<L, load_shed::Layer>> {
-        self.push(load_shed::Layer)
-    }
-
     pub fn push_make_ready<Req>(self) -> Layers<Pair<L, stack::MakeReadyLayer<Req>>> {
         self.push(stack::MakeReadyLayer::new())
     }
@@ -222,10 +218,6 @@ impl<S> Stack<S> {
         self.push(concurrency_limit::Layer::new(max))
     }
 
-    pub fn push_load_shed(self) -> Stack<load_shed::LoadShed<S>> {
-        self.push(load_shed::Layer)
-    }
-
     pub fn push_timeout(self, timeout: Duration) -> Stack<tower::timeout::Timeout<S>> {
         self.push(tower::timeout::TimeoutLayer::new(timeout))
     }
@@ -257,21 +249,14 @@ impl<S> Stack<S> {
         self.push(http::insert::target::layer())
     }
 
-    pub fn spawn_cache<T>(
-        self,
-        capacity: usize,
-        max_idle_age: Duration,
-    ) -> Stack<cache::Service<T, S>>
+    pub fn cache<T, L, U>(self, track: L) -> Stack<cache::Cache<T, cache::layer::NewTrack<L, S>>>
     where
-        T: Clone + Eq + std::hash::Hash + Send + 'static,
-        S: NewService<T> + Send + 'static,
-        S::Service: Clone + Send + 'static,
+        T: Eq + std::hash::Hash,
+        S: NewService<T> + Clone,
+        L: tower::layer::Layer<cache::layer::Track<S>> + Clone,
+        L::Service: NewService<T, Service = U>,
     {
-        Stack(
-            cache::Layer::new(capacity, max_idle_age)
-                .layer(self.0)
-                .spawn(),
-        )
+        self.push(cache::CacheLayer::new(track))
     }
 
     pub fn push_fallback<F: Clone>(self, fallback: F) -> Stack<stack::Fallback<S, F>> {
@@ -394,22 +379,6 @@ where
     }
 }
 
-/// Proivdes a cloneable Layer, unlike tower::load_shed.
-pub mod load_shed {
-    pub use tower::load_shed::LoadShed;
-
-    #[derive(Copy, Clone, Debug)]
-    pub struct Layer;
-
-    impl<S> super::Layer<S> for Layer {
-        type Service = LoadShed<S>;
-
-        fn layer(&self, inner: S) -> Self::Service {
-            LoadShed::new(inner)
-        }
-    }
-}
-
 pub mod make_response {
     use super::Oneshot;
     use crate::Error;

linkerd/app/inbound/src/lib.rs

@@ -77,7 +77,7 @@ impl<A: OrigDstAddr> Config<A> {
                 ProxyConfig {
                     server: ServerConfig { bind, h2_settings },
                     connect,
-                    cache_capacity,
+                    buffer_capacity,
                     cache_max_idle_age,
                     disable_protocol_detection_for_ports,
                     dispatch_timeout,
@@ -115,12 +115,20 @@ impl<A: OrigDstAddr> Config<A> {
                     move |_| Ok(backoff.stream())
                 }))
                 .into_new_service()
-                .push_on_response(
-                    svc::layers()
-                        .push_lock()
-                        .push(metrics.stack.layer(stack_labels("endpoint"))),
+                .cache(
+                    svc::layers().push_on_response(
+                        svc::layers()
+                            // If the service has been ready & unused for `cache_max_idle_age`,
+                            // fail it.
+                            .push_idle_timeout(cache_max_idle_age)
+                            // If the service has been unavailable for an extend time, eagerly
+                            // fail requests.
+                            .push_failfast(dispatch_timeout)
+                            // Shares the service, ensuring discovery errors are propagated.
+                            .push_spawn_buffer(buffer_capacity)
+                            .push(metrics.stack.layer(stack_labels("endpoint"))),
+                    ),
                 )
-                .spawn_cache(cache_capacity, cache_max_idle_age)
                 .instrument(|ep: &HttpEndpoint| {
                     info_span!(
                         "endpoint",
@@ -159,14 +167,24 @@ impl<A: OrigDstAddr> Config<A> {
                 .push(normalize_uri::layer())
                 .push(http_target_observability)
                 .into_new_service()
-                .push_on_response(
-                    svc::layers()
-                        .push_lock()
-                        .push(metrics.stack.layer(stack_labels("target"))),
+                .cache(
+                    svc::layers().push_on_response(
+                        svc::layers()
+                            // If the service has been ready & unused for `cache_max_idle_age`,
+                            // fail it.
+                            .push_idle_timeout(cache_max_idle_age)
+                            // If the service has been unavailable for an extend time, eagerly
+                            // fail requests.
+                            .push_failfast(dispatch_timeout)
+                            // Shares the service, ensuring discovery errors are propagated.
+                            .push_spawn_buffer(buffer_capacity)
+                            .push(metrics.stack.layer(stack_labels("target"))),
+                    ),
                 )
-                .check_new_clone_service::<Target>()
-                .spawn_cache(cache_capacity, cache_max_idle_age)
                 .instrument(|_: &Target| info_span!("target"))
+                // Prevent the cache's lock from being acquired in poll_ready, ensuring this happens
+                // in the response future. This prevents buffers from holding the cache's lock.
+                .push_oneshot()
                 .check_service::<Target>();
 
             // Routes targets to a Profile stack, i.e. so that profile
@@ -181,16 +199,26 @@ impl<A: OrigDstAddr> Config<A> {
                     http_profile_route_proxy.into_inner(),
                 ))
                 .into_new_service()
-                .push_on_response(
-                    svc::layers()
-                        .push_lock()
-                        .push(metrics.stack.layer(stack_labels("profile"))),
-                )
                 // Caches profile stacks.
-                .check_new_clone_service::<Profile>()
-                .spawn_cache(cache_capacity, cache_max_idle_age)
+                .check_new_service_routes::<Profile, Target>()
+                .cache(
+                    svc::layers().push_on_response(
+                        svc::layers()
+                            // If the service has been ready & unused for `cache_max_idle_age`,
+                            // fail it.
+                            .push_idle_timeout(cache_max_idle_age)
+                            // If the service has been unavailable for an extend time, eagerly
+                            // fail requests.
+                            .push_failfast(dispatch_timeout)
+                            // Shares the service, ensuring discovery errors are propagated.
+                            .push_spawn_buffer(buffer_capacity)
+                            .push(metrics.stack.layer(stack_labels("profile"))),
+                    ),
+                )
                 .instrument(|p: &Profile| info_span!("profile", addr = %p.addr()))
-                .check_service::<Profile>()
+                .check_make_service::<Profile, Target>()
+                // Ensures that cache's lock isn't held in poll_ready.
+                .push_oneshot()
                 .push(router::Layer::new(|()| ProfileTarget))
                 .check_new_service_routes::<(), Target>()
                 .new_service(());
@@ -205,16 +233,10 @@ impl<A: OrigDstAddr> Config<A> {
             let http_admit_request = svc::layers()
                 // Downgrades the protocol if upgraded by an outbound proxy.
                 .push(svc::layer::mk(orig_proto::Downgrade::new))
-                // Ensures that load is not shed if the inner service is in-use.
-                .push_oneshot()
                 // Limits the number of in-flight requests.
                 .push_concurrency_limit(max_in_flight_requests)
-                // Sheds load if too many requests are in flight.
-                //
-                // XXX Can this be removed? Is it okay to just backpressure onto
-                // the client? Should we instead limit the number of active
-                // connections?
-                .push_load_shed()
+                // Eagerly fail requests when the proxy is out of capacity for some time period.
+                .push_failfast(dispatch_timeout)
                 .push(metrics.http_errors)
                 // Synthesizes responses for proxy errors.
                 .push(errors::layer());