Implement profile discovery as a NewService (#673)

The profile discovery layer is always coerced to a NewService. This
change formalizes this to avoid stack boilerplate.

While we're here, the GetProfile trait has ben simplified to avoid
exposing poll_ready in its public API. Instead, we provide an impl of
Service for GetProfile that does this (though it's not currently
needed).

Author: olix0r

linkerd/app/inbound/src/lib.rs

@@ -248,7 +248,6 @@ impl Config {
             ))
             .push_map_target(endpoint::Logical::from)
             .push(profiles::discover::layer(profiles_client))
-            .into_new_service()
             .instrument(|_: &Target| debug_span!("profile"))
             .push_on_response(svc::layers().box_http_response())
             .check_new_service::<Target, http::Request<http::boxed::Payload>>();

linkerd/app/outbound/src/lib.rs

@@ -361,7 +361,6 @@ impl Config {
             // Discovers the service profile from the control plane and passes
             // it to inner stack to build the router and traffic split.
             .push(profiles::discover::layer(profiles_client))
-            .into_new_service()
             .check_new_service::<HttpLogical, http::Request<_>>();
 
         // Caches clients that bypass discovery/balancing.

linkerd/service-profiles/src/discover.rs

@@ -1,11 +1,7 @@
 use super::{GetProfile, Receiver};
 use futures::prelude::*;
-use linkerd2_stack::{layer, NewService};
-use std::{
-    future::Future,
-    pin::Pin,
-    task::{Context, Poll},
-};
+use linkerd2_stack::{layer, FutureService, NewService};
+use std::{future::Future, pin::Pin};
 
 pub fn layer<G: Clone, M>(
     get_profile: G,
@@ -22,28 +18,25 @@ pub struct Discover<G, M> {
     inner: M,
 }
 
-impl<T, G, M> tower::Service<T> for Discover<G, M>
+impl<T, G, M> NewService<T> for Discover<G, M>
 where
     T: Clone + Send + 'static,
     G: GetProfile<T>,
     G::Future: Send + 'static,
     G::Error: Send,
     M: NewService<(Receiver, T)> + Clone + Send + 'static,
 {
-    type Response = M::Service;
-    type Error = G::Error;
-    type Future = Pin<Box<dyn Future<Output = Result<M::Service, G::Error>> + Send + 'static>>;
+    type Service = FutureService<
+        Pin<Box<dyn Future<Output = Result<M::Service, G::Error>> + Send + 'static>>,
+        M::Service,
+    >;
 
-    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.get_profile.poll_ready(cx)
-    }
-
-    fn call(&mut self, target: T) -> Self::Future {
+    fn new_service(&mut self, target: T) -> Self::Service {
         let mut inner = self.inner.clone();
-        Box::pin(
+        FutureService::new(Box::pin(
             self.get_profile
                 .get_profile(target.clone())
                 .map_ok(move |rx| inner.new_service((rx, target))),
-        )
+        ))
     }
 }

linkerd/service-profiles/src/lib.rs

@@ -6,6 +6,7 @@ use std::{
     future::Future,
     task::{Context, Poll},
 };
+use tower::util::{Oneshot, ServiceExt};
 
 mod client;
 pub mod discover;
@@ -28,29 +29,50 @@ pub struct Target {
     pub weight: u32,
 }
 
+#[derive(Clone, Debug)]
+pub struct GetProfileService<P>(P);
+
 /// Watches a destination's Profile.
 pub trait GetProfile<T> {
     type Error: Into<Error>;
     type Future: Future<Output = Result<Receiver, Self::Error>>;
 
-    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
-
     fn get_profile(&mut self, target: T) -> Self::Future;
+
+    fn into_service(self) -> GetProfileService<Self>
+    where
+        Self: Sized,
+    {
+        GetProfileService(self)
+    }
 }
 
 impl<T, S> GetProfile<T> for S
 where
-    S: tower::Service<T, Response = Receiver>,
+    S: tower::Service<T, Response = Receiver> + Clone,
     S::Error: Into<Error>,
 {
     type Error = S::Error;
-    type Future = S::Future;
+    type Future = Oneshot<S, T>;
 
-    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        tower::Service::poll_ready(self, cx)
+    fn get_profile(&mut self, target: T) -> Self::Future {
+        self.clone().oneshot(target)
     }
+}
 
-    fn get_profile(&mut self, target: T) -> Self::Future {
-        tower::Service::call(self, target)
+impl<T, P> tower::Service<T> for GetProfileService<P>
+where
+    P: GetProfile<T>,
+{
+    type Response = Receiver;
+    type Error = P::Error;
+    type Future = P::Future;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, target: T) -> Self::Future {
+        self.0.get_profile(target)
     }
 }