internal: Decouple TCP forwarding from protocol dispatch (#389)

In preparation of adding discovery logic to outbound TCP forwarding,
this change extracts the TCP forwarding logic as an Accept so that the
outbound proxy will be able to provide an alternate implementation.

Author: olix0r

Cargo.lock

@@ -563,6 +563,7 @@ dependencies = [
  "linkerd2-proxy-identity 0.1.0",
  "linkerd2-proxy-resolve 0.1.0",
  "linkerd2-proxy-tap 0.1.0",
+ "linkerd2-proxy-tcp 0.1.0",
  "linkerd2-proxy-transport 0.1.0",
  "linkerd2-reconnect 0.1.0",
  "linkerd2-request-filter 0.1.0",
@@ -951,6 +952,18 @@ dependencies = [
  "tracing-futures 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "linkerd2-proxy-tcp"
+version = "0.1.0"
+dependencies = [
+ "futures 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)",
+ "linkerd2-duplex 0.1.0",
+ "linkerd2-error 0.1.0",
+ "tokio 0.1.20 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tower 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tower-load 0.1.0 (git+https://github.com/tower-rs/tower)",
+]
+
 [[package]]
 name = "linkerd2-proxy-transport"
 version = "0.1.0"

Cargo.toml

@@ -27,6 +27,7 @@ members = [
     "linkerd/proxy/identity",
     "linkerd/proxy/resolve",
     "linkerd/proxy/tap",
+    "linkerd/proxy/tcp",
     "linkerd/proxy/transport",
     "linkerd/request-filter",
     "linkerd/reconnect",

linkerd/app/core/Cargo.toml

@@ -36,6 +36,7 @@ linkerd2-proxy-identity = { path = "../../proxy/identity" }
 linkerd2-proxy-http = { path = "../../proxy/http" }
 linkerd2-proxy-resolve = { path = "../../proxy/resolve" }
 linkerd2-proxy-tap = { path = "../../proxy/tap" }
+linkerd2-proxy-tcp = { path = "../../proxy/tcp" }
 linkerd2-proxy-transport = { path = "../../proxy/transport" }
 linkerd2-reconnect = { path = "../../reconnect" }
 linkerd2-request-filter = { path = "../../request-filter" }

linkerd/app/core/src/proxy/mod.rs

@@ -8,10 +8,10 @@ pub use linkerd2_proxy_http::{self as http, grpc};
 pub use linkerd2_proxy_identity as identity;
 pub use linkerd2_proxy_resolve as resolve;
 pub use linkerd2_proxy_tap as tap;
+pub use linkerd2_proxy_tcp as tcp;
 
 pub mod buffer;
 pub mod pending;
 pub mod server;
-mod tcp;
 
 pub use self::server::Server;

linkerd/app/core/src/proxy/server.rs

@@ -1,25 +1,23 @@
-use crate::proxy::http::{
-    glue::{HttpBody, HyperServerSvc},
-    upgrade, Version as HttpVersion,
+use crate::{
+    drain,
+    proxy::{
+        core::Accept,
+        detect,
+        http::{
+            glue::{HttpBody, HyperServerSvc},
+            h2::Settings as H2Settings,
+            upgrade, Version as HttpVersion,
+        },
+    },
+    svc::{MakeService, Service, ServiceExt},
+    transport::{self, io::BoxedIo, labels::Key as TransportKey, metrics::TransportLabels, tls},
+    Error, Never,
 };
-use crate::proxy::{detect, tcp};
-use crate::svc::{MakeService, Service};
-use crate::transport::{
-    self, io::BoxedIo, labels::Key as TransportKey, metrics::TransportLabels, tls,
-};
-use futures::future::{self, Either};
-use futures::{Future, Poll};
+use futures::{future::Either, Future, Poll};
 use http;
 use hyper;
 use indexmap::IndexSet;
-use linkerd2_drain as drain;
-use linkerd2_error::{Error, Never};
-use linkerd2_proxy_core::listen::Accept;
-use linkerd2_proxy_http::h2::Settings as H2Settings;
-use std::net::SocketAddr;
 use std::sync::Arc;
-use std::{error, fmt};
-use tokio::io::{AsyncRead, AsyncWrite};
 use tracing::{info_span, trace};
 use tracing_futures::Instrument;
 
@@ -77,7 +75,7 @@ impl detect::Detect<tls::accept::Meta> for ProtocolDetect {
 ///
 /// *  Otherwise, an `H`-typed `Service` is used to build a service that
 ///    can route HTTP  requests for the `tls::accept::Meta`.
-pub struct Server<L, C, H, B>
+pub struct Server<L, F, H, B>
 where
     // Used when forwarding a TCP stream (e.g. with telemetry, timeouts).
     L: TransportLabels<Protocol, Labels = TransportKey>,
@@ -94,57 +92,12 @@ where
     h2_settings: H2Settings,
     transport_labels: L,
     transport_metrics: transport::MetricsRegistry,
-    connect: ForwardConnect<C>,
+    forward_tcp: F,
     make_http: H,
     drain: drain::Watch,
 }
 
-/// Establishes connections for forwarded connections.
-///
-/// Fails to produce a `Connect` if this would connect to the listener that
-/// already accepted this.
-#[derive(Clone, Debug)]
-struct ForwardConnect<C>(C);
-
-/// An error indicating an accepted socket did not have an SO_ORIGINAL_DST
-/// address and therefore could not be forwarded.
-#[derive(Clone, Debug)]
-pub struct NoForwardTarget;
-
-impl<C> Service<tls::accept::Meta> for ForwardConnect<C>
-where
-    C: Service<SocketAddr>,
-    C::Error: Into<Error>,
-{
-    type Response = C::Response;
-    type Error = Error;
-    type Future = future::Either<
-        future::FutureResult<C::Response, Error>,
-        future::MapErr<C::Future, fn(C::Error) -> Error>,
-    >;
-
-    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
-        self.0.poll_ready().map_err(Into::into)
-    }
-
-    fn call(&mut self, meta: tls::accept::Meta) -> Self::Future {
-        if meta.addrs.target_addr_is_local() {
-            return future::Either::A(future::err(NoForwardTarget.into()));
-        }
-
-        future::Either::B(self.0.call(meta.addrs.target_addr()).map_err(Into::into))
-    }
-}
-
-impl error::Error for NoForwardTarget {}
-
-impl fmt::Display for NoForwardTarget {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "Could not forward to a target address")
-    }
-}
-
-impl<L, C, H, B> Server<L, C, H, B>
+impl<L, F, H, B> Server<L, F, H, B>
 where
     L: TransportLabels<Protocol, Labels = TransportKey>,
     H: MakeService<
@@ -160,7 +113,7 @@ where
     pub fn new(
         transport_labels: L,
         transport_metrics: transport::MetricsRegistry,
-        connect: C,
+        forward_tcp: F,
         make_http: H,
         h2_settings: H2Settings,
         drain: drain::Watch,
@@ -173,21 +126,19 @@ where
                 h2_settings,
                 transport_labels,
                 transport_metrics,
-                connect: ForwardConnect(connect),
+                forward_tcp,
                 make_http,
                 drain,
             },
         )
     }
 }
 
-impl<L, C, H, B> Service<Connection> for Server<L, C, H, B>
+impl<L, F, H, B> Service<Connection> for Server<L, F, H, B>
 where
     L: TransportLabels<Protocol, Labels = TransportKey>,
-    C: Service<SocketAddr> + Clone + Send + 'static,
-    C::Response: AsyncRead + AsyncWrite + Send + 'static,
-    C::Future: Send + 'static,
-    C::Error: Into<Error>,
+    F: Accept<(tls::accept::Meta, transport::metrics::Io<BoxedIo>)> + Clone + Send + 'static,
+    F::Future: Send + 'static,
     H: MakeService<
             tls::accept::Meta,
             http::Request<HttpBody>,
@@ -228,8 +179,12 @@ where
             Some(http) => http,
             None => {
                 trace!("did not detect protocol; forwarding TCP");
-                let fwd = tcp::forward(io, self.connect.clone(), proto.tls);
-                return Box::new(drain.watch(fwd, |_| {}));
+                let fwd = self
+                    .forward_tcp
+                    .clone()
+                    .into_service()
+                    .oneshot((proto.tls, io));
+                return Box::new(drain.watch(fwd.map_err(Into::into), |_| {}));
             }
         };
 
@@ -279,10 +234,10 @@ where
     }
 }
 
-impl<L, C, H, B> Clone for Server<L, C, H, B>
+impl<L, F, H, B> Clone for Server<L, F, H, B>
 where
     L: TransportLabels<Protocol, Labels = TransportKey> + Clone,
-    C: Clone,
+    F: Clone,
     H: MakeService<
             tls::accept::Meta,
             http::Request<HttpBody>,
@@ -297,7 +252,7 @@ where
             h2_settings: self.h2_settings.clone(),
             transport_labels: self.transport_labels.clone(),
             transport_metrics: self.transport_metrics.clone(),
-            connect: self.connect.clone(),
+            forward_tcp: self.forward_tcp.clone(),
             make_http: self.make_http.clone(),
             drain: self.drain.clone(),
         }

linkerd/app/core/src/proxy/tcp.rs

@@ -22,7 +22,7 @@ use linkerd2_app_core::{
         },
         identity,
         server::{Protocol as ServerProtocol, Server},
-        tap,
+        tap, tcp,
     },
     reconnect, serve,
     spans::SpanConverter,
@@ -269,12 +269,18 @@ impl<A: OrigDstAddr> Config<A> {
                 .push(metrics.http_handle_time.layer())
                 .serves::<tls::accept::Meta>();
 
+            let forward_tcp = tcp::Forward::new(
+                svc::stack(connect_stack)
+                    .push(svc::map_target::layer(|meta: tls::accept::Meta| {
+                        Endpoint::from(meta.addrs.target_addr())
+                    }))
+                    .into_inner(),
+            );
+
             let server = Server::new(
                 TransportLabels,
                 metrics.transport,
-                svc::stack(connect_stack)
-                    .push(svc::map_target::layer(Endpoint::from))
-                    .into_inner(),
+                forward_tcp,
                 source_stack,
                 h2_settings,
                 drain.clone(),

linkerd/app/inbound/src/lib.rs

@@ -25,7 +25,7 @@ use linkerd2_app_core::{
         },
         identity,
         resolve::map_endpoint,
-        tap, Server,
+        tap, tcp, Server,
     },
     reconnect, serve,
     spans::SpanConverter,
@@ -329,12 +329,18 @@ impl<A: OrigDstAddr> Config<A> {
             ))
                 .push(metrics.http_handle_time.layer());
 
+            let forward_tcp = tcp::Forward::new(
+                svc::stack(connect_stack)
+                    .push(svc::map_target::layer(|meta: tls::accept::Meta| {
+                        Endpoint::from(meta.addrs.target_addr())
+                    }))
+                    .into_inner(),
+            );
+
             let proxy = Server::new(
                 TransportLabels,
                 metrics.transport,
-                svc::stack(connect_stack)
-                    .push(svc::map_target::layer(Endpoint::from))
-                    .into_inner(),
+                forward_tcp,
                 server_stack,
                 h2_settings,
                 drain.clone(),

linkerd/app/outbound/src/lib.rs

@@ -42,8 +42,18 @@ pub trait Accept<C> {
     fn poll_ready(&mut self) -> Poll<(), Self::Error>;
 
     fn accept(&mut self, connection: C) -> Self::Future;
+
+    fn into_service(self) -> AcceptService<Self>
+    where
+        Self: Sized,
+    {
+        AcceptService(self)
+    }
 }
 
+#[derive(Clone, Debug)]
+pub struct AcceptService<S>(S);
+
 impl<C, S> Accept<C> for S
 where
     S: Service<C, Response = ()>,
@@ -52,17 +62,29 @@ where
     type Error = S::Error;
     type Future = S::Future;
 
-    #[inline]
     fn poll_ready(&mut self) -> Poll<(), Self::Error> {
         Service::poll_ready(self)
     }
 
-    #[inline]
     fn accept(&mut self, connection: C) -> Self::Future {
         Service::call(self, connection)
     }
 }
 
+impl<C, S: Accept<C>> Service<C> for AcceptService<S> {
+    type Response = ();
+    type Error = S::Error;
+    type Future = S::Future;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        self.0.poll_ready()
+    }
+
+    fn call(&mut self, connection: C) -> Self::Future {
+        self.0.accept(connection)
+    }
+}
+
 pub struct Serve<L, A, E> {
     listen: L,
     accept: A,

linkerd/proxy/core/src/listen.rs

@@ -0,0 +1,15 @@
+[package]
+name = "linkerd2-proxy-tcp"
+version = "0.1.0"
+authors = ["Linkerd Developers <[email protected]>"]
+edition = "2018"
+publish = false
+
+
+[dependencies]
+futures = "0.1"
+linkerd2-duplex = { path = "../../duplex" }
+linkerd2-error = { path = "../../error" }
+tokio = "0.1.14"
+tower = "0.1"
+tower-load = { git = "https://github.com/tower-rs/tower" }