Make HTTP client stacks synchronous (#674)

olix0r · web-flow · commit 048afd1e0da1 · 2020-09-24T11:32:03.000-07:00
This change implements `NewService` for `Reconnect` and other
HTTP-client middlewares. HTTP clients are almost always coerced to
`NewService`, so this reduces boilerplate and type complexity.
diff --git a/Cargo.lock b/Cargo.lock
@@ -1444,6 +1444,7 @@ version = "0.1.0"
 dependencies = [
  "futures 0.3.5",
  "linkerd2-error",
+ "linkerd2-stack",
  "pin-project",
  "tower",
  "tracing",
diff --git a/linkerd/app/core/src/control.rs b/linkerd/app/core/src/control.rs
@@ -62,14 +62,11 @@ impl Config {
             .push_timeout(self.connect.timeout)
             .push(self::client::layer())
             .push(reconnect::layer(backoff.clone()))
-            .push_spawn_ready()
-            .into_new_service()
             .push(self::resolve::layer(dns, backoff))
             .push_on_response(self::control::balance::layer())
-            .push(metrics.into_layer::<classify::Response>())
-            .push(self::add_origin::Layer::new())
             .into_new_service()
-            .check_new_service()
+            .push(metrics.into_layer::<classify::Response>())
+            .push(self::add_origin::layer())
             .push_on_response(svc::layers().push_spawn_buffer(self.buffer_capacity))
             .new_service(self.addr)
     }
@@ -78,126 +75,54 @@ impl Config {
 /// Sets the request's URI from `Config`.
 mod add_origin {
     use super::ControlAddr;
-    use futures::{ready, TryFuture};
-    use linkerd2_error::Error;
-    use pin_project::pin_project;
-    use std::future::Future;
+    use linkerd2_stack::{layer, NewService, ResultService};
     use std::marker::PhantomData;
-    use std::pin::Pin;
-    use std::task::{Context, Poll};
     use tower_request_modifier::{Builder, RequestModifier};
 
-    #[derive(Debug)]
-    pub struct Layer<B> {
-        _marker: PhantomData<fn(B)>,
+    pub fn layer<M, B>() -> impl layer::Layer<M, Service = NewAddOrigin<M, B>> + Clone {
+        layer::mk(|inner| NewAddOrigin {
+            inner,
+            _marker: PhantomData,
+        })
     }
 
     #[derive(Debug)]
-    pub struct MakeAddOrigin<M, B> {
+    pub struct NewAddOrigin<M, B> {
         inner: M,
         _marker: PhantomData<fn(B)>,
     }
 
-    #[pin_project]
-    pub struct MakeFuture<F, B> {
-        #[pin]
-        inner: F,
-        authority: http::uri::Authority,
-        _marker: PhantomData<fn(B)>,
-    }
+    // === impl NewAddOrigin ===
 
-    // === impl Layer ===
-
-    impl<B> Layer<B> {
-        pub fn new() -> Self {
-            Layer {
-                _marker: PhantomData,
-            }
-        }
-    }
-
-    impl<B> Clone for Layer<B> {
+    impl<M: Clone, B> Clone for NewAddOrigin<M, B> {
         fn clone(&self) -> Self {
             Self {
+                inner: self.inner.clone(),
                 _marker: self._marker,
             }
         }
     }
 
-    impl<M, B> tower::layer::Layer<M> for Layer<B> {
-        type Service = MakeAddOrigin<M, B>;
-
-        fn layer(&self, inner: M) -> Self::Service {
-            Self::Service {
-                inner,
-                _marker: PhantomData,
-            }
-        }
-    }
-
-    // === impl MakeAddOrigin ===
-
-    impl<M, B> tower::Service<ControlAddr> for MakeAddOrigin<M, B>
+    impl<M, B> NewService<ControlAddr> for NewAddOrigin<M, B>
     where
-        M: tower::Service<ControlAddr>,
-        M::Error: Into<Error>,
+        M: NewService<ControlAddr>,
     {
-        type Response = RequestModifier<M::Response, B>;
-        type Error = Error;
-        type Future = MakeFuture<M::Future, B>;
+        type Service = ResultService<RequestModifier<M::Service, B>, BuildError>;
 
-        fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-            self.inner.poll_ready(cx).map_err(Into::into)
-        }
-
-        fn call(&mut self, target: ControlAddr) -> Self::Future {
+        fn new_service(&mut self, target: ControlAddr) -> Self::Service {
             let authority = target.addr.to_http_authority();
-            let inner = self.inner.call(target);
-            MakeFuture {
-                inner,
-                authority,
-                _marker: PhantomData,
-            }
-        }
-    }
-
-    impl<M, B> Clone for MakeAddOrigin<M, B>
-    where
-        M: tower::Service<ControlAddr> + Clone,
-    {
-        fn clone(&self) -> Self {
-            Self {
-                inner: self.inner.clone(),
-                _marker: PhantomData,
-            }
-        }
-    }
-
-    // === impl MakeFuture ===
-
-    impl<F, B> Future for MakeFuture<F, B>
-    where
-        F: TryFuture,
-        F::Error: Into<Error>,
-    {
-        type Output = Result<RequestModifier<F::Ok, B>, Error>;
-
-        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-            let this = self.project();
-            let inner = ready!(this.inner.try_poll(cx).map_err(Into::into))?;
-
-            Poll::Ready(
+            ResultService::from(
                 Builder::new()
-                    .set_origin(format!("http://{}", this.authority))
-                    .build(inner)
-                    .map_err(|_| BuildError.into()),
+                    .set_origin(format!("http://{}", authority))
+                    .build(self.inner.new_service(target))
+                    .map_err(|_| BuildError(())),
             )
         }
     }
 
     // XXX the request_modifier build error does not implement Error...
     #[derive(Debug)]
-    struct BuildError;
+    pub struct BuildError(());
 
     impl std::error::Error for BuildError {}
     impl std::fmt::Display for BuildError {
diff --git a/linkerd/app/inbound/src/lib.rs b/linkerd/app/inbound/src/lib.rs
@@ -203,7 +203,7 @@ impl Config {
                 let backoff = connect.backoff.clone();
                 move |_| Ok(backoff.stream())
             }))
-            .check_make_service::<HttpEndpoint, http::Request<_>>();
+            .check_new_service::<HttpEndpoint, http::Request<_>>();
 
         let observe = svc::layers()
             // Registers the stack to be tapped.
@@ -220,7 +220,6 @@ impl Config {
             .push_map_target(HttpEndpoint::from)
             .push(observe)
             .push_on_response(svc::layers().box_http_response())
-            .into_new_service()
             .check_new_service::<Target, http::Request<_>>();
 
         // Attempts to discover a service profile for each logical target (as
diff --git a/linkerd/app/outbound/src/lib.rs b/linkerd/app/outbound/src/lib.rs
@@ -20,8 +20,8 @@ use linkerd2_app_core::{
     spans::SpanConverter,
     svc::{self},
     transport::{self, listen, tls},
-    Addr, Conditional, DiscoveryRejected, Error, Never, ProxyMetrics, StackMetrics,
-    TraceContextLayer, CANONICAL_DST_HEADER, DST_OVERRIDE_HEADER, L5D_REQUIRE_ID,
+    Addr, Conditional, DiscoveryRejected, Error, ProxyMetrics, StackMetrics, TraceContextLayer,
+    CANONICAL_DST_HEADER, DST_OVERRIDE_HEADER, L5D_REQUIRE_ID,
 };
 use std::{
     collections::HashMap,
@@ -173,18 +173,15 @@ impl Config {
         tap_layer: tap::Layer,
         metrics: ProxyMetrics,
         span_sink: Option<mpsc::Sender<oc::Span>>,
-    ) -> impl tower::Service<
+    ) -> impl svc::NewService<
         HttpEndpoint,
-        Error = Never,
-        Future = impl Unpin + Send,
-        Response = impl tower::Service<
+        Service = impl tower::Service<
             http::Request<B>,
             Response = http::Response<http::boxed::Payload>,
             Error = Error,
             Future = impl Send,
         > + Send,
-    > + Unpin
-           + Clone
+    > + Clone
            + Send
     where
         B: http::HttpBody<Error = Error> + std::fmt::Debug + Default + Send + 'static,
@@ -226,14 +223,15 @@ impl Config {
                     }
                 }
             }))
+            .check_new::<HttpEndpoint>()
             .push(observability.clone())
             .push(identity_headers.clone())
             .push(http::override_authority::Layer::new(vec![
                 ::http::header::HOST.as_str(),
                 CANONICAL_DST_HEADER,
             ]))
             .push_on_response(svc::layers().box_http_response())
-            .check_service::<HttpEndpoint>()
+            .check_new::<HttpEndpoint>()
             .instrument(|e: &HttpEndpoint| info_span!("endpoint", peer.addr = %e.addr))
             .into_inner()
     }
@@ -260,9 +258,7 @@ impl Config {
     where
         B: http::HttpBody<Error = Error> + std::fmt::Debug + Default + Send + 'static,
         B::Data: Send + 'static,
-        E: tower::Service<HttpEndpoint, Response = S> + Unpin + Clone + Send + Sync + 'static,
-        E::Error: Into<Error>,
-        E::Future: Unpin + Send,
+        E: svc::NewService<HttpEndpoint, Service = S> + Clone + Send + Sync + 'static,
         S: tower::Service<
                 http::Request<http::boxed::Payload>,
                 Response = http::Response<http::boxed::Payload>,
@@ -302,14 +298,12 @@ impl Config {
 
         // Builds a balancer for each concrete destination.
         let concrete = svc::stack(endpoint.clone())
-            .check_make_service::<HttpEndpoint, http::Request<http::boxed::Payload>>()
+            .check_new_service::<HttpEndpoint, http::Request<http::boxed::Payload>>()
             .push_on_response(
                 svc::layers()
                     .push(metrics.stack.layer(stack_labels("balance.endpoint")))
                     .box_http_request(),
             )
-            .push_spawn_ready()
-            .into_new_service()
             .check_new_service::<HttpEndpoint, http::Request<_>>()
             .push(discover)
             .check_service::<HttpConcrete>()
@@ -365,7 +359,6 @@ impl Config {
 
         // Caches clients that bypass discovery/balancing.
         let forward = svc::stack(endpoint)
-            .into_new_service()
             .instrument(|t: &HttpEndpoint| debug_span!("forward", peer.id = ?t.identity))
             .check_new_service::<HttpEndpoint, http::Request<_>>();
 
diff --git a/linkerd/proxy/http/src/override_authority.rs b/linkerd/proxy/http/src/override_authority.rs
@@ -1,10 +1,7 @@
 use super::h1;
-use futures::{ready, TryFuture};
 use http::{self, header::AsHeaderName, uri::Authority};
-use pin_project::pin_project;
+use linkerd2_stack::NewService;
 use std::fmt;
-use std::future::Future;
-use std::pin::Pin;
 use std::task::{Context, Poll};
 use tracing::debug;
 
@@ -23,14 +20,6 @@ pub struct MakeSvc<H, M> {
     inner: M,
 }
 
-#[pin_project]
-pub struct MakeSvcFut<M, H> {
-    authority: Option<Authority>,
-    headers_to_strip: Vec<H>,
-    #[pin]
-    inner: M,
-}
-
 #[derive(Clone, Debug)]
 pub struct Service<S, H> {
     authority: Option<Authority>,
@@ -71,49 +60,25 @@ where
     }
 }
 
-impl<H, T, M> tower::Service<T> for MakeSvc<H, M>
+impl<H, T, M> NewService<T> for MakeSvc<H, M>
 where
     T: CanOverrideAuthority + Clone + Send + Sync + 'static,
-    M: tower::Service<T>,
+    M: NewService<T>,
     H: AsHeaderName + Clone,
 {
-    type Response = Service<M::Response, H>;
-    type Error = M::Error;
-    type Future = MakeSvcFut<M::Future, H>;
-
-    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), M::Error>> {
-        self.inner.poll_ready(cx)
-    }
+    type Service = Service<M::Service, H>;
 
-    fn call(&mut self, t: T) -> Self::Future {
+    fn new_service(&mut self, t: T) -> Self::Service {
         let authority = t.override_authority();
-        let inner = self.inner.call(t);
-        MakeSvcFut {
+        let inner = self.inner.new_service(t);
+        Service {
             authority,
             headers_to_strip: self.headers_to_strip.clone(),
             inner,
         }
     }
 }
 
-impl<F, H> Future for MakeSvcFut<F, H>
-where
-    F: TryFuture,
-    H: AsHeaderName + Clone,
-{
-    type Output = Result<Service<F::Ok, H>, F::Error>;
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let this = self.project();
-        let inner = ready!(this.inner.try_poll(cx))?;
-        Poll::Ready(Ok(Service {
-            authority: this.authority.clone(),
-            headers_to_strip: this.headers_to_strip.clone(),
-            inner,
-        }))
-    }
-}
-
 // === impl Service ===
 
 impl<S, H, B> tower::Service<http::Request<B>> for Service<S, H>
diff --git a/linkerd/reconnect/Cargo.toml b/linkerd/reconnect/Cargo.toml
@@ -7,6 +7,7 @@ publish = false
 
 [dependencies]
 linkerd2-error = { path = "../error" }
+linkerd2-stack = { path = "../stack" }
 futures = "0.3"
 tower = { version = "0.3", default-features = false }
 tracing = "0.1.19"
diff --git a/linkerd/reconnect/src/layer.rs b/linkerd/reconnect/src/layer.rs
diff --git a/linkerd/reconnect/src/lib.rs b/linkerd/reconnect/src/lib.rs
diff --git a/linkerd/reconnect/src/service.rs b/linkerd/reconnect/src/service.rs
