outbound: TCP discovery and load balancing (#652)

The proxy only forward TCP connections to the original destination address, never
using the mesh's mTLS.

This change modifies the outbound TCP forwarding stack to do discovery based on
the original destination address, resolving service IPs to the individual endpoint IPs
(and their associated identities). This enables mTLS for meshed TCP connections.
When the endpoint cannot be discovered (i.e. due to an InvalidArgument response
from the controller), the connection is forwarded as before.

The PeakEWMA balancer is repurposed for this, using connection latency (and
pending connections) as the load metric. In the future, this should be modified
to count active connections towards the load (but this won't Just Work with the
PeakEwma balancer, so this is deferred for now).

This has been tested [manually](https://github.com/olix0r/init-net-test/blob/b3860861c54344667a797ece5b685a0574798fb3/k8s.yml).
In follow-up changes, we'll extend the transparency tests to validate this behavior.

Author: olix0r

Cargo.lock

@@ -1373,7 +1373,9 @@ dependencies = [
  "futures 0.3.5",
  "linkerd2-duplex",
  "linkerd2-error",
+ "linkerd2-stack",
  "pin-project",
+ "rand 0.7.2",
  "tokio",
  "tower",
 ]

linkerd/app/core/src/svc.rs

@@ -159,6 +159,14 @@ impl<S> Stack<S> {
         self.push(stack::map_target::MapTargetLayer::new(map_target))
     }
 
+    /// Wraps a `Service<T>` as a `Service<()>`.
+    ///
+    /// Each time the service is called, the `T`-typed request is cloned and
+    /// issued into the inner service.
+    pub fn push_make_thunk(self) -> Stack<stack::make_thunk::MakeThunk<S>> {
+        self.push(layer::mk(stack::make_thunk::MakeThunk::new))
+    }
+
     pub fn instrument<G: Clone>(self, get_span: G) -> Stack<InstrumentMake<G, S>> {
         self.push(InstrumentMakeLayer::new(get_span))
     }

linkerd/app/inbound/src/lib.rs

@@ -371,7 +371,9 @@ impl Config {
         // tasks from their constructor. This helps to ensure that tasks are
         // spawned on the same runtime as the proxy.
         // Forwards TCP streams that cannot be decoded as HTTP.
-        let tcp_forward = svc::stack(tcp::Forward::new(tcp_connect))
+        let tcp_forward = svc::stack(tcp_connect)
+            .push_make_thunk()
+            .push(svc::layer::mk(tcp::Forward::new))
             .push(admit::AdmitLayer::new(prevent_loop.into()));
 
         let http = DetectHttp::new(

linkerd/app/outbound/src/endpoint.rs

@@ -58,6 +58,7 @@ pub struct HttpEndpoint {
 pub struct TcpEndpoint {
     pub addr: SocketAddr,
     pub identity: tls::PeerIdentity,
+    //pub labels: Option<String>,
 }
 
 // === impl HttpConrete ===
@@ -302,15 +303,22 @@ impl Into<EndpointLabels> for HttpEndpoint {
 
 // === impl TcpEndpoint ===
 
-impl From<listen::Addrs> for TcpEndpoint {
-    fn from(addrs: listen::Addrs) -> Self {
+impl From<SocketAddr> for TcpEndpoint {
+    fn from(addr: SocketAddr) -> Self {
         Self {
-            addr: addrs.target_addr(),
+            addr,
             identity: Conditional::None(tls::ReasonForNoPeerName::NotHttp.into()),
+            //labels: None,
         }
     }
 }
 
+impl From<listen::Addrs> for TcpEndpoint {
+    fn from(addrs: listen::Addrs) -> Self {
+        addrs.target_addr().into()
+    }
+}
+
 impl Into<SocketAddr> for TcpEndpoint {
     fn into(self) -> SocketAddr {
         self.addr
@@ -323,14 +331,35 @@ impl tls::HasPeerIdentity for TcpEndpoint {
     }
 }
 
-impl Into<EndpointLabels> for TcpEndpoint {
-    fn into(self) -> EndpointLabels {
-        use linkerd2_app_core::metric_labels::{Direction, TlsId};
-        EndpointLabels {
-            direction: Direction::Out,
-            tls_id: self.identity.as_ref().map(|id| TlsId::ServerId(id.clone())),
-            authority: None,
-            labels: None,
+// impl Into<EndpointLabels> for TcpEndpoint {
+//     fn into(self) -> EndpointLabels {
+//         use linkerd2_app_core::metric_labels::{Direction, TlsId};
+//         EndpointLabels {
+//             authority: None,
+//             direction: Direction::Out,
+//             labels: self.labels,
+//             tls_id: self.identity.as_ref().map(|id| TlsId::ServerId(id.clone())),
+//         }
+//     }
+// }
+
+impl MapEndpoint<Addr, Metadata> for FromMetadata {
+    type Out = TcpEndpoint;
+
+    fn map_endpoint(&self, dst: &Addr, addr: SocketAddr, metadata: Metadata) -> Self::Out {
+        tracing::debug!(%dst, %addr, ?metadata, "Resolved endpoint");
+        let identity = metadata
+            .identity()
+            .cloned()
+            .map(Conditional::Some)
+            .unwrap_or_else(|| {
+                Conditional::None(tls::ReasonForNoPeerName::NotProvidedByServiceDiscovery.into())
+            });
+
+        TcpEndpoint {
+            addr,
+            identity,
+            //labels: prefix_labels("dst", metadata.labels().into_iter()),
         }
     }
 }

linkerd/app/outbound/src/lib.rs

@@ -22,10 +22,14 @@ use linkerd2_app_core::{
     spans::SpanConverter,
     svc::{self, NewService},
     transport::{self, listen, tls},
-    Conditional, DiscoveryRejected, Error, ProxyMetrics, StackMetrics, TraceContextLayer,
+    Addr, Conditional, DiscoveryRejected, Error, ProxyMetrics, StackMetrics, TraceContextLayer,
     CANONICAL_DST_HEADER, DST_OVERRIDE_HEADER, L5D_REQUIRE_ID,
 };
-use std::{collections::HashMap, net::IpAddr, time::Duration};
+use std::{
+    collections::HashMap,
+    net::{IpAddr, SocketAddr},
+    time::Duration,
+};
 use tokio::sync::mpsc;
 use tracing::{info, info_span};
 
@@ -363,18 +367,22 @@ impl Config {
             .into_inner()
     }
 
-    pub async fn build_server<R, C, H, S>(
+    pub async fn build_server<E, R, C, H, S>(
         self,
         listen_addr: std::net::SocketAddr,
         listen: impl Stream<Item = std::io::Result<listen::Connection>> + Send + 'static,
         refine: R,
+        resolve: E,
         tcp_connect: C,
         http_router: H,
         metrics: ProxyMetrics,
         span_sink: Option<mpsc::Sender<oc::Span>>,
         drain: drain::Watch,
     ) -> Result<(), Error>
     where
+        E: Resolve<Addr, Endpoint = proxy::api_resolve::Metadata> + Unpin + Clone + Send + 'static,
+        E::Future: Unpin + Send,
+        E::Resolution: Unpin + Send,
         R: tower::Service<dns::Name, Error = Error, Response = dns::Name>
             + Unpin
             + Clone
@@ -404,6 +412,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;
@@ -448,22 +458,57 @@ impl Config {
             .into_inner()
             .into_make_service();
 
-        // The stack is served lazily since caching layers spawn tasks from
-        // their constructor. This helps to ensure that tasks are spawned on the
-        // same runtime as the proxy.
-        // Forwards TCP streams that cannot be decoded as HTTP.
-        let tcp_forward = svc::stack(tcp::Forward::new(tcp_connect))
+        // Load balances TCP streams that cannot be decoded as HTTP.
+        let tcp_balance = svc::stack(tcp_connect.clone())
+            .push_make_thunk()
+            .instrument(|t: &TcpEndpoint| info_span!("tcp.endpoint", peer.addr = %t.addr, peer.id = ?t.identity))
             .push(admit::AdmitLayer::new(prevent_loop))
-            .push_map_target(TcpEndpoint::from);
+            .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))
+            .instrument(|a: &SocketAddr| info_span!("tcp.balance", dst = %a))
+            .push_fallback_with_predicate(
+                svc::stack(tcp_connect.clone())
+                    .push_make_thunk()
+                    .push(admit::AdmitLayer::new(prevent_loop))
+                    .push_map_target(TcpEndpoint::from)
+                    .instrument(|a: &SocketAddr| info_span!("tcp.forward", peer.addr = %a)),
+                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))
+            .push_map_target(|a: listen::Addrs| a.target_addr());
 
         let http = http::DetectHttp::new(
             h2_settings,
             detect_protocol_timeout,
             http_server,
-            tcp_forward.clone(),
+            tcp_balance,
             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(SkipDetect::new(skip_detect, http, tcp_forward))
             .push(metrics.transport.layer_accept(TransportLabels));
 

linkerd/app/src/lib.rs

@@ -143,7 +143,7 @@ impl Config {
 
             let outbound_http = outbound.build_http_router(
                 outbound_http_endpoint,
-                dst.resolve,
+                dst.resolve.clone(),
                 dst.profiles.clone(),
                 outbound_metrics.clone(),
             );
@@ -156,6 +156,7 @@ impl Config {
                         svc::stack(refine.clone())
                             .push_map_response(|(n, _)| n)
                             .into_inner(),
+                        dst.resolve,
                         outbound_connect,
                         outbound_http.clone(),
                         outbound_metrics,

linkerd/proxy/http/src/version.rs

@@ -1,4 +1,5 @@
 use linkerd2_proxy_transport::io::{self, Peekable, PrefixedIo};
+use tracing::{debug, trace};
 
 #[derive(Copy, Clone, Debug, PartialEq, Eq)]
 pub enum Version {
@@ -16,6 +17,7 @@ impl Version {
         // http2 is easiest to detect
         if bytes.len() >= Self::H2_PREFACE.len() {
             if &bytes[..Self::H2_PREFACE.len()] == Self::H2_PREFACE {
+                trace!("Detected H2");
                 return Some(Version::H2);
             }
         }
@@ -34,11 +36,14 @@ impl Version {
             // We didn't want to keep parsing headers, just validate that
             // the first line is HTTP1.
             Ok(_) | Err(httparse::Error::TooManyHeaders) => {
+                trace!("Detected H1");
                 return Some(Version::Http1);
             }
             _ => {}
         }
 
+        debug!("Not HTTP");
+        trace!(?bytes);
         None
     }
 

linkerd/proxy/tcp/Cargo.toml

@@ -10,6 +10,8 @@ publish = false
 futures = { version = "0.3", features = ["compat"] }
 linkerd2-duplex = { path = "../../duplex" }
 linkerd2-error = { path = "../../error" }
+linkerd2-stack = { path = "../../stack" }
+rand = { version = "0.7", features = ["small_rng"] }
 tokio = { version = "0.2" }
 tower = { version = "0.3", default-features = false }
-pin-project = "0.4"
+pin-project = "0.4"

linkerd/proxy/tcp/src/balance.rs

@@ -0,0 +1,29 @@
+use linkerd2_error::Error;
+use linkerd2_stack::layer;
+use rand::{rngs::SmallRng, SeedableRng};
+use std::{hash::Hash, time::Duration};
+pub use tower::{
+    balance::p2c::Balance,
+    load::{Load, PeakEwmaDiscover},
+};
+use tower::{discover::Discover, load::CompleteOnResponse};
+
+/// Produces a PeakEWMA balancer that uses connect latency (and pending
+/// connections) as its load metric.
+pub fn layer<T, D>(
+    default_rtt: Duration,
+    decay: Duration,
+) -> impl tower::layer::Layer<D, Service = Balance<PeakEwmaDiscover<D, CompleteOnResponse>, T>> + Clone
+where
+    D: Discover,
+    D::Key: Hash,
+    D::Service: tower::Service<T>,
+    <D::Service as tower::Service<T>>::Error: Into<Error>,
+{
+    let rng = SmallRng::from_entropy();
+    layer::mk(move |discover| {
+        let loaded =
+            PeakEwmaDiscover::new(discover, default_rtt, decay, CompleteOnResponse::default());
+        Balance::from_rng(loaded, rng.clone()).expect("RNG must be valid")
+    })
+}

linkerd/proxy/tcp/src/forward.rs

@@ -1,49 +1,53 @@
 use futures::{future, prelude::*};
 use linkerd2_duplex::Duplex;
 use linkerd2_error::Error;
-use std::future::Future;
-use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::{
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
 use tokio::io::{AsyncRead, AsyncWrite};
 use tower::Service;
 
 #[derive(Clone, Debug)]
-pub struct Forward<C> {
-    connect: C,
+pub struct Forward<M> {
+    make_connect: M,
 }
 
 #[derive(Clone, Debug)]
-pub struct Accept<C, T> {
+pub struct Accept<C> {
     connect: C,
-    target: T,
 }
 
-impl<C> Forward<C> {
-    pub fn new(connect: C) -> Self {
-        Self { connect }
+impl<M> Forward<M> {
+    pub fn new(make_connect: M) -> Self {
+        Self { make_connect }
     }
 }
 
-impl<C: Clone, T> Service<T> for Forward<C> {
-    type Response = Accept<C, T>;
-    type Error = Error;
-    type Future = future::Ready<Result<Self::Response, Error>>;
+impl<M, T> Service<T> for Forward<M>
+where
+    M: Service<T>,
+{
+    type Response = Accept<M::Response>;
+    type Error = M::Error;
+    type Future = future::MapOk<M::Future, fn(M::Response) -> Accept<M::Response>>;
 
-    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), self::Error>> {
-        Poll::Ready(Ok(()))
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), M::Error>> {
+        self.make_connect.poll_ready(cx)
     }
 
     fn call(&mut self, target: T) -> Self::Future {
-        let connect = self.connect.clone();
-        future::ok(Accept { connect, target })
+        self.make_connect
+            .call(target)
+            .map_ok(|connect| Accept { connect })
     }
 }
 
-impl<C, T, I> Service<I> for Accept<C, T>
+impl<C, I> Service<I> for Accept<C>
 where
-    T: Clone,
     I: AsyncRead + AsyncWrite + Send + Unpin + 'static,
-    C: tower::Service<T> + Send + 'static,
+    C: tower::Service<()> + Send + 'static,
     C::Error: Into<Error>,
     C::Future: Send + 'static,
     C::Response: AsyncRead + AsyncWrite + Send + Unpin + 'static,
@@ -58,7 +62,7 @@ where
     }
 
     fn call(&mut self, src_io: I) -> Self::Future {
-        let connect = self.connect.call(self.target.clone()).err_into::<Error>();
+        let connect = self.connect.call(()).err_into::<Error>();
         Box::pin(async move {
             let dst_io = connect.await?;
             Duplex::new(src_io, dst_io).err_into::<Error>().await