outbound: Configure endpoint construction in logical stack (#949)

olix0r · web-flow · commit 1009645dc0bb · 2021-03-18T14:36:51.000-07:00
ebf67d6 moved endpoint construction logic out of the logical stacks. In doing so, the gateway was incorrectly configured to mark endpoints with disabled identity, preventing mTLS between the gateway and destination service. This change moves endpoint construction into each logical stack so that it need not be configured on each instantiation.
diff --git a/linkerd/app/gateway/src/lib.rs b/linkerd/app/gateway/src/lib.rs
@@ -8,7 +8,11 @@ use self::gateway::NewGateway;
 use linkerd_app_core::{
     config::ProxyConfig,
     detect, discovery_rejected, io, metrics, profiles,
-    proxy::{api_resolve::Metadata, core::Resolve, http, resolve::map_endpoint},
+    proxy::{
+        api_resolve::{ConcreteAddr, Metadata},
+        core::Resolve,
+        http,
+    },
     svc::{self, Param},
     tls,
     transport::OrigDstAddr,
@@ -87,13 +91,10 @@ where
     P: profiles::GetProfile<profiles::LogicalAddr> + Clone + Send + Sync + Unpin + 'static,
     P::Future: Send + 'static,
     P::Error: Send,
-    R: Resolve<outbound::http::Concrete, Endpoint = Metadata, Error = Error>
-        + Resolve<outbound::tcp::Concrete, Endpoint = Metadata, Error = Error>,
     R: Clone + Send + Sync + Unpin + 'static,
-    <R as Resolve<outbound::http::Concrete>>::Resolution: Send,
-    <R as Resolve<outbound::http::Concrete>>::Future: Send + Unpin,
-    <R as Resolve<outbound::tcp::Concrete>>::Resolution: Send,
-    <R as Resolve<outbound::tcp::Concrete>>::Future: Send + Unpin,
+    R: Resolve<ConcreteAddr, Endpoint = Metadata, Error = Error>,
+    R::Resolution: Send,
+    R::Future: Send + Unpin,
 {
     let ProxyConfig {
         buffer_capacity,
@@ -116,10 +117,7 @@ where
     let tcp = outbound
         .clone()
         .push_tcp_endpoint()
-        .push_tcp_logical(map_endpoint::Resolve::new(
-            outbound::target::EndpointFromMetadata::default(),
-            resolve.clone(),
-        ))
+        .push_tcp_logical(resolve.clone())
         .into_stack()
         .push_request_filter(|(p, _): (Option<profiles::Receiver>, _)| match p {
             Some(rx) if rx.borrow().name.is_some() => Ok(outbound::tcp::Logical {
@@ -163,10 +161,7 @@ where
     let http = outbound
         .push_tcp_endpoint()
         .push_http_endpoint()
-        .push_http_logical(map_endpoint::Resolve::new(
-            outbound::target::EndpointFromMetadata::default(),
-            resolve,
-        ))
+        .push_http_logical(resolve)
         .into_stack()
         .push(NewGateway::layer(local_id))
         .push(profiles::discover::layer(profiles, move |t: HttpTarget| {
diff --git a/linkerd/app/outbound/src/http/logical.rs b/linkerd/app/outbound/src/http/logical.rs
@@ -1,8 +1,13 @@
-use super::{CanonicalDstHeader, Concrete, Logical};
-use crate::{resolve, stack_labels, Outbound};
+use super::{CanonicalDstHeader, Concrete, Endpoint, Logical};
+use crate::{resolve, stack_labels, target, Outbound};
 use linkerd_app_core::{
     classify, config, profiles,
-    proxy::{core::Resolve, http},
+    proxy::{
+        api_resolve::{ConcreteAddr, Metadata},
+        core::Resolve,
+        http,
+        resolve::map_endpoint,
+    },
     retry, svc, tls, Error, Never, DST_OVERRIDE_HEADER,
 };
 use tracing::debug_span;
@@ -25,14 +30,13 @@ impl<E> Outbound<E> {
     where
         B: http::HttpBody<Error = Error> + std::fmt::Debug + Default + Send + 'static,
         B::Data: Send + 'static,
-        E: svc::NewService<R::Endpoint, Service = ESvc> + Clone + Send + 'static,
+        E: svc::NewService<Endpoint, Service = ESvc> + Clone + Send + 'static,
         ESvc: svc::Service<http::Request<http::BoxBody>, Response = http::Response<http::BoxBody>>
             + Send
             + 'static,
         ESvc::Error: Into<Error>,
         ESvc::Future: Send,
-        R: Resolve<Concrete, Error = Error> + Clone + Send + 'static,
-        R::Endpoint: From<(tls::NoClientTls, Logical)> + Clone + Send,
+        R: Resolve<ConcreteAddr, Error = Error, Endpoint = Metadata> + Clone + Send + 'static,
         R::Resolution: Send,
         R::Future: Send + Unpin,
     {
@@ -49,9 +53,27 @@ impl<E> Outbound<E> {
         } = config.proxy;
         let watchdog = cache_max_idle_age * 2;
 
+        let identity_disabled = rt.identity.is_none();
+        let no_tls_reason = if identity_disabled {
+            tls::NoClientTls::Disabled
+        } else {
+            tls::NoClientTls::NotProvidedByServiceDiscovery
+        };
+        let resolve = svc::stack(resolve.into_service())
+            .check_service::<ConcreteAddr>()
+            .push_request_filter(|c: Concrete| Ok::<_, Never>(c.resolve))
+            .push(svc::layer::mk(move |inner| {
+                map_endpoint::Resolve::new(
+                    target::EndpointFromMetadata { identity_disabled },
+                    inner,
+                )
+            }))
+            .check_service::<Concrete>()
+            .into_inner();
+
         let stack = endpoint
             .clone()
-            .check_new_service::<R::Endpoint, http::Request<http::BoxBody>>()
+            .check_new_service::<Endpoint, http::Request<http::BoxBody>>()
             .push_on_response(
                 svc::layers()
                     .push(http::BoxRequest::layer())
@@ -64,7 +86,7 @@ impl<E> Outbound<E> {
                     // the balancer need not drive them all directly.
                     .push(svc::layer::mk(svc::SpawnReady::new)),
             )
-            .check_new_service::<R::Endpoint, http::Request<_>>()
+            .check_new_service::<Endpoint, http::Request<_>>()
             // Resolve the service to its endpoints and balance requests over them.
             //
             // If the balancer has been empty/unavailable, eagerly fail requests.
@@ -100,12 +122,7 @@ impl<E> Outbound<E> {
                             .push(http::BoxRequest::layer())
                             .push(http::BoxResponse::layer()),
                     )
-                    .push_map_target(|logical: Logical| {
-                        R::Endpoint::from((
-                            tls::NoClientTls::NotProvidedByServiceDiscovery,
-                            logical,
-                        ))
-                    })
+                    .push_map_target(move |l: Logical| Endpoint::from((no_tls_reason, l)))
                     .into_inner(),
             ))
             // Distribute requests over a distribution of balancers via a
@@ -154,7 +171,7 @@ impl<E> Outbound<E> {
             )
             .instrument(|l: &Logical| debug_span!("logical", dst = %l.addr()))
             .push_switch(
-                |logical: Logical| {
+                move |logical: Logical| {
                     let should_resolve = match logical.profile.as_ref() {
                         Some(p) => {
                             let p = p.borrow();
@@ -166,10 +183,7 @@ impl<E> Outbound<E> {
                     if should_resolve {
                         Ok::<_, Never>(svc::Either::A(logical))
                     } else {
-                        Ok(svc::Either::B(R::Endpoint::from((
-                            tls::NoClientTls::NotProvidedByServiceDiscovery,
-                            logical,
-                        ))))
+                        Ok(svc::Either::B(Endpoint::from((no_tls_reason, logical))))
                     }
                 },
                 svc::stack(endpoint)
diff --git a/linkerd/app/outbound/src/http/tests.rs b/linkerd/app/outbound/src/http/tests.rs
@@ -1,7 +1,6 @@
 use super::Endpoint;
 
 use crate::{
-    target,
     test_util::{
         support::{connect::Connect, http_util, profile, resolver, track},
         *,
@@ -12,7 +11,6 @@ use bytes::Bytes;
 use hyper::{client::conn::Builder as ClientBuilder, Body, Request, Response};
 use linkerd_app_core::{
     io,
-    proxy::resolve::map_endpoint,
     svc::{self, NewService},
     tls,
     transport::{listen, ClientAddr, Local, OrigDstAddr, Remote, ServerAddr},
@@ -76,10 +74,7 @@ where
     out.clone().with_stack(NoTcpBalancer).push_detect_http(
         out.with_stack(connect)
             .push_http_endpoint()
-            .push_http_logical(map_endpoint::Resolve::new(
-                target::EndpointFromMetadata::default(),
-                resolver,
-            ))
+            .push_http_logical(resolver)
             .push_http_server()
             .into_inner(),
     )
diff --git a/linkerd/app/outbound/src/lib.rs b/linkerd/app/outbound/src/lib.rs
@@ -17,7 +17,10 @@ pub(crate) mod test_util;
 use linkerd_app_core::{
     config::ProxyConfig,
     io, metrics, profiles,
-    proxy::{api_resolve::Metadata, core::Resolve, resolve::map_endpoint},
+    proxy::{
+        api_resolve::{ConcreteAddr, Metadata},
+        core::Resolve,
+    },
     serve, svc, tls,
     transport::listen,
     AddrMatch, Error, ProxyRuntime,
@@ -105,40 +108,29 @@ impl<S> Outbound<S> {
         <S as svc::Service<http::Endpoint>>::Response:
             tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin,
         <S as svc::Service<http::Endpoint>>::Future: Send + Unpin,
-        R: Resolve<http::Concrete, Endpoint = Metadata, Error = Error>,
-        <R as Resolve<http::Concrete>>::Resolution: Send,
-        <R as Resolve<http::Concrete>>::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,
-        R: Resolve<tcp::Concrete, Endpoint = Metadata, Error = Error>,
-        <R as Resolve<tcp::Concrete>>::Resolution: Send,
-        <R as Resolve<tcp::Concrete>>::Future: Send + Unpin,
         R: Clone + Send + 'static,
+        R: Resolve<ConcreteAddr, Endpoint = Metadata, Error = Error>,
+        R::Resolution: Send,
+        R::Future: Send + Unpin,
         P: profiles::GetProfile<profiles::LogicalAddr> + Clone + Send + 'static,
         P::Future: Send,
         P::Error: Send,
         I: io::AsyncRead + io::AsyncWrite + io::PeerAddr + std::fmt::Debug + Send + Unpin + 'static,
     {
-        let identity_disabled = self.runtime.identity.is_none();
-
         let http = self
             .clone()
             .push_tcp_endpoint()
             .push_http_endpoint()
-            .push_http_logical(map_endpoint::Resolve::new(
-                target::EndpointFromMetadata { identity_disabled },
-                resolve.clone(),
-            ))
+            .push_http_logical(resolve.clone())
             .push_http_server()
             .into_inner();
 
         self.push_tcp_endpoint()
-            .push_tcp_logical(map_endpoint::Resolve::new(
-                target::EndpointFromMetadata { identity_disabled },
-                resolve,
-            ))
+            .push_tcp_logical(resolve)
             .push_detect_http(http)
             .push_discover(profiles)
             .into_inner()
@@ -148,13 +140,10 @@ impl<S> Outbound<S> {
 impl Outbound<()> {
     pub fn serve<P, R>(self, profiles: P, resolve: R) -> (SocketAddr, impl Future<Output = ()>)
     where
-        R: Resolve<http::Concrete, Endpoint = Metadata, Error = Error>,
-        <R as Resolve<http::Concrete>>::Resolution: Send,
-        <R as Resolve<http::Concrete>>::Future: Send + Unpin,
-        R: Resolve<tcp::Concrete, Endpoint = Metadata, Error = Error>,
-        <R as Resolve<tcp::Concrete>>::Resolution: Send,
-        <R as Resolve<tcp::Concrete>>::Future: Send + Unpin,
         R: Clone + Send + Sync + Unpin + 'static,
+        R: Resolve<ConcreteAddr, Endpoint = Metadata, Error = Error>,
+        R::Resolution: Send,
+        R::Future: Send + Unpin,
         P: profiles::GetProfile<profiles::LogicalAddr> + Clone + Send + Sync + Unpin + 'static,
         P::Future: Send,
         P::Error: Send,
@@ -170,7 +159,6 @@ impl Outbound<()> {
         let serve = async move {
             if self.config.ingress_mode {
                 info!("Outbound routing in ingress-mode");
-                let identity_disabled = self.runtime.identity.is_none();
                 let tcp = self
                     .to_tcp_connect()
                     .push_tcp_endpoint()
@@ -180,10 +168,7 @@ impl Outbound<()> {
                     .to_tcp_connect()
                     .push_tcp_endpoint()
                     .push_http_endpoint()
-                    .push_http_logical(map_endpoint::Resolve::new(
-                        target::EndpointFromMetadata { identity_disabled },
-                        resolve,
-                    ))
+                    .push_http_logical(resolve)
                     .into_inner();
                 let stack = self.to_ingress(profiles, tcp, http);
                 let shutdown = self.runtime.drain.signaled();
diff --git a/linkerd/app/outbound/src/target.rs b/linkerd/app/outbound/src/target.rs
@@ -13,7 +13,7 @@ use std::net::SocketAddr;
 use tracing::debug;
 
 #[derive(Copy, Clone)]
-pub struct EndpointFromMetadata {
+pub(crate) struct EndpointFromMetadata {
     pub identity_disabled: bool,
 }
 
@@ -191,7 +191,7 @@ impl<P> From<(tls::NoClientTls, Logical<P>)> for Endpoint<P> {
             },
             Some((addr, metadata)) => Self {
                 addr: Remote(ServerAddr(addr)),
-                tls: EndpointFromMetadata::client_tls(&metadata),
+                tls: EndpointFromMetadata::client_tls(&metadata, reason),
                 metadata,
                 logical_addr: logical.addr(),
                 protocol: logical.protocol,
@@ -252,16 +252,8 @@ impl<P: std::hash::Hash> std::hash::Hash for Endpoint<P> {
 
 // === EndpointFromMetadata ===
 
-impl Default for EndpointFromMetadata {
-    fn default() -> Self {
-        Self {
-            identity_disabled: false,
-        }
-    }
-}
-
 impl EndpointFromMetadata {
-    fn client_tls(metadata: &Metadata) -> tls::ConditionalClientTls {
+    fn client_tls(metadata: &Metadata, reason: tls::NoClientTls) -> tls::ConditionalClientTls {
         // If we're transporting an opaque protocol OR we're communicating with
         // a gateway, then set an ALPN value indicating support for a transport
         // header.
@@ -283,9 +275,7 @@ impl EndpointFromMetadata {
                     },
                 })
             })
-            .unwrap_or(Conditional::None(
-                tls::NoClientTls::NotProvidedByServiceDiscovery,
-            ))
+            .unwrap_or(Conditional::None(reason))
     }
 }
 
@@ -302,7 +292,7 @@ impl<P: Copy + std::fmt::Debug> MapEndpoint<Concrete<P>, Metadata> for EndpointF
         let tls = if self.identity_disabled {
             tls::ConditionalClientTls::None(tls::NoClientTls::Disabled)
         } else {
-            Self::client_tls(&metadata)
+            Self::client_tls(&metadata, tls::NoClientTls::NotProvidedByServiceDiscovery)
         };
         Endpoint {
             addr: Remote(ServerAddr(addr)),
diff --git a/linkerd/app/outbound/src/tcp/logical.rs b/linkerd/app/outbound/src/tcp/logical.rs
diff --git a/linkerd/app/outbound/src/tcp/tests.rs b/linkerd/app/outbound/src/tcp/tests.rs
