Decouple discovery buffering from endpoint conversion (#631)

The discovery layer handles both building a service for each discovered
endpoint, as well as a buffer that fails when the service is not polled
(i.e. because it has been leaked somehow).

This coupling doesn't match other use cases though when, for instance,
the client may not be used frequently, as is the case for
proxy-to-control-plane communication.

This change decouples these layers so that the buffering is decoupled
from endpoint-service construction.

Cargo.lock

@@ -1248,6 +1248,7 @@ dependencies = [
  "indexmap",
  "linkerd2-error",
  "linkerd2-proxy-core",
+ "linkerd2-stack",
  "pin-project",
  "tokio",
  "tokio-test",

linkerd/app/outbound/src/lib.rs

@@ -254,11 +254,12 @@ impl Config {
         // This buffer controls how many discovery updates may be pending/unconsumed by the
         // balancer before backpressure is applied on the resolution stream. If the buffer is
         // full for `cache_max_idle_age`, then the resolution task fails.
-        let discover = {
-            const BUFFER_CAPACITY: usize = 1_000;
-            let resolve = map_endpoint::Resolve::new(endpoint::FromMetadata, resolve.clone());
-            discover::Layer::new(BUFFER_CAPACITY, cache_max_idle_age, resolve)
-        };
+        let discover = svc::layers()
+            .push(discover::resolve(map_endpoint::Resolve::new(
+                endpoint::FromMetadata,
+                resolve.clone(),
+            )))
+            .push(discover::buffer(1_000, cache_max_idle_age));
 
         // Builds a balancer for each concrete destination.
         let http_balancer = svc::stack(http_endpoint.clone())

linkerd/proxy/discover/Cargo.toml

@@ -13,6 +13,7 @@ Utilities to implement a Discover with the core Resolve type
 futures = "0.3"
 linkerd2-error = { path = "../../error" }
 linkerd2-proxy-core = { path = "../core" }
+linkerd2-stack = { path = "../../stack" }
 indexmap = "1.0"
 tokio = { version = "0.2", features = ["sync", "time", "stream"] }
 tracing = "0.1.19"
@@ -29,4 +30,4 @@ features = ["discover"]
 [dev-dependencies]
 tower-test = "0.3"
 tokio-test = "0.2"
-tower = { version = "0.3", default-features = false, features = ["discover", "util"]}
+tower = { version = "0.3", default-features = false, features = ["discover", "util"]}

linkerd/proxy/discover/src/buffer.rs

@@ -11,6 +11,7 @@ use tokio::time::{self, Delay};
 use tower::discover;
 use tracing::warn;
 use tracing_futures::Instrument;
+
 #[derive(Clone, Debug)]
 pub struct Buffer<M> {
     capacity: usize,
@@ -51,10 +52,7 @@ pub struct Daemon<D: discover::Discover> {
 pub struct Lost(());
 
 impl<M> Buffer<M> {
-    pub fn new<T>(capacity: usize, watchdog_timeout: Duration, inner: M) -> Self
-    where
-        Self: tower::Service<T>,
-    {
+    pub fn new(capacity: usize, watchdog_timeout: Duration, inner: M) -> Self {
         Self {
             capacity,
             watchdog_timeout,

linkerd/proxy/discover/src/lib.rs

@@ -1,8 +1,7 @@
 #![deny(warnings, rust_2018_idioms)]
 
-use linkerd2_error::Error;
 use linkerd2_proxy_core::Resolve;
-use std::fmt;
+use linkerd2_stack::layer;
 use std::time::Duration;
 
 pub mod buffer;
@@ -13,49 +12,15 @@ use self::buffer::Buffer;
 use self::from_resolve::FromResolve;
 use self::make_endpoint::MakeEndpoint;
 
-#[derive(Clone, Debug)]
-pub struct Layer<T, R> {
-    capacity: usize,
-    watchdog: Duration,
-    resolve: R,
-    _marker: std::marker::PhantomData<fn(T)>,
-}
-
-// === impl Layer ===
-
-impl<T, R> Layer<T, R> {
-    pub fn new(capacity: usize, watchdog: Duration, resolve: R) -> Self
-    where
-        R: Resolve<T> + Clone,
-        R::Endpoint: fmt::Debug + Clone + PartialEq,
-    {
-        Self {
-            capacity,
-            watchdog,
-            resolve,
-            _marker: std::marker::PhantomData,
-        }
-    }
+pub fn buffer<M>(capacity: usize, watchdog: Duration) -> impl layer::Layer<M, Service = Buffer<M>> {
+    layer::mk(move |inner: M| Buffer::new(capacity, watchdog, inner))
 }
 
-impl<T, R, M> tower::layer::Layer<M> for Layer<T, R>
+pub fn resolve<T, R, M>(
+    resolve: R,
+) -> impl layer::Layer<M, Service = MakeEndpoint<FromResolve<R, R::Endpoint>, M>>
 where
-    T: fmt::Display,
-    R: Resolve<T> + Send + Clone + 'static,
-    R::Error: Into<Error>,
-    R::Endpoint: fmt::Debug + Clone + PartialEq + Send + 'static,
-    R::Resolution: Send + 'static,
-    R::Future: Send + 'static,
-    M: tower::Service<R::Endpoint> + Clone + Send + 'static,
-    M::Error: Into<Error>,
-    M::Response: Send + 'static,
-    M::Future: Send + 'static,
+    R: Resolve<T> + Clone,
 {
-    type Service = Buffer<MakeEndpoint<FromResolve<R, R::Endpoint>, M>>;
-
-    fn layer(&self, make_endpoint: M) -> Self::Service {
-        let make_discover =
-            MakeEndpoint::new(make_endpoint, FromResolve::new(self.resolve.clone()));
-        Buffer::new(self.capacity, self.watchdog, make_discover)
-    }
+    layer::mk(move |inner: M| MakeEndpoint::new(inner, FromResolve::new(resolve.clone())))
 }

linkerd/proxy/discover/src/make_endpoint.rs

@@ -59,15 +59,7 @@ enum MakeError<E> {
 // === impl MakeEndpoint ===
 
 impl<D, E> MakeEndpoint<D, E> {
-    pub fn new<T, InnerDiscover>(make_endpoint: E, make_discover: D) -> Self
-    where
-        D: tower::Service<T, Response = InnerDiscover>,
-        InnerDiscover: discover::Discover,
-        InnerDiscover::Key: Clone,
-        InnerDiscover::Error: Into<Error>,
-        E: tower::Service<InnerDiscover::Service> + Clone,
-        E::Error: Into<Error>,
-    {
+    pub fn new(make_endpoint: E, make_discover: D) -> Self {
         Self {
             make_discover,
             make_endpoint,