inbound: Reorganize server into smaller stacks (#1156)

The inbound server stack is fairly large. While sketching out changes to
support inbound policy, it's helpful to split this larger stack into a
few smaller components (using the recently-introduced `map_stack`
helper).

This change:

* Modifies `Inbound::push_http_server` to return a boxed service. This
  service was being boxed by `Inbound::into_server`. This change helps
  reduce boilerplate in our test utilities, too. (In fact, this change
  requires that we eliminate some unnecessary type constraints).
* Changes `Inbound::to_tcp_connect` to `Inbound::into_tcp_connect`. The
  internal cloning is not necessary.
* Changes `Inbound::into_server` to `Inbound::push_server`. Internally,
  this function now uses several calls to `map_stack` to build the
  inbound server stack.

Author: olix0r

linkerd/app/inbound/src/http/mod.rs

@@ -25,12 +25,7 @@ use tracing::debug_span;
 impl<H> Inbound<H> {
     pub fn push_http_server<T, I, HSvc>(
         self,
-    ) -> Inbound<
-        svc::BoxNewService<
-            T,
-            impl svc::Service<I, Response = (), Error = Error, Future = impl Send> + Clone,
-        >,
-    >
+    ) -> Inbound<svc::BoxNewService<T, svc::BoxService<I, (), Error>>>
     where
         T: Param<Version>
             + Param<http::normalize_uri::DefaultAuthority>
@@ -84,6 +79,7 @@ impl<H> Inbound<H> {
                 .check_new_service::<T, http::Request<_>>()
                 .instrument(|t: &T| debug_span!("http", v=%Param::<Version>::param(t)))
                 .push(http::NewServeHttp::layer(h2_settings, rt.drain.clone()))
+                .push_on_response(svc::BoxService::layer())
                 .push(svc::BoxNewService::layer())
         })
     }
@@ -355,16 +351,7 @@ pub mod fuzz_logic {
         rt: ProxyRuntime,
         profiles: resolver::Profiles,
         connect: Connect<Remote<ServerAddr>>,
-    ) -> impl svc::NewService<
-        HttpAccept,
-        Service = impl tower::Service<
-            I,
-            Response = (),
-            Error = impl Into<linkerd_app_core::Error>,
-            Future = impl Send + 'static,
-        > + Send
-                      + Clone,
-    > + Clone
+    ) -> svc::BoxNewService<HttpAccept, svc::BoxService<I, (), linkerd_app_core::Error>>
     where
         I: io::AsyncRead + io::AsyncWrite + io::PeerAddr + Send + Unpin + 'static,
     {

linkerd/app/inbound/src/http/tests.rs

@@ -13,7 +13,7 @@ use linkerd_app_core::{
     svc::{self, NewService, Param},
     tls,
     transport::{ClientAddr, Remote, ServerAddr},
-    Conditional, NameAddr, ProxyRuntime,
+    Conditional, Error, NameAddr, ProxyRuntime,
 };
 use linkerd_app_test::connect::ConnectFuture;
 use linkerd_tracing::test::trace_init;
@@ -24,16 +24,7 @@ fn build_server<I>(
     rt: ProxyRuntime,
     profiles: resolver::Profiles,
     connect: Connect<Remote<ServerAddr>>,
-) -> svc::BoxNewService<
-    HttpAccept,
-    impl tower::Service<
-            I,
-            Response = (),
-            Error = impl Into<linkerd_app_core::Error>,
-            Future = impl Send + 'static,
-        > + Send
-        + Clone,
->
+) -> svc::BoxNewService<HttpAccept, svc::BoxService<I, (), Error>>
 where
     I: io::AsyncRead + io::AsyncWrite + io::PeerAddr + Send + Unpin + 'static,
 {

linkerd/app/inbound/src/lib.rs

@@ -97,27 +97,32 @@ impl Inbound<()> {
 
     /// Readies the inbound stack to make TCP connections (for both TCP
     // forwarding and HTTP proxying).
-    pub fn to_tcp_connect<T: svc::Param<u16> + 'static>(
-        &self,
+    pub fn into_tcp_connect<T>(
+        self,
     ) -> Inbound<
         impl svc::Service<
                 T,
                 Response = impl io::AsyncRead + io::AsyncWrite + Send,
                 Error = Error,
                 Future = impl Send,
             > + Clone,
-    > {
-        self.clone().map_stack(|config, _, _| {
+    >
+    where
+        T: svc::Param<u16> + 'static,
+    {
+        self.map_stack(|config, _, _| {
             // Establishes connections to remote peers (for both TCP
             // forwarding and HTTP proxying).
             let ConnectConfig {
-                keepalive, timeout, ..
-            } = config.proxy.connect.clone();
+                ref keepalive,
+                ref timeout,
+                ..
+            } = config.proxy.connect;
 
-            svc::stack(transport::ConnectTcp::new(keepalive))
+            svc::stack(transport::ConnectTcp::new(*keepalive))
                 .push_map_target(|t: T| Remote(ServerAddr(([127, 0, 0, 1], t.param()).into())))
                 // Limits the time we wait for a connection to be established.
-                .push_connect_timeout(timeout)
+                .push_connect_timeout(*timeout)
                 .push(svc::stack::BoxFuture::layer())
         })
     }
@@ -142,19 +147,20 @@ impl Inbound<()> {
         P::Error: Send,
         P::Future: Send,
     {
-        let (listen_addr, listen) = bind
+        let (Local(ServerAddr(la)), listen) = bind
             .bind(&self.config.proxy.server)
             .expect("Failed to bind inbound listener");
 
         let serve = async move {
-            let stack =
-                self.to_tcp_connect()
-                    .into_server(listen_addr.as_ref().port(), profiles, gateway);
-            let shutdown = self.runtime.drain.signaled();
+            let shutdown = self.runtime.drain.clone().signaled();
+            let stack = self
+                .into_tcp_connect()
+                .push_server(la.port(), profiles, gateway)
+                .into_inner();
             serve::serve(listen, stack, shutdown).await
         };
 
-        (listen_addr, serve)
+        (Local(ServerAddr(la)), serve)
     }
 }
 
@@ -199,12 +205,12 @@ where
         })
     }
 
-    pub fn into_server<T, I, G, GSvc, P>(
+    pub fn push_server<T, I, G, GSvc, P>(
         self,
         server_port: u16,
         profiles: P,
         gateway: G,
-    ) -> svc::BoxNewService<T, svc::BoxService<I, (), Error>>
+    ) -> Inbound<svc::BoxNewService<T, svc::BoxService<I, (), Error>>>
     where
         T: svc::Param<Remote<ClientAddr>> + svc::Param<OrigDstAddr>,
         T: Clone + Send + 'static,
@@ -219,86 +225,93 @@ where
         P::Error: Send,
         P::Future: Send,
     {
-        let Self {
-            config:
-                Config {
-                    proxy: config,
-                    require_identity_for_inbound_ports: require_id,
-                    disable_protocol_detection_for_ports: disable_detect,
-                    ..
-                },
-            runtime: rt,
-            stack: _,
-        } = self.clone();
+        // Handles inbound connections that target an opaque port.
+        let opaque = self
+            .clone()
+            .push_tcp_forward(server_port)
+            .map_stack(|_, rt, tcp| {
+                tcp.push_map_target(TcpEndpoint::from)
+                    .push(rt.metrics.transport.layer_accept())
+                    .check_new_service::<TcpAccept, _>()
+            })
+            .into_stack();
+
+        // Handles inbound connections that could not be detected as HTTP.
+        let tcp = self.clone().push_tcp_forward(server_port);
 
-        self.clone()
-            .push_http_router(profiles)
+        // Handles connections targeting the inbound proxy port--either by acting as a gateway to
+        // the outbound stack or by forwarding connections locally (for opauque transport).
+        let direct = tcp
+            .clone()
+            .push_direct(gateway)
+            .into_stack()
+            .instrument(|_: &_| debug_span!("direct"));
+
+        self.push_http_router(profiles)
             .push_http_server()
-            .stack
-            .push_map_target(HttpAccept::from)
-            .push(svc::UnwrapOr::layer(
-                // When HTTP detection fails, forward the connection to the
-                // application as an opaque TCP stream.
-                self.clone()
-                    .push_tcp_forward(server_port)
-                    .into_stack()
-                    .push_map_target(TcpEndpoint::from)
-                    .push_on_response(svc::BoxService::layer())
-                    .into_inner(),
-            ))
-            .push_on_response(svc::BoxService::layer())
-            .push_map_target(detect::allow_timeout)
-            .push(svc::BoxNewService::layer())
-            .push(detect::NewDetectService::layer(
-                config.detect_protocol_timeout,
-                http::DetectHttp::default(),
-            ))
-            .push_request_filter(require_id)
-            .push(rt.metrics.transport.layer_accept())
-            .push_request_filter(TcpAccept::try_from)
-            .push(svc::BoxNewService::layer())
-            .push(tls::NewDetectTls::layer(
-                rt.identity.clone(),
-                config.detect_protocol_timeout,
-            ))
-            .instrument(|_: &_| debug_span!("proxy"))
-            .push_switch(
-                move |t: T| {
-                    let OrigDstAddr(addr) = t.param();
-                    if !disable_detect.contains(&addr.port()) {
-                        Ok::<_, Never>(svc::Either::A(t))
-                    } else {
-                        Ok(svc::Either::B(TcpAccept::port_skipped(t)))
-                    }
-                },
-                self.clone()
-                    .push_tcp_forward(server_port)
-                    .stack
-                    .push_map_target(TcpEndpoint::from)
+            .map_stack(|cfg, rt, http| {
+                let detect_timeout = cfg.proxy.detect_protocol_timeout;
+                let require_id = cfg.require_identity_for_inbound_ports.clone();
+
+                http.push_map_target(HttpAccept::from)
+                    .push(svc::UnwrapOr::layer(
+                        // When HTTP detection fails, forward the connection to the application as
+                        // an opaque TCP stream.
+                        tcp.into_stack()
+                            .push_map_target(TcpEndpoint::from)
+                            .push_on_response(svc::BoxService::layer())
+                            .into_inner(),
+                    ))
+                    .push_map_target(detect::allow_timeout)
+                    .push(svc::BoxNewService::layer())
+                    .push(detect::NewDetectService::layer(
+                        detect_timeout,
+                        http::DetectHttp::default(),
+                    ))
+                    .push_request_filter(require_id)
                     .push(rt.metrics.transport.layer_accept())
-                    .check_new_service::<TcpAccept, _>()
-                    .instrument(|_: &TcpAccept| debug_span!("forward"))
-                    .into_inner(),
-            )
-            .check_new_service::<T, I>()
-            .push_on_response(svc::BoxService::layer())
-            .push(svc::BoxNewService::layer())
-            .push_switch(
-                PreventLoop::from(server_port).to_switch(),
-                self.push_tcp_forward(server_port)
-                    .push_direct(gateway)
-                    .stack
-                    .instrument(|_: &_| debug_span!("direct"))
-                    .into_inner(),
-            )
-            .instrument(|a: &T| {
-                let OrigDstAddr(target_addr) = a.param();
-                info_span!("server", port = target_addr.port())
+                    .push_request_filter(TcpAccept::try_from)
+                    .push(svc::BoxNewService::layer())
+                    .push(tls::NewDetectTls::layer(
+                        rt.identity.clone(),
+                        detect_timeout,
+                    ))
+            })
+            .map_stack(|cfg, _, detect| {
+                let disable_detect = cfg.disable_protocol_detection_for_ports.clone();
+                detect
+                    .instrument(|_: &_| debug_span!("proxy"))
+                    .push_switch(
+                        move |t: T| {
+                            let OrigDstAddr(addr) = t.param();
+                            if !disable_detect.contains(&addr.port()) {
+                                Ok::<_, Never>(svc::Either::A(t))
+                            } else {
+                                Ok(svc::Either::B(TcpAccept::port_skipped(t)))
+                            }
+                        },
+                        opaque
+                            .instrument(|_: &TcpAccept| debug_span!("forward"))
+                            .into_inner(),
+                    )
+                    .check_new_service::<T, I>()
+                    .push_on_response(svc::BoxService::layer())
+                    .push(svc::BoxNewService::layer())
+            })
+            .map_stack(|_, rt, accept| {
+                accept
+                    .push_switch(
+                        PreventLoop::from(server_port).to_switch(),
+                        direct.into_inner(),
+                    )
+                    .instrument(|a: &T| {
+                        let OrigDstAddr(target_addr) = a.param();
+                        info_span!("server", port = target_addr.port())
+                    })
+                    .push(rt.metrics.tcp_accept_errors.layer())
+                    .push_on_response(svc::BoxService::layer())
+                    .push(svc::BoxNewService::layer())
             })
-            .push(rt.metrics.tcp_accept_errors.layer())
-            .push_on_response(svc::BoxService::layer())
-            .push(svc::BoxNewService::layer())
-            .into_inner()
     }
 }
 

linkerd/app/test/src/http_util.rs

@@ -7,6 +7,7 @@ use hyper::{
     client::conn::{Builder as ClientBuilder, SendRequest},
     Body, Request, Response,
 };
+use linkerd_app_core::svc::BoxService;
 use std::{
     future::Future,
     sync::{Arc, Mutex},
@@ -22,6 +23,8 @@ pub struct Server {
 
 type HandleFuture = Box<dyn (FnMut(Request<Body>) -> Result<Response<Body>, Error>) + Send>;
 
+type BoxServer = BoxService<io::DuplexStream, (), Error>;
+
 impl Default for Server {
     fn default() -> Self {
         Self {
@@ -36,18 +39,11 @@ impl Default for Server {
     }
 }
 
-pub async fn run_proxy<S>(mut server: S) -> (io::DuplexStream, JoinHandle<Result<(), Error>>)
-where
-    S: tower::Service<io::DuplexStream> + Send + Sync + 'static,
-    S::Error: Into<Error>,
-    S::Response: std::fmt::Debug + Send + Sync + 'static,
-    S::Future: Send,
-{
+pub async fn run_proxy(mut server: BoxServer) -> (io::DuplexStream, JoinHandle<Result<(), Error>>) {
     let (client_io, server_io) = io::duplex(4096);
     let f = server
         .ready()
         .await
-        .map_err(Into::into)
         .expect("proxy server failed to become ready")
         .call(server_io);
 
@@ -79,16 +75,10 @@ pub async fn connect_client(
     (client, tokio::spawn(client_bg))
 }
 
-pub async fn connect_and_accept<S>(
+pub async fn connect_and_accept(
     client_settings: &mut ClientBuilder,
-    server: S,
-) -> (SendRequest<Body>, impl Future<Output = Result<(), Error>>)
-where
-    S: tower::Service<io::DuplexStream> + Send + Sync + 'static,
-    S::Error: Into<Error>,
-    S::Response: std::fmt::Debug + Send + Sync + 'static,
-    S::Future: Send,
-{
+    server: BoxServer,
+) -> (SendRequest<Body>, impl Future<Output = Result<(), Error>>) {
     tracing::info!(settings = ?client_settings, "connecting client with");
     let (client_io, proxy) = run_proxy(server).await;
     let (client, client_bg) = connect_client(client_settings, client_io).await;