outbound: Make HTTP endpoint stack generic on its target (#952)

The HTTP endpoint stack is coupled to the Endpoint target type.

In order to make the outbound stack more flexible so that the endpoint
stack can be used without the logical stack, this change updates the
outbound HTTP endpoint stack to be generic over its target type.

Author: olix0r

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

@@ -1,19 +1,18 @@
-use super::{require_identity_on_endpoint::NewRequireIdentity, Endpoint};
+use super::require_identity_on_endpoint::NewRequireIdentity;
 use crate::Outbound;
 use linkerd_app_core::{
-    classify, config, http_tracing,
+    classify, config, http_tracing, metrics,
     proxy::{http, tap},
-    reconnect, svc, Error, CANONICAL_DST_HEADER, L5D_REQUIRE_ID,
+    reconnect, svc, tls, Error, CANONICAL_DST_HEADER, L5D_REQUIRE_ID,
 };
 use tokio::io;
-use tracing::debug_span;
 
 impl<C> Outbound<C> {
-    pub fn push_http_endpoint<B>(
+    pub fn push_http_endpoint<T, B>(
         self,
     ) -> Outbound<
         impl svc::NewService<
-                Endpoint,
+                T,
                 Service = impl svc::Service<
                     http::Request<B>,
                     Response = http::Response<http::BoxBody>,
@@ -23,9 +22,15 @@ impl<C> Outbound<C> {
             > + Clone,
     >
     where
+        T: Clone + Send + Sync + 'static,
+        T: svc::Param<http::client::Settings>
+            + svc::Param<Option<http::AuthorityOverride>>
+            + svc::Param<metrics::EndpointLabels>
+            + svc::Param<tls::ConditionalClientTls>
+            + tap::Inspect,
         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: svc::Service<T> + Clone + Send + Sync + Unpin + 'static,
         C::Response: io::AsyncRead + io::AsyncWrite + Send + Unpin,
         C::Error: Into<Error>,
         C::Future: Send + Unpin,
@@ -52,7 +57,6 @@ impl<C> Outbound<C> {
                 let backoff = backoff;
                 move |_| Ok(backoff.stream())
             }))
-            .check_new::<Endpoint>()
             .push(tap::NewTapHttp::layer(rt.tap.clone()))
             .push(rt.metrics.http_endpoint.to_layer::<classify::Response, _>())
             .push_on_response(http_tracing::client(
@@ -65,9 +69,7 @@ impl<C> Outbound<C> {
                 "host",
                 CANONICAL_DST_HEADER,
             ]))
-            .push_on_response(http::BoxResponse::layer())
-            .check_new::<Endpoint>()
-            .instrument(|e: &Endpoint| debug_span!("endpoint", peer.addr = %e.addr));
+            .push_on_response(http::BoxResponse::layer());
 
         Outbound {
             config,

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

@@ -53,6 +53,9 @@ impl<E> Outbound<E> {
         } = config.proxy;
         let watchdog = cache_max_idle_age * 2;
 
+        let endpoint =
+            endpoint.instrument(|e: &Endpoint| debug_span!("endpoint", server.addr = %e.addr));
+
         let identity_disabled = rt.identity.is_none();
         let no_tls_reason = if identity_disabled {
             tls::NoClientTls::Disabled

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

@@ -1,13 +1,10 @@
-use crate::http::Endpoint;
 use futures::{
     future::{self, Either},
     TryFutureExt,
 };
 use linkerd_app_core::{
-    errors::IdentityRequired,
-    proxy::http::identity_from_header,
-    svc::{layer, NewService, Service},
-    tls, Conditional, Error, L5D_REQUIRE_ID,
+    errors::IdentityRequired, proxy::http::identity_from_header, svc, tls, Conditional, Error,
+    L5D_REQUIRE_ID,
 };
 use std::task::{Context, Poll};
 use tracing::debug;
@@ -30,19 +27,20 @@ impl<N> NewRequireIdentity<N> {
         Self { inner }
     }
 
-    pub fn layer() -> impl layer::Layer<N, Service = Self> + Clone + Copy {
-        layer::mk(Self::new)
+    pub fn layer() -> impl svc::layer::Layer<N, Service = Self> + Clone + Copy {
+        svc::layer::mk(Self::new)
     }
 }
 
-impl<N> NewService<Endpoint> for NewRequireIdentity<N>
+impl<T, N> svc::NewService<T> for NewRequireIdentity<N>
 where
-    N: NewService<Endpoint>,
+    T: svc::Param<tls::ConditionalClientTls>,
+    N: svc::NewService<T>,
 {
     type Service = RequireIdentity<N::Service>;
 
-    fn new_service(&mut self, target: Endpoint) -> Self::Service {
-        let tls = target.tls.clone();
+    fn new_service(&mut self, target: T) -> Self::Service {
+        let tls = target.param();
         let inner = self.inner.new_service(target);
         RequireIdentity { tls, inner }
     }
@@ -53,15 +51,16 @@ where
 type ResponseFuture<F, T, E> =
     Either<future::Ready<Result<T, Error>>, future::MapErr<F, fn(E) -> Error>>;
 
-impl<S, A> Service<http::Request<A>> for RequireIdentity<S>
+impl<S, A> svc::Service<http::Request<A>> for RequireIdentity<S>
 where
-    S: Service<http::Request<A>>,
+    S: svc::Service<http::Request<A>>,
     S::Error: Into<Error>,
 {
     type Response = S::Response;
     type Error = Error;
     type Future = ResponseFuture<S::Future, S::Response, S::Error>;
 
+    #[inline]
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         self.inner.poll_ready(cx).map_err(Into::into)
     }