outbound: Decouple the resolver from concrete target types (#938)

The endpoint resolver module currently has a fixed dependency on the
`Concrete` and `Endpoint` target types.

This change extracts endpoint-type construction so that the HTTP logical
stack is agnostic of its endpoint target type. It also updates the
`EndpointFromMetadata` type to be responsible for marking identity as
disabled when appropriate. Finally, we are able to re-enable the
outbound TLS hinting test.

Author: olix0r

linkerd/app/gateway/src/lib.rs

@@ -8,7 +8,7 @@ use self::gateway::NewGateway;
 use linkerd_app_core::{
     config::ProxyConfig,
     detect, discovery_rejected, io, metrics, profiles,
-    proxy::{api_resolve::Metadata, core::Resolve, http},
+    proxy::{api_resolve::Metadata, core::Resolve, http, resolve::map_endpoint},
     svc::{self, Param},
     tls,
     transport::OrigDstAddr,
@@ -116,7 +116,10 @@ where
     let tcp = outbound
         .clone()
         .push_tcp_endpoint()
-        .push_tcp_logical(resolve.clone())
+        .push_tcp_logical(map_endpoint::Resolve::new(
+            outbound::target::EndpointFromMetadata::default(),
+            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 {
@@ -160,7 +163,10 @@ where
     let http = outbound
         .push_tcp_endpoint()
         .push_http_endpoint()
-        .push_http_logical(resolve)
+        .push_http_logical(map_endpoint::Resolve::new(
+            outbound::target::EndpointFromMetadata::default(),
+            resolve,
+        ))
         .into_stack()
         .push(NewGateway::layer(local_id))
         .push(profiles::discover::layer(profiles, move |t: HttpTarget| {

linkerd/app/integration/src/tests/telemetry.rs

@@ -50,7 +50,6 @@ impl Fixture {
         let labels = metrics::labels()
             .label("authority", "tele.test.svc.cluster.local")
             .label("direction", "inbound")
-            .label("tls", "disabled")
             .label("target_addr", orig_dst);
         Fixture {
             client,
@@ -79,7 +78,6 @@ impl Fixture {
         let client = client::new(proxy.outbound, "tele.test.svc.cluster.local");
         let labels = metrics::labels()
             .label("direction", "outbound")
-            .label("tls", "disabled")
             .label("authority", authority)
             .label("target_addr", orig_dst);
         Fixture {
@@ -175,7 +173,6 @@ impl TcpFixture {
         let dst_labels = metrics::labels()
             .label("direction", "outbound")
             .label("peer", "dst")
-            .label("tls", "disabled")
             .label("authority", orig_dst);
 
         TcpFixture {
@@ -525,7 +522,6 @@ mod outbound_dst_labels {
         let client = client::new(proxy.outbound, dest);
         let labels = metrics::labels()
             .label("direction", "outbound")
-            .label("tls", "disabled")
             .label("authority", dest_and_port)
             .label("target_addr", addr);
         let f = Fixture {

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

@@ -37,7 +37,6 @@ where
             runtime: rt,
             stack: connect,
         } = self;
-        let identity_disabled = rt.identity.is_none();
         let config::ConnectConfig {
             h1_settings,
             h2_settings,
@@ -70,14 +69,7 @@ where
             ]))
             .push_on_response(http::BoxResponse::layer())
             .check_new::<Endpoint>()
-            .instrument(|e: &Endpoint| debug_span!("endpoint", peer.addr = %e.addr))
-            .push_map_target(move |e: Endpoint| {
-                if identity_disabled {
-                    e.identity_disabled()
-                } else {
-                    e
-                }
-            });
+            .instrument(|e: &Endpoint| debug_span!("endpoint", peer.addr = %e.addr));
 
         Outbound {
             config,

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

@@ -1,9 +1,9 @@
-use super::{Concrete, Endpoint, Logical};
+use super::{Concrete, Logical};
 use crate::{resolve, stack_labels, Outbound};
 use linkerd_app_core::{
     classify, config, profiles,
-    proxy::{api_resolve::Metadata, core::Resolve, http},
-    retry, svc, tls, Error, CANONICAL_DST_HEADER, DST_OVERRIDE_HEADER,
+    proxy::{core::Resolve, http},
+    retry, svc, tls, Error, Never, CANONICAL_DST_HEADER, DST_OVERRIDE_HEADER,
 };
 use tracing::debug_span;
 
@@ -25,13 +25,14 @@ impl<E> Outbound<E> {
     where
         B: http::HttpBody<Error = Error> + std::fmt::Debug + Default + Send + 'static,
         B::Data: Send + 'static,
-        E: svc::NewService<Endpoint, Service = ESvc> + Clone + Send + 'static,
+        E: svc::NewService<R::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, Endpoint = Metadata, Error = Error> + Clone + Send + 'static,
+        R: Resolve<Concrete, Error = Error> + Clone + Send + 'static,
+        R::Endpoint: From<(tls::NoClientTls, Logical)> + Clone + Send,
         R::Resolution: Send,
         R::Future: Send + Unpin,
     {
@@ -50,6 +51,7 @@ impl<E> Outbound<E> {
 
         let stack = endpoint
             .clone()
+            .check_new_service::<R::Endpoint, http::Request<http::BoxBody>>()
             .push_on_response(
                 svc::layers()
                     .push(http::BoxRequest::layer())
@@ -62,6 +64,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<_>>()
             // Resolve the service to its endpoints and balance requests over them.
             //
             // If the balancer has been empty/unavailable, eagerly fail requests.
@@ -79,6 +82,7 @@ impl<E> Outbound<E> {
                     .push(svc::FailFast::layer("HTTP Balancer", dispatch_timeout))
                     .push(http::BoxResponse::layer()),
             )
+            .check_make_service::<Concrete, http::Request<_>>()
             .push(svc::MapErrLayer::new(Into::into))
             // Drives the initial resolution via the service's readiness.
             .into_new_service()
@@ -96,9 +100,12 @@ impl<E> Outbound<E> {
                             .push(http::BoxRequest::layer())
                             .push(http::BoxResponse::layer()),
                     )
-                    .push_map_target(Endpoint::from_logical(
-                        tls::NoClientTls::NotProvidedByServiceDiscovery,
-                    ))
+                    .push_map_target(|logical: Logical| {
+                        R::Endpoint::from((
+                            tls::NoClientTls::NotProvidedByServiceDiscovery,
+                            logical,
+                        ))
+                    })
                     .into_inner(),
             ))
             // Distribute requests over a distribution of balancers via a
@@ -147,7 +154,24 @@ impl<E> Outbound<E> {
             )
             .instrument(|l: &Logical| debug_span!("logical", dst = %l.addr()))
             .push_switch(
-                Logical::or_endpoint(tls::NoClientTls::NotProvidedByServiceDiscovery),
+                |logical: Logical| {
+                    let should_resolve = match logical.profile.as_ref() {
+                        Some(p) => {
+                            let p = p.borrow();
+                            p.endpoint.is_none() && (p.name.is_some() || !p.targets.is_empty())
+                        }
+                        None => false,
+                    };
+
+                    if should_resolve {
+                        Ok::<_, Never>(svc::Either::A(logical))
+                    } else {
+                        Ok(svc::Either::B(R::Endpoint::from((
+                            tls::NoClientTls::NotProvidedByServiceDiscovery,
+                            logical,
+                        ))))
+                    }
+                },
                 svc::stack(endpoint)
                     .push_on_response(http::BoxRequest::layer())
                     .into_inner(),

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

@@ -1,6 +1,7 @@
 use super::Endpoint;
 
 use crate::{
+    target,
     test_util::{
         support::{connect::Connect, http_util, profile, resolver, track},
         *,
@@ -11,6 +12,7 @@ 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},
@@ -74,7 +76,10 @@ where
     out.clone().with_stack(NoTcpBalancer).push_detect_http(
         out.with_stack(connect)
             .push_http_endpoint()
-            .push_http_logical(resolver)
+            .push_http_logical(map_endpoint::Resolve::new(
+                target::EndpointFromMetadata::default(),
+                resolver,
+            ))
             .push_http_server()
             .into_inner(),
     )

linkerd/app/outbound/src/ingress.rs

@@ -63,7 +63,9 @@ impl Outbound<()> {
 
         let tcp = svc::stack(tcp)
             .push_on_response(drain::Retain::layer(self.runtime.drain.clone()))
-            .push_map_target(tcp::Endpoint::from_accept(tls::NoClientTls::IngressNonHttp))
+            .push_map_target(|a: tcp::Accept| {
+                tcp::Endpoint::from((tls::NoClientTls::IngressNonHttp, a))
+            })
             .into_inner();
 
         svc::stack(http)

linkerd/app/outbound/src/lib.rs

@@ -17,7 +17,7 @@ pub(crate) mod test_util;
 use linkerd_app_core::{
     config::ProxyConfig,
     io, metrics, profiles,
-    proxy::{api_resolve::Metadata, core::Resolve},
+    proxy::{api_resolve::Metadata, core::Resolve, resolve::map_endpoint},
     serve, svc, tls,
     transport::listen,
     AddrMatch, Error, ProxyRuntime,
@@ -121,16 +121,24 @@ impl<S> Outbound<S> {
         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(resolve.clone())
+            .push_http_logical(map_endpoint::Resolve::new(
+                target::EndpointFromMetadata { identity_disabled },
+                resolve.clone(),
+            ))
             .push_http_server()
             .into_inner();
 
         self.push_tcp_endpoint()
-            .push_tcp_logical(resolve)
+            .push_tcp_logical(map_endpoint::Resolve::new(
+                target::EndpointFromMetadata { identity_disabled },
+                resolve,
+            ))
             .push_detect_http(http)
             .push_discover(profiles)
             .into_inner()
@@ -162,6 +170,7 @@ 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()
@@ -171,7 +180,10 @@ impl Outbound<()> {
                     .to_tcp_connect()
                     .push_tcp_endpoint()
                     .push_http_endpoint()
-                    .push_http_logical(resolve)
+                    .push_http_logical(map_endpoint::Resolve::new(
+                        target::EndpointFromMetadata { identity_disabled },
+                        resolve,
+                    ))
                     .into_inner();
                 let stack = self.to_ingress(profiles, tcp, http);
                 let shutdown = self.runtime.drain.signaled();

linkerd/app/outbound/src/resolve.rs

@@ -1,51 +1,30 @@
-#![allow(warnings)]
-
-use crate::target::{Concrete, Endpoint, EndpointFromMetadata};
-use futures::{future, prelude::*, stream};
 use linkerd_app_core::{
-    discovery_rejected, is_discovery_rejected,
     proxy::{
-        api_resolve::Metadata,
-        core::{Resolve, ResolveService, Update},
-        discover::{self, Buffer, FromResolve, MakeEndpoint},
-        resolve::map_endpoint,
-    },
-    svc::{
-        layer,
-        stack::{Filter, Param, Predicate},
-        NewService,
+        core::Resolve,
+        discover::{self, Buffer},
     },
-    Addr, Error,
+    svc::{layer, NewService},
 };
-use std::{
-    future::Future,
-    pin::Pin,
-    task::{Context, Poll},
-    time::Duration,
-};
-
-type Stack<P, R, N> =
-    Buffer<discover::Stack<N, map_endpoint::Resolve<EndpointFromMetadata, R>, Endpoint<P>>>;
+use std::time::Duration;
 
-pub fn layer<P, R, N>(
+pub fn layer<T, R, N>(
     resolve: R,
     watchdog: Duration,
-) -> impl layer::Layer<N, Service = Stack<P, R, N>> + Clone
+) -> impl layer::Layer<N, Service = Buffer<discover::Stack<N, R, R::Endpoint>>> + Clone
 where
-    P: Copy + Send + std::fmt::Debug,
-    R: Resolve<Concrete<P>, Endpoint = Metadata> + Clone,
+    T: Clone + Send + std::fmt::Debug,
+    R: Resolve<T> + Clone,
     R::Resolution: Send,
     R::Future: Send,
-    N: NewService<Endpoint<P>>,
+    N: NewService<R::Endpoint>,
 {
     const ENDPOINT_BUFFER_CAPACITY: usize = 1_000;
 
-    let to_endpoint = EndpointFromMetadata;
     layer::mk(move |new_endpoint| {
-        let endpoints = discover::resolve(
-            new_endpoint,
-            map_endpoint::Resolve::new(to_endpoint, resolve.clone()),
-        );
-        Buffer::new(ENDPOINT_BUFFER_CAPACITY, watchdog, endpoints)
+        Buffer::new(
+            ENDPOINT_BUFFER_CAPACITY,
+            watchdog,
+            discover::resolve(new_endpoint, resolve.clone()),
+        )
     })
 }