Avoid buffering control plane discovery

Cargo.lock

@@ -1249,6 +1249,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/app/src/identity.rs

@@ -1,12 +1,12 @@
 pub use linkerd2_app_core::proxy::identity::{
     certify, Crt, CrtKey, Csr, InvalidName, Key, Local, Name, TokenSource, TrustAnchors,
 };
-use linkerd2_app_core::proxy::{discover, http};
 use linkerd2_app_core::{
     classify,
     config::{ControlAddr, ControlConfig},
     control, dns,
     exp_backoff::{ExponentialBackoff, ExponentialBackoffStream},
+    proxy::{discover, http},
     reconnect,
     svc::{self, NewService},
     transport::tls,
@@ -49,15 +49,6 @@ impl Config {
             Config::Enabled { control, certify } => {
                 const EWMA_DEFAULT_RTT: Duration = Duration::from_millis(30);
                 const EWMA_DECAY: Duration = Duration::from_secs(10);
-                let discover = {
-                    const BUFFER_CAPACITY: usize = 1_000;
-                    let cache_timeout = Duration::from_secs(60);
-                    discover::Layer::new(
-                        BUFFER_CAPACITY,
-                        cache_timeout,
-                        control::dns_resolve::Resolve::new(dns),
-                    )
-                };
 
                 let (local, crt_store) = Local::new(&certify);
 
@@ -68,7 +59,7 @@ impl Config {
                     )))
                     .push_timeout(control.connect.timeout)
                     .push(control::client::layer())
-                    .push(discover)
+                    .push(discover::resolve(control::dns_resolve::Resolve::new(dns)))
                     .push_on_response(http::balance::layer(EWMA_DEFAULT_RTT, EWMA_DECAY))
                     .push(reconnect::layer(Recover(control.connect.backoff)))
                     .push(metrics.into_layer::<classify::Response>())

linkerd/app/src/lib.rs

@@ -17,7 +17,7 @@ use linkerd2_app_core::{
     classify,
     config::ControlAddr,
     control, dns, drain,
-    proxy::http,
+    proxy::{discover, http},
     reconnect,
     svc::{self, NewService},
     transport::tls,
@@ -110,24 +110,12 @@ impl Config {
             let metrics = metrics.control.clone();
             let dns = dns.resolver.clone();
 
-            use linkerd2_app_core::proxy::discover;
-
-            let discover = {
-                const BUFFER_CAPACITY: usize = 1_000;
-                let cache_timeout = Duration::from_secs(60);
-                discover::Layer::new(
-                    BUFFER_CAPACITY,
-                    cache_timeout,
-                    control::dns_resolve::Resolve::new(dns),
-                )
-            };
-
             info_span!("dst").in_scope(|| {
                 let dst_connect = svc::connect(dst.control.connect.keepalive)
                     .push(tls::ConnectLayer::new(identity.local()))
                     .push_timeout(dst.control.connect.timeout)
                     .push(control::client::layer())
-                    .push(discover)
+                    .push(discover::resolve(control::dns_resolve::Resolve::new(dns)))
                     .push_on_response(http::balance::layer(EWMA_DEFAULT_RTT, EWMA_DECAY))
                     .push(reconnect::layer({
                         let backoff = dst.control.connect.backoff;

linkerd/app/src/oc_collector.rs

@@ -57,15 +57,6 @@ impl Config {
             } => {
                 const EWMA_DEFAULT_RTT: Duration = Duration::from_millis(30);
                 const EWMA_DECAY: Duration = Duration::from_secs(10);
-                let discover = {
-                    const BUFFER_CAPACITY: usize = 1_000;
-                    let cache_timeout = Duration::from_secs(60);
-                    discover::Layer::new(
-                        BUFFER_CAPACITY,
-                        cache_timeout,
-                        control::dns_resolve::Resolve::new(dns),
-                    )
-                };
 
                 let addr = control.addr;
                 let svc = svc::connect(control.connect.keepalive)
@@ -76,7 +67,7 @@ impl Config {
                     // TODO: we should have metrics of some kind, but the standard
                     // HTTP metrics aren't useful for a client where we never read
                     // the response.
-                    .push(discover)
+                    .push(discover::resolve(control::dns_resolve::Resolve::new(dns)))
                     .push_on_response(http::balance::layer(EWMA_DEFAULT_RTT, EWMA_DECAY))
                     .push(reconnect::layer({
                         let backoff = control.connect.backoff;

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,
@@ -296,7 +288,6 @@ mod tests {
     use std::net::SocketAddr;
     use tokio::sync::mpsc;
     use tokio_test::{assert_pending, assert_ready, assert_ready_ok, task};
-    use tower::discover::Change;
     use tower::util::service_fn;
     use tower::Service;
     use tower_test::mock;