discover: Make endpoints with NewService (#671)

This branch changes `linkerd2_proxy_discover::MakeEndpoint` to wrap a
`NewService` rather than a `MakeService`, and construct endpoint
services synchronously rather than asynchronously. This allows removing
a great deal of complex logic for driving and cancelling `MakeService`
futures that was previously necessary.

For now, we just wrap the underlying endpoint stacks in
`into_new_service` to create them lazily. In the future, these stacks
will likely be changed to native `NewService` impls.

Author: hawkw

Cargo.lock

@@ -750,6 +750,7 @@ version = "0.1.0"
 dependencies = [
  "futures 0.3.5",
  "linkerd2-error",
+ "linkerd2-stack",
  "tower",
  "tracing",
 ]

linkerd/admit/Cargo.toml

@@ -8,5 +8,6 @@ publish = false
 [dependencies]
 futures = "0.3"
 linkerd2-error = { path = "../error" }
+linkerd2-stack = { path = "../stack" }
 tower = { version = "0.3", default-features = false, features = ["util"] }
-tracing = "0.1.19"
+tracing = "0.1.19"

linkerd/admit/src/lib.rs

@@ -4,6 +4,7 @@
 
 use futures::{future, TryFutureExt};
 use linkerd2_error::Error;
+use linkerd2_stack::{NewService, ResultService};
 use std::task::{Context, Poll};
 
 pub struct AdmitLayer<A>(A);
@@ -37,6 +38,21 @@ impl<A: Clone, S> tower::layer::Layer<S> for AdmitLayer<A> {
     }
 }
 
+impl<A, T, S> NewService<T> for AdmitService<A, S>
+where
+    A: Admit<T>,
+    S: NewService<T>,
+{
+    type Service = ResultService<S::Service, A::Error>;
+
+    fn new_service(&mut self, t: T) -> Self::Service {
+        match self.admit.admit(&t) {
+            Ok(()) => ResultService::ok(self.inner.new_service(t)),
+            Err(e) => ResultService::err(e),
+        }
+    }
+}
+
 impl<A, T, S> tower::Service<T> for AdmitService<A, S>
 where
     A: Admit<T>,

linkerd/app/core/src/control.rs

@@ -63,6 +63,7 @@ impl Config {
             .push(self::client::layer())
             .push(reconnect::layer(backoff.clone()))
             .push_spawn_ready()
+            .into_new_service()
             .push(self::resolve::layer(dns, backoff))
             .push_on_response(self::control::balance::layer())
             .push(metrics.into_layer::<classify::Response>())

linkerd/app/inbound/src/lib.rs

@@ -219,6 +219,7 @@ impl Config {
         let target = endpoint
             .push_map_target(HttpEndpoint::from)
             .push(observe)
+            .push_on_response(svc::layers().box_http_response())
             .into_new_service()
             .check_new_service::<Target, http::Request<_>>();
 
@@ -248,50 +249,41 @@ impl Config {
             .push_map_target(endpoint::Logical::from)
             .push(profiles::discover::layer(profiles_client))
             .into_new_service()
-            .cache(
-                svc::layers().push_on_response(
-                    svc::layers()
-                        .push_failfast(dispatch_timeout)
-                        .push_spawn_buffer_with_idle_timeout(buffer_capacity, cache_max_idle_age)
-                        .push(metrics.stack.layer(stack_labels("profile")))
-                        .box_http_response(),
-                ),
-            )
-            .into_make_service()
-            .spawn_buffer(buffer_capacity)
             .instrument(|_: &Target| debug_span!("profile"))
-            .check_make_service::<Target, http::Request<_>>();
+            .push_on_response(svc::layers().box_http_response())
+            .check_new_service::<Target, http::Request<http::boxed::Payload>>();
 
         let forward = target
-            .cache(
-                svc::layers().push_on_response(
-                    svc::layers()
-                        .push_failfast(dispatch_timeout)
-                        .push_spawn_buffer_with_idle_timeout(buffer_capacity, cache_max_idle_age)
-                        .push(metrics.stack.layer(stack_labels("forward")))
-                        .box_http_response(),
-                ),
-            )
-            .into_make_service()
-            .spawn_buffer(buffer_capacity)
             .instrument(|_: &Target| debug_span!("forward"))
-            .check_make_service::<Target, http::Request<http::boxed::Payload>>();
+            .check_new_service::<Target, http::Request<http::boxed::Payload>>();
 
         // Attempts to resolve the target as a service profile or, if that
         // fails, skips that stack to forward to the local endpoint.
         profile
-            .push_make_ready()
             .push_fallback(forward)
+            .check_new_service::<Target, http::Request<http::boxed::Payload>>()
             // If the traffic is targeted at the inbound port, send it through
-            // the loopback service (i.e. as a gateway). This is done before
-            // caching so that the loopback stack can determine whether it
-            // should cache or not.
+            // the loopback service (i.e. as a gateway).
             .push(admit::AdmitLayer::new(prevent_loop))
+            .check_new_service::<Target, http::Request<http::boxed::Payload>>()
             .push_fallback_on_error::<prevent_loop::LoopPrevented, _>(
                 svc::stack(loopback)
-                    .check_make_service::<Target, http::Request<_>>()
+                    .into_new_service()
+                    .check_new_service::<Target, http::Request<_>>()
                     .into_inner(),
             )
+            .check_new_service::<Target, http::Request<http::boxed::Payload>>()
+            .cache(
+                svc::layers().push_on_response(
+                    svc::layers()
+                        .push_failfast(dispatch_timeout)
+                        .push_spawn_buffer_with_idle_timeout(buffer_capacity, cache_max_idle_age)
+                        .push(metrics.stack.layer(stack_labels("logical")))
+                        .box_http_response(),
+                ),
+            )
+            .into_make_service()
+            .spawn_buffer(buffer_capacity)
             .check_make_service::<Target, http::Request<http::boxed::Payload>>()
             .into_inner()
     }
@@ -361,9 +353,6 @@ impl Config {
         ));
 
         let http_server = svc::stack(http_router)
-            // Ensures that the built service is ready before it is returned
-            // to the router to dispatch a request.
-            .push_make_ready()
             // Limits the amount of time each request waits to obtain a
             // ready service.
             .push_timeout(dispatch_timeout)

linkerd/app/outbound/src/lib.rs

@@ -85,21 +85,18 @@ impl Config {
         &self,
         connect: C,
         resolve: E,
-    ) -> impl tower::Service<
+    ) -> impl svc::NewService<
         SocketAddr,
-        Error = impl Into<Error>,
-        Future = impl Unpin + Send + 'static,
-        Response = impl tower::Service<
+        Service = impl tower::Service<
             I,
             Response = (),
             Future = impl Unpin + Send + 'static,
             Error = impl Into<Error>,
         > + Unpin
-                       + Clone
-                       + Send
-                       + 'static,
-    > + Unpin
-           + Clone
+                      + Send
+                      + 'static,
+    > + Clone
+           + Unpin
            + Send
            + 'static
     where
@@ -112,10 +109,7 @@ impl Config {
         I: tokio::io::AsyncRead + tokio::io::AsyncWrite + std::fmt::Debug + Unpin + Send + 'static,
     {
         let ProxyConfig {
-            dispatch_timeout,
-            cache_max_idle_age,
-            buffer_capacity,
-            ..
+            cache_max_idle_age, ..
         } = self.proxy;
 
         svc::stack(connect)
@@ -132,18 +126,6 @@ impl Config {
             .push_on_response(svc::layer::mk(tcp::Forward::new))
             .into_new_service()
             .check_new_service::<SocketAddr, I>()
-            .cache(
-                svc::layers().push_on_response(
-                    svc::layers()
-                        .push_failfast(dispatch_timeout)
-                        .push_spawn_buffer_with_idle_timeout(buffer_capacity, cache_max_idle_age)
-                ),
-            )
-            .into_make_service()
-            .spawn_buffer(buffer_capacity)
-            .push_make_ready()
-            .instrument(|_: &_| info_span!("tcp"))
-            .check_make_service::<SocketAddr, I>()
     }
 
     pub fn build_dns_refine(
@@ -327,7 +309,8 @@ impl Config {
                     .box_http_request(),
             )
             .push_spawn_ready()
-            .check_make_service::<HttpEndpoint, http::Request<_>>()
+            .into_new_service()
+            .check_new_service::<HttpEndpoint, http::Request<_>>()
             .push(discover)
             .check_service::<HttpConcrete>()
             .push_on_response(
@@ -379,36 +362,13 @@ impl Config {
             // it to inner stack to build the router and traffic split.
             .push(profiles::discover::layer(profiles_client))
             .into_new_service()
-            .cache(
-                svc::layers().push_on_response(
-                    svc::layers()
-                        .push_failfast(dispatch_timeout)
-                        .push_spawn_buffer_with_idle_timeout(buffer_capacity, cache_max_idle_age)
-                        .push(metrics.stack.layer(stack_labels("profile"))),
-                ),
-            )
-            .into_make_service()
-            .spawn_buffer(buffer_capacity)
-            .push_make_ready()
-            .check_make_service::<HttpLogical, http::Request<_>>();
+            .check_new_service::<HttpLogical, http::Request<_>>();
 
         // Caches clients that bypass discovery/balancing.
         let forward = svc::stack(endpoint)
-            .check_make_service::<HttpEndpoint, http::Request<http::boxed::Payload>>()
             .into_new_service()
-            .cache(
-                svc::layers().push_on_response(
-                    svc::layers()
-                        .push_failfast(dispatch_timeout)
-                        .push_spawn_buffer_with_idle_timeout(buffer_capacity, cache_max_idle_age)
-                        .box_http_request()
-                        .push(metrics.stack.layer(stack_labels("forward.endpoint"))),
-                ),
-            )
-            .into_make_service()
-            .spawn_buffer(buffer_capacity)
             .instrument(|t: &HttpEndpoint| debug_span!("forward", peer.id = ?t.identity))
-            .check_make_service::<HttpEndpoint, http::Request<_>>();
+            .check_new_service::<HttpEndpoint, http::Request<_>>();
 
         // Attempts to route route request to a logical services that uses
         // control plane for discovery. If the discovery is rejected, the
@@ -422,6 +382,17 @@ impl Config {
                     .into_inner(),
                 is_discovery_rejected,
             )
+            .cache(
+                svc::layers().push_on_response(
+                    svc::layers()
+                        .push_failfast(dispatch_timeout)
+                        .push_spawn_buffer_with_idle_timeout(buffer_capacity, cache_max_idle_age)
+                        .push(metrics.stack.layer(stack_labels("logical"))),
+                ),
+            )
+            .into_make_service()
+            .spawn_buffer(buffer_capacity)
+            .check_make_service::<HttpLogical, http::Request<_>>()
             .push(http::header_from_target::layer(CANONICAL_DST_HEADER))
             // Strips headers that may be set by this proxy.
             .push_on_response(http::strip_header::request::layer(DST_OVERRIDE_HEADER))
@@ -486,6 +457,8 @@ impl Config {
             dispatch_timeout,
             max_in_flight_requests,
             detect_protocol_timeout,
+            cache_max_idle_age,
+            buffer_capacity,
             ..
         } = self.proxy;
         let canonicalize_timeout = self.canonicalize_timeout;
@@ -536,9 +509,22 @@ impl Config {
         // Load balances TCP streams that cannot be decoded as HTTP.
         let tcp_balance = svc::stack(self.build_tcp_balance(tcp_connect, resolve))
             .push_fallback_with_predicate(
-                svc::stack(tcp_forward.clone()).push_map_target(TcpEndpoint::from),
+                svc::stack(tcp_forward.clone())
+                    .check_new::<TcpEndpoint>()
+                    .push_map_target(TcpEndpoint::from)
+                    .into_inner(),
                 is_discovery_rejected,
             )
+            .cache(
+                svc::layers().push_on_response(
+                    svc::layers()
+                        .push_failfast(dispatch_timeout)
+                        .push_spawn_buffer_with_idle_timeout(buffer_capacity, cache_max_idle_age),
+                ),
+            )
+            .into_make_service()
+            .spawn_buffer(buffer_capacity)
+            .instrument(|_: &_| info_span!("tcp"))
             .push_map_target(|a: listen::Addrs| a.target_addr())
             .into_inner();
 

linkerd/app/outbound/src/tests.rs

@@ -1,6 +1,8 @@
 use crate::Config;
 use futures::prelude::*;
-use linkerd2_app_core::{config, exp_backoff, proxy::http::h2, transport::listen, Addr, Error};
+use linkerd2_app_core::{
+    config, exp_backoff, proxy::http::h2, svc::NewService, transport::listen, Addr, Error,
+};
 use linkerd2_app_test as test_support;
 use std::{net::SocketAddr, time::Duration};
 use tower::ServiceExt;
@@ -68,14 +70,12 @@ async fn plaintext_tcp() {
     );
 
     // Build the outbound TCP balancer stack.
-    let make = cfg
+    let forward = cfg
         .build_tcp_balance(connect, resolver)
-        .oneshot(target_addr)
-        .err_into::<Error>()
-        .await
-        .expect("make service should succeed");
+        .new_service(target_addr);
 
-    make.oneshot(client_io)
+    forward
+        .oneshot(client_io)
         .err_into::<Error>()
         .await
         .expect("conn should succeed");

linkerd/io/src/boxed.rs

@@ -86,6 +86,12 @@ impl Io for BoxedIo {
     }
 }
 
+impl std::fmt::Debug for BoxedIo {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("BoxedIo").finish()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;