Decouple discovery buffering from endpoint conversion

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,