outbound: Reorganize outbound stacks

linkerd/app/core/src/lib.rs

@@ -122,7 +122,7 @@ impl From<Addr> for DiscoveryRejected {
     }
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Default)]
 pub struct SkipByPort(std::sync::Arc<indexmap::IndexSet<u16>>);
 
 impl From<indexmap::IndexSet<u16>> for SkipByPort {

linkerd/app/inbound/src/lib.rs

@@ -366,7 +366,7 @@ impl Config {
         // Forwards TCP streams that cannot be decoded as HTTP.
         let tcp_forward = svc::stack(tcp_connect)
             .push_make_thunk()
-            .push(svc::layer::mk(tcp::Forward::new));
+            .push_on_response(svc::layer::mk(tcp::Forward::new));
 
         let http = DetectHttp::new(
             h2_settings,

linkerd/app/outbound/src/lib.rs

@@ -6,7 +6,6 @@
 #![deny(warnings, rust_2018_idioms)]
 
 pub use self::endpoint::{HttpConcrete, HttpEndpoint, HttpLogical, LogicalPerRequest, TcpEndpoint};
-use ::http::header::HOST;
 use futures::{future, prelude::*};
 use linkerd2_app_core::{
     admit, classify,
@@ -21,8 +20,8 @@ use linkerd2_app_core::{
     spans::SpanConverter,
     svc::{self, NewService},
     transport::{self, listen, tls},
-    Addr, Conditional, DiscoveryRejected, Error, ProxyMetrics, StackMetrics, TraceContextLayer,
-    CANONICAL_DST_HEADER, DST_OVERRIDE_HEADER, L5D_REQUIRE_ID,
+    Addr, Conditional, DiscoveryRejected, Error, Never, ProxyMetrics, StackMetrics,
+    TraceContextLayer, CANONICAL_DST_HEADER, DST_OVERRIDE_HEADER, L5D_REQUIRE_ID,
 };
 use std::{
     collections::HashMap,
@@ -53,6 +52,7 @@ pub struct Config {
 impl Config {
     pub fn build_tcp_connect(
         &self,
+        prevent_loop: impl Into<PreventLoop>,
         local_identity: tls::Conditional<identity::Local>,
         metrics: &ProxyMetrics,
     ) -> impl tower::Service<
@@ -76,9 +76,75 @@ impl Config {
             // Limits the time we wait for a connection to be established.
             .push_timeout(self.proxy.connect.timeout)
             .push(metrics.transport.layer_connect(TransportLabels))
+            .push(admit::AdmitLayer::new(prevent_loop.into()))
             .into_inner()
     }
 
+    /// Constructs a TCP load balancer.
+    pub fn build_tcp_balance<C, E, I>(
+        &self,
+        connect: C,
+        resolve: E,
+    ) -> impl tower::Service<
+        SocketAddr,
+        Error = impl Into<Error>,
+        Future = impl Unpin + Send + 'static,
+        Response = impl tower::Service<
+            I,
+            Response = (),
+            Future = impl Unpin + Send + 'static,
+            Error = impl Into<Error>,
+        > + Unpin
+                       + Clone
+                       + Send
+                       + 'static,
+    > + Unpin
+           + Clone
+           + Send
+           + 'static
+    where
+        C: tower::Service<TcpEndpoint, Error = Error> + Unpin + Clone + Send + Sync + 'static,
+        C::Response: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
+        C::Future: Unpin + Send,
+        E: Resolve<Addr, Endpoint = proxy::api_resolve::Metadata> + Unpin + Clone + Send + 'static,
+        E::Future: Unpin + Send,
+        E::Resolution: Unpin + Send,
+        I: tokio::io::AsyncRead + tokio::io::AsyncWrite + std::fmt::Debug + Unpin + Send + 'static,
+    {
+        let ProxyConfig {
+            dispatch_timeout,
+            cache_max_idle_age,
+            buffer_capacity,
+            ..
+        } = self.proxy;
+
+        svc::stack(connect)
+            .push_make_thunk()
+            .instrument(|t: &TcpEndpoint| info_span!("endpoint", peer.addr = %t.addr, peer.id = ?t.identity))
+            .check_make_service::<TcpEndpoint, ()>()
+            .push(discover::resolve(map_endpoint::Resolve::new(
+                endpoint::FromMetadata,
+                resolve,
+            )))
+            .push(discover::buffer(1_000, cache_max_idle_age))
+            .push_map_target(Addr::from)
+            .push_on_response(tcp::balance::layer(EWMA_DEFAULT_RTT, EWMA_DECAY))
+            .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)
+                ),
+            )
+            .spawn_buffer(buffer_capacity)
+            .push_make_ready()
+            .instrument(|_: &_| info_span!("tcp"))
+            .check_make_service::<SocketAddr, I>()
+    }
+
     pub fn build_dns_refine(
         &self,
         dns_resolver: dns::Resolver,
@@ -119,14 +185,13 @@ impl Config {
 
     pub fn build_http_endpoint<B, C>(
         &self,
-        prevent_loop: impl Into<PreventLoop>,
         tcp_connect: C,
         tap_layer: tap::Layer,
         metrics: ProxyMetrics,
         span_sink: Option<mpsc::Sender<oc::Span>>,
     ) -> impl tower::Service<
         HttpEndpoint,
-        Error = Error,
+        Error = Never,
         Future = impl Unpin + Send,
         Response = impl tower::Service<
             http::Request<B>,
@@ -169,18 +234,24 @@ impl Config {
             // Re-establishes a connection when the client fails.
             .push(reconnect::layer({
                 let backoff = self.proxy.connect.backoff.clone();
-                move |_| Ok(backoff.stream())
+                move |e: Error| {
+                    if is_loop(&*e) {
+                        Err(e)
+                    } else {
+                        Ok(backoff.stream())
+                    }
+                }
             }))
-            .push(admit::AdmitLayer::new(prevent_loop.into()))
             .push(observability.clone())
             .push(identity_headers.clone())
             .push(http::override_authority::Layer::new(vec![
-                HOST.as_str(),
+                ::http::header::HOST.as_str(),
                 CANONICAL_DST_HEADER,
             ]))
             .push_on_response(svc::layers().box_http_response())
             .check_service::<HttpEndpoint>()
             .instrument(|e: &HttpEndpoint| info_span!("endpoint", peer.addr = %e.addr))
+            .into_inner()
     }
 
     pub fn build_http_router<B, E, S, R, P>(
@@ -205,12 +276,8 @@ impl Config {
     where
         B: http::HttpBody<Error = Error> + std::fmt::Debug + Default + Send + 'static,
         B::Data: Send + 'static,
-        E: tower::Service<HttpEndpoint, Error = Error, Response = S>
-            + Unpin
-            + Clone
-            + Send
-            + Sync
-            + 'static,
+        E: tower::Service<HttpEndpoint, Response = S> + Unpin + Clone + Send + Sync + 'static,
+        E::Error: Into<Error>,
         E::Future: Unpin + Send,
         S: tower::Service<
                 http::Request<http::boxed::Payload>,
@@ -319,6 +386,7 @@ impl Config {
                 ),
             )
             .spawn_buffer(buffer_capacity)
+            .push_make_ready()
             .check_make_service::<HttpLogical, http::Request<_>>();
 
         // Caches clients that bypass discovery/balancing.
@@ -343,7 +411,6 @@ impl Config {
         // `forward` stack is used instead, bypassing load balancing, etc.
         logical
             .push_on_response(svc::layers().box_http_response())
-            .push_make_ready()
             .push_fallback_with_predicate(
                 forward
                     .push_map_target(HttpEndpoint::from)
@@ -359,82 +426,6 @@ impl Config {
             .into_inner()
     }
 
-    /// Constructs a TCP load balancer.
-    pub fn build_tcp_balance<C, E, I>(
-        &self,
-        tcp_connect: &C,
-        resolve: E,
-        prevent_loop: PreventLoop,
-        metrics: &ProxyMetrics,
-    ) -> impl tower::Service<
-        SocketAddr,
-        Error = Error,
-        Future = impl Unpin + Send + 'static,
-        Response = impl tower::Service<
-            I,
-            Response = (),
-            Future = impl Unpin + Send + 'static,
-            Error = Error,
-        > + Unpin
-                       + Clone
-                       + Send
-                       + 'static,
-    > + Unpin
-           + Clone
-           + Send
-           + Sync
-           + 'static
-    where
-        C: tower::Service<TcpEndpoint, Error = Error> + Unpin + Clone + Send + Sync + 'static,
-        C::Response: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
-        C::Future: Unpin + Send,
-        E: Resolve<Addr, Endpoint = proxy::api_resolve::Metadata> + Unpin + Clone + Send + 'static,
-        E::Future: Unpin + Send,
-        E::Resolution: Unpin + Send,
-        I: tokio::io::AsyncRead + tokio::io::AsyncWrite + std::fmt::Debug + Unpin + Send + 'static,
-    {
-        let ProxyConfig {
-            dispatch_timeout,
-            cache_max_idle_age,
-            buffer_capacity,
-            ..
-        } = self.proxy;
-        svc::stack(tcp_connect.clone())
-            .push_make_thunk()
-            .instrument(|t: &TcpEndpoint| info_span!("endpoint", peer.addr = %t.addr, peer.id = ?t.identity))
-            .push(admit::AdmitLayer::new(prevent_loop))
-            .check_make_service::<TcpEndpoint, ()>()
-            .push(discover::resolve(map_endpoint::Resolve::new(
-                endpoint::FromMetadata,
-                resolve,
-            )))
-            .push(discover::buffer(1_000, cache_max_idle_age))
-            .push_map_target(Addr::from)
-            .push_on_response(tcp::balance::layer(EWMA_DEFAULT_RTT, EWMA_DECAY))
-            .push_fallback_with_predicate(
-                svc::stack(tcp_connect.clone())
-                    .push_make_thunk()
-                    .push(admit::AdmitLayer::new(prevent_loop))
-                    .push_map_target(TcpEndpoint::from)
-                    .instrument(|_: &SocketAddr| debug_span!("forward")),
-                is_discovery_rejected,
-            )
-            .into_new_service()
-            .check_new_service::<SocketAddr, ()>()
-            .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("tcp"))),
-                ),
-            )
-            .spawn_buffer(buffer_capacity)
-            .check_make_service::<SocketAddr, ()>()
-            .push(svc::layer::mk(tcp::Forward::new))
-            .instrument(|_: &SocketAddr| info_span!("tcp"))
-    }
-
     pub async fn build_server<E, R, C, H, S>(
         self,
         listen_addr: std::net::SocketAddr,
@@ -483,13 +474,6 @@ impl Config {
             ..
         } = self.proxy;
         let canonicalize_timeout = self.canonicalize_timeout;
-        let prevent_loop = PreventLoop::from(listen_addr.port());
-
-        // Load balances TCP streams that cannot be decoded as HTTP.
-        let tcp_balance =
-            svc::stack(self.build_tcp_balance(&tcp_connect, resolve, prevent_loop, &metrics))
-                .push_map_target(|a: listen::Addrs| a.target_addr())
-                .into_inner();
 
         let http_admit_request = svc::layers()
             // Limits the number of in-flight requests.
@@ -510,7 +494,6 @@ impl Config {
             // its canonical FQDN to use for routing.
             .push(http::canonicalize::Layer::new(refine, canonicalize_timeout))
             .check_make_service::<HttpLogical, http::Request<_>>()
-            .push_make_ready()
             .push_timeout(dispatch_timeout)
             .push(router::Layer::new(LogicalPerRequest::from))
             .check_new_service::<listen::Addrs, http::Request<_>>()
@@ -529,6 +512,21 @@ impl Config {
             .into_inner()
             .into_make_service();
 
+        let tcp_forward = svc::stack(tcp_connect.clone())
+            .push_make_thunk()
+            .push_on_response(svc::layer::mk(tcp::Forward::new))
+            .instrument(|_: &TcpEndpoint| info_span!("forward"))
+            .check_service::<TcpEndpoint>();
+
+        // 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),
+                is_discovery_rejected,
+            )
+            .push_map_target(|a: listen::Addrs| a.target_addr())
+            .into_inner();
+
         let http = http::DetectHttp::new(
             h2_settings,
             detect_protocol_timeout,
@@ -537,16 +535,10 @@ impl Config {
             drain.clone(),
         );
 
-        let tcp_forward = svc::stack(tcp_connect)
-            .push_make_thunk()
-            .push(svc::layer::mk(tcp::Forward::new))
-            .push(admit::AdmitLayer::new(prevent_loop))
-            .push_map_target(TcpEndpoint::from);
-
         let accept = svc::stack(svc::stack::MakeSwitch::new(
             skip_detect.clone(),
             http,
-            tcp_forward,
+            tcp_forward.push_map_target(TcpEndpoint::from),
         ))
         .push(metrics.transport.layer_accept(TransportLabels));
 
@@ -604,3 +596,7 @@ fn is_discovery_rejected(err: &Error) -> bool {
     tracing::debug!(rejected, %err);
     rejected
 }
+
+fn is_loop(err: &(dyn std::error::Error + 'static)) -> bool {
+    err.is::<prevent_loop::LoopPrevented>() || err.source().map(is_loop).unwrap_or(false)
+}