outbound: Decouple the TCP connect stack from the target type (#951)

In order to make the server and logical stack more flexible, we need to
make stacks generic over their input target type.

This change modifies the innermost `tcp::connect` stack to not depend on
a concrete target type, instead relying on `Param` constraints. A new
`tcp::Connect` target type is introduced to be constructed by the TCP
connect stack.

Author: olix0r

linkerd/app/gateway/src/lib.rs

@@ -79,15 +79,11 @@ pub fn stack<I, O, P, R>(
        + Send
 where
     I: io::AsyncRead + io::AsyncWrite + io::PeerAddr + fmt::Debug + Send + Sync + Unpin + 'static,
-    O: svc::Service<outbound::http::Endpoint, Error = io::Error>
-        + svc::Service<outbound::tcp::Endpoint, Error = io::Error>,
     O: Clone + Send + Sync + Unpin + 'static,
-    <O as svc::Service<outbound::http::Endpoint>>::Response:
+    O: svc::Service<outbound::tcp::Connect, Error = io::Error>,
+    O::Response:
         io::AsyncRead + io::AsyncWrite + tls::HasNegotiatedProtocol + Send + Unpin + 'static,
-    <O as svc::Service<outbound::http::Endpoint>>::Future: Send + Unpin + 'static,
-    <O as svc::Service<outbound::tcp::Endpoint>>::Response:
-        io::AsyncRead + io::AsyncWrite + tls::HasNegotiatedProtocol + Send + Unpin + 'static,
-    <O as svc::Service<outbound::tcp::Endpoint>>::Future: Send + Unpin + 'static,
+    O::Future: Send + Unpin + 'static,
     P: profiles::GetProfile<profiles::LogicalAddr> + Clone + Send + Sync + Unpin + 'static,
     P::Future: Send + 'static,
     P::Error: Send,

linkerd/app/outbound/src/http/endpoint.rs

@@ -8,13 +8,7 @@ use linkerd_app_core::{
 use tokio::io;
 use tracing::debug_span;
 
-impl<C> Outbound<C>
-where
-    C: svc::Service<Endpoint> + Clone + Send + Sync + Unpin + 'static,
-    C::Response: io::AsyncRead + io::AsyncWrite + Send + Unpin,
-    C::Error: Into<Error>,
-    C::Future: Send + Unpin,
-{
+impl<C> Outbound<C> {
     pub fn push_http_endpoint<B>(
         self,
     ) -> Outbound<
@@ -31,6 +25,10 @@ where
     where
         B: http::HttpBody<Error = Error> + std::fmt::Debug + Default + Send + 'static,
         B::Data: Send + 'static,
+        C: svc::Service<Endpoint> + Clone + Send + Sync + Unpin + 'static,
+        C::Response: io::AsyncRead + io::AsyncWrite + Send + Unpin,
+        C::Error: Into<Error>,
+        C::Future: Send + Unpin,
     {
         let Self {
             config,

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

@@ -135,15 +135,6 @@ impl Param<Option<SessionProtocol>> for Endpoint {
     }
 }
 
-impl Param<Option<AuthorityOverride>> for Endpoint {
-    fn param(&self) -> Option<AuthorityOverride> {
-        self.metadata
-            .authority_override()
-            .cloned()
-            .map(AuthorityOverride)
-    }
-}
-
 impl tap::Inspect for Endpoint {
     fn src_addr<B>(&self, req: &Request<B>) -> Option<SocketAddr> {
         req.extensions().get::<ClientHandle>().map(|c| c.addr)

linkerd/app/outbound/src/lib.rs

@@ -101,17 +101,11 @@ impl<S> Outbound<S> {
         Service = impl svc::Service<I, Response = (), Error = Error, Future = impl Send>,
     >
     where
-        S: svc::Service<http::Endpoint, Error = io::Error>
-            + svc::Service<tcp::Endpoint, Error = io::Error>,
         S: Clone + Send + Sync + Unpin + 'static,
-        <S as svc::Service<http::Endpoint>>::Response: tls::HasNegotiatedProtocol,
-        <S as svc::Service<http::Endpoint>>::Response:
-            tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin,
-        <S as svc::Service<http::Endpoint>>::Future: Send + Unpin,
-        <S as svc::Service<tcp::Endpoint>>::Response: tls::HasNegotiatedProtocol,
-        <S as svc::Service<tcp::Endpoint>>::Response:
-            tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin,
-        <S as svc::Service<tcp::Endpoint>>::Future: Send,
+        S: svc::Service<tcp::Connect, Error = io::Error>,
+        S::Response: tls::HasNegotiatedProtocol,
+        S::Response: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin,
+        S::Future: Send + Unpin,
         R: Clone + Send + 'static,
         R: Resolve<ConcreteAddr, Endpoint = Metadata, Error = Error>,
         R::Resolution: Send,

linkerd/app/outbound/src/target.rs

@@ -1,7 +1,9 @@
+use crate::tcp::opaque_transport;
 use linkerd_app_core::{
     metrics, profiles,
     proxy::{
         api_resolve::{ConcreteAddr, Metadata},
+        http,
         resolve::map_endpoint::MapEndpoint,
     },
     svc::{self, Param},
@@ -12,11 +14,6 @@ use linkerd_app_core::{
 use std::net::SocketAddr;
 use tracing::debug;
 
-#[derive(Copy, Clone)]
-pub(crate) struct EndpointFromMetadata {
-    pub identity_disabled: bool,
-}
-
 #[derive(Copy, Clone, Debug, Eq, PartialEq, Hash)]
 pub struct Accept<P> {
     pub orig_dst: OrigDstAddr,
@@ -45,6 +42,11 @@ pub struct Endpoint<P> {
     pub protocol: P,
 }
 
+#[derive(Copy, Clone)]
+pub struct EndpointFromMetadata {
+    pub identity_disabled: bool,
+}
+
 // === impl Accept ===
 
 impl<P> Param<transport::labels::Key> for Accept<P> {
@@ -218,6 +220,23 @@ impl<P> Param<tls::ConditionalClientTls> for Endpoint<P> {
     }
 }
 
+impl<P> Param<Option<opaque_transport::PortOverride>> for Endpoint<P> {
+    fn param(&self) -> Option<opaque_transport::PortOverride> {
+        self.metadata
+            .opaque_transport_port()
+            .map(opaque_transport::PortOverride)
+    }
+}
+
+impl<P> Param<Option<http::AuthorityOverride>> for Endpoint<P> {
+    fn param(&self) -> Option<http::AuthorityOverride> {
+        self.metadata
+            .authority_override()
+            .cloned()
+            .map(http::AuthorityOverride)
+    }
+}
+
 impl<P> Param<transport::labels::Key> for Endpoint<P> {
     fn param(&self) -> transport::labels::Key {
         transport::labels::Key::OutboundConnect(self.param())

linkerd/app/outbound/src/tcp/connect.rs

@@ -1,15 +1,23 @@
-use super::opaque_transport::OpaqueTransport;
-use crate::{target::Endpoint, Outbound};
+use super::opaque_transport::{self, OpaqueTransport};
+use crate::Outbound;
 use futures::future;
 use linkerd_app_core::{
-    io, svc, tls,
-    transport::{ConnectTcp, Remote, ServerAddr},
+    io,
+    proxy::http,
+    svc, tls,
+    transport::{self, ConnectTcp, Remote, ServerAddr},
     transport_header::SessionProtocol,
     Error,
 };
 use std::task::{Context, Poll};
 use tracing::debug_span;
 
+#[derive(Clone, Debug)]
+pub struct Connect {
+    pub addr: Remote<ServerAddr>,
+    pub tls: tls::ConditionalClientTls,
+}
+
 /// Prevents outbound connections on the loopback interface, unless the
 /// `allow-loopback` feature is enabled.
 #[derive(Clone, Debug)]
@@ -25,19 +33,24 @@ impl Outbound<()> {
 }
 
 impl<C> Outbound<C> {
-    pub fn push_tcp_endpoint<P>(
+    pub fn push_tcp_endpoint<T>(
         self,
     ) -> Outbound<
         impl svc::Service<
-                Endpoint<P>,
+                T,
                 Response = impl io::AsyncRead + io::AsyncWrite + Send + Unpin,
                 Error = Error,
                 Future = impl Send,
             > + Clone,
     >
     where
-        Endpoint<P>: svc::Param<Option<SessionProtocol>>,
-        C: svc::Service<Endpoint<P>, Error = io::Error> + Clone + Send + 'static,
+        T: svc::Param<Remote<ServerAddr>>
+            + svc::Param<tls::ConditionalClientTls>
+            + svc::Param<Option<opaque_transport::PortOverride>>
+            + svc::Param<Option<http::AuthorityOverride>>
+            + svc::Param<Option<SessionProtocol>>
+            + svc::Param<transport::labels::Key>,
+        C: svc::Service<Connect, Error = io::Error> + Clone + Send + 'static,
         C::Response: tls::HasNegotiatedProtocol,
         C::Response: io::AsyncRead + io::AsyncWrite + Send + Unpin + 'static,
         C::Future: Send + 'static,
@@ -147,3 +160,17 @@ where
         future::Either::Left(self.0.call(ep))
     }
 }
+
+// === impl Connect ===
+
+impl svc::Param<Remote<ServerAddr>> for Connect {
+    fn param(&self) -> Remote<ServerAddr> {
+        self.addr
+    }
+}
+
+impl svc::Param<tls::ConditionalClientTls> for Connect {
+    fn param(&self) -> tls::ConditionalClientTls {
+        self.tls.clone()
+    }
+}

linkerd/app/outbound/src/tcp/mod.rs

@@ -4,6 +4,7 @@ pub mod opaque_transport;
 #[cfg(test)]
 mod tests;
 
+pub use self::connect::Connect;
 use crate::target;
 pub use linkerd_app_core::proxy::tcp::Forward;
 use linkerd_app_core::{