Remove Resolution trait (#606)

zaharidichev · web-flow · commit 5a3c5634282e · 2020-08-12T17:06:04.000-07:00
The `Resolution` trait was initially introduce to ensure that the update stream
is infinite; but this isn't actually desirable in a general way. It's fine for resolution
streams to complete. Consumers (balancers) need to handle this.

This change removes this trait, in favor of a simple `TryStream`. This makes it
easier to use ready-made stream combinators and constructors.
diff --git a/linkerd/proxy/api-resolve/src/resolve.rs b/linkerd/proxy/api-resolve/src/resolve.rs
@@ -104,14 +104,10 @@ where
     }
 }
 
-impl resolve::Resolution for Resolution {
-    type Endpoint = Metadata;
-    type Error = grpc::Status;
+impl Stream for Resolution {
+    type Item = Result<resolve::Update<Metadata>, grpc::Status>;
 
-    fn poll(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Update<Self::Endpoint>, Self::Error>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let mut this = self.project();
         loop {
             match ready!(this.inner.as_mut().poll_next(cx)) {
@@ -126,7 +122,7 @@ impl resolve::Resolution for Resolution {
                             .collect::<Vec<_>>();
                         if !addr_metas.is_empty() {
                             debug!(endpoints = %addr_metas.len(), "Add");
-                            return Poll::Ready(Ok(Update::Add(addr_metas)));
+                            return Poll::Ready(Some(Ok(Update::Add(addr_metas))));
                         }
                     }
 
@@ -137,7 +133,7 @@ impl resolve::Resolution for Resolution {
                             .collect::<Vec<_>>();
                         if !sock_addrs.is_empty() {
                             debug!(endpoints = %sock_addrs.len(), "Remove");
-                            return Poll::Ready(Ok(Update::Remove(sock_addrs)));
+                            return Poll::Ready(Some(Ok(Update::Remove(sock_addrs))));
                         }
                     }
 
@@ -148,14 +144,12 @@ impl resolve::Resolution for Resolution {
                         } else {
                             Update::DoesNotExist
                         };
-                        return Poll::Ready(Ok(update.into()));
+                        return Poll::Ready(Some(Ok(update.into())));
                     }
 
                     None => {} // continue
                 },
-                None => {
-                    return Poll::Ready(Err(grpc::Status::new(grpc::Code::Ok, "end of stream")))
-                }
+                None => return Poll::Ready(None),
             };
         }
     }
diff --git a/linkerd/proxy/core/src/lib.rs b/linkerd/proxy/core/src/lib.rs
@@ -2,4 +2,4 @@
 
 pub mod resolve;
 
-pub use self::resolve::{Resolution, Resolve};
+pub use self::{resolve::Resolve, resolve::Update};
diff --git a/linkerd/proxy/core/src/resolve.rs b/linkerd/proxy/core/src/resolve.rs
@@ -1,14 +1,14 @@
+use futures::stream::TryStream;
 use linkerd2_error::Error;
 use std::future::Future;
 use std::net::SocketAddr;
-use std::pin::Pin;
 use std::task::{Context, Poll};
 
-/// Resolves `T`-typed names/addresses as a `Resolution`.
+/// Resolves `T`-typed names/addresses as an infinite stream of `Update<Self::Endpoint>`.
 pub trait Resolve<T> {
     type Endpoint;
     type Error: Into<Error>;
-    type Resolution: Resolution<Endpoint = Self::Endpoint>;
+    type Resolution: TryStream<Ok = Update<Self::Endpoint>, Error = Self::Error>;
     type Future: Future<Output = Result<Self::Resolution, Self::Error>>;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
@@ -23,27 +23,6 @@ pub trait Resolve<T> {
     }
 }
 
-/// An infinite stream of endpoint updates.
-pub trait Resolution {
-    type Endpoint;
-    type Error: Into<Error>;
-
-    fn poll(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Update<Self::Endpoint>, Self::Error>>;
-
-    fn poll_unpin(
-        &mut self,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Update<Self::Endpoint>, Self::Error>>
-    where
-        Self: Unpin,
-    {
-        Pin::new(self).poll(cx)
-    }
-}
-
 #[derive(Clone, Debug)]
 pub struct Service<S>(S);
 
@@ -57,13 +36,13 @@ pub enum Update<T> {
 
 // === impl Resolve ===
 
-impl<S, T, R> Resolve<T> for S
+impl<S, T, R, E> Resolve<T> for S
 where
     S: tower::Service<T, Response = R>,
     S::Error: Into<Error>,
-    R: Resolution,
+    R: TryStream<Ok = Update<E>, Error = S::Error>,
 {
-    type Endpoint = <R as Resolution>::Endpoint;
+    type Endpoint = E;
     type Error = S::Error;
     type Resolution = S::Response;
     type Future = S::Future;
diff --git a/linkerd/proxy/discover/src/from_resolve.rs b/linkerd/proxy/discover/src/from_resolve.rs
@@ -1,6 +1,6 @@
-use futures::{ready, Stream, TryFuture};
+use futures::{ready, Stream, TryFuture, TryStream};
 use indexmap::IndexSet;
-use linkerd2_proxy_core::resolve::{Resolution, Resolve, Update};
+use linkerd2_proxy_core::resolve::{Resolve, Update};
 use pin_project::pin_project;
 use std::collections::VecDeque;
 use std::future::Future;
@@ -10,45 +10,50 @@ use std::task::{Context, Poll};
 use tower::discover::Change;
 
 #[derive(Clone, Debug)]
-pub struct FromResolve<R> {
+pub struct FromResolve<R, E> {
     resolve: R,
+    _marker: std::marker::PhantomData<fn(E)>,
 }
 
 #[pin_project]
 #[derive(Debug)]
-pub struct DiscoverFuture<F> {
+pub struct DiscoverFuture<F, E> {
     #[pin]
     future: F,
+    _marker: std::marker::PhantomData<fn(E)>,
 }
 
 /// Observes an `R`-typed resolution stream, using an `M`-typed endpoint stack to
 /// build a service for each endpoint.
 #[pin_project]
-pub struct Discover<R: Resolution> {
+pub struct Discover<R: TryStream, E> {
     #[pin]
     resolution: R,
     active: IndexSet<SocketAddr>,
-    pending: VecDeque<Change<SocketAddr, R::Endpoint>>,
+    pending: VecDeque<Change<SocketAddr, E>>,
 }
 
 // === impl FromResolve ===
 
-impl<R> FromResolve<R> {
+impl<R, E> FromResolve<R, E> {
     pub fn new<T>(resolve: R) -> Self
     where
         R: Resolve<T>,
     {
-        Self { resolve }
+        Self {
+            resolve,
+            _marker: std::marker::PhantomData,
+        }
     }
 }
 
-impl<T, R> tower::Service<T> for FromResolve<R>
+impl<T, R, E> tower::Service<T> for FromResolve<R, E>
 where
     R: Resolve<T> + Clone,
 {
-    type Response = Discover<R::Resolution>;
+    type Response = Discover<R::Resolution, E>;
     type Error = R::Error;
-    type Future = DiscoverFuture<R::Future>;
+    type Future = DiscoverFuture<R::Future, E>;
 
     #[inline]
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
@@ -59,18 +64,19 @@ where
     fn call(&mut self, target: T) -> Self::Future {
         Self::Future {
             future: self.resolve.resolve(target),
+            _marker: std::marker::PhantomData,
         }
     }
 }
 
 // === impl DiscoverFuture ===
 
-impl<F> Future for DiscoverFuture<F>
+impl<F, E> Future for DiscoverFuture<F, E>
 where
     F: TryFuture,
-    F::Ok: Resolution,
+    F::Ok: TryStream,
 {
-    type Output = Result<Discover<F::Ok>, F::Error>;
+    type Output = Result<Discover<F::Ok, E>, F::Error>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let resolution = ready!(self.project().future.try_poll(cx))?;
@@ -80,7 +86,7 @@ where
 
 // === impl Discover ===
 
-impl<R: Resolution> Discover<R> {
+impl<R: TryStream, E> Discover<R, E> {
     pub fn new(resolution: R) -> Self {
         Self {
             resolution,
@@ -90,8 +96,11 @@ impl<R: Resolution> Discover<R> {
     }
 }
 
-impl<R: Resolution> Stream for Discover<R> {
-    type Item = Result<Change<SocketAddr, R::Endpoint>, R::Error>;
+impl<R, E> Stream for Discover<R, E>
+where
+    R: TryStream<Ok = Update<E>>,
+{
+    type Item = Result<Change<SocketAddr, E>, R::Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         loop {
@@ -100,24 +109,27 @@ impl<R: Resolution> Stream for Discover<R> {
                 return Poll::Ready(Some(Ok(change)));
             }
 
-            match ready!(this.resolution.poll(cx))? {
-                Update::Add(endpoints) => {
-                    for (addr, endpoint) in endpoints.into_iter() {
-                        this.active.insert(addr);
-                        this.pending.push_back(Change::Insert(addr, endpoint));
+            match ready!(this.resolution.try_poll_next(cx)) {
+                Some(update) => match update? {
+                    Update::Add(endpoints) => {
+                        for (addr, endpoint) in endpoints.into_iter() {
+                            this.active.insert(addr);
+                            this.pending.push_back(Change::Insert(addr, endpoint));
+                        }
                     }
-                }
-                Update::Remove(addrs) => {
-                    for addr in addrs.into_iter() {
-                        if this.active.remove(&addr) {
-                            this.pending.push_back(Change::Remove(addr));
+                    Update::Remove(addrs) => {
+                        for addr in addrs.into_iter() {
+                            if this.active.remove(&addr) {
+                                this.pending.push_back(Change::Remove(addr));
+                            }
                         }
                     }
-                }
-                Update::DoesNotExist | Update::Empty => {
-                    this.pending
-                        .extend(this.active.drain(..).map(Change::Remove));
-                }
+                    Update::DoesNotExist | Update::Empty => {
+                        this.pending
+                            .extend(this.active.drain(..).map(Change::Remove));
+                    }
+                },
+                None => return Poll::Ready(None),
             }
         }
     }
diff --git a/linkerd/proxy/discover/src/lib.rs b/linkerd/proxy/discover/src/lib.rs
@@ -43,15 +43,15 @@ where
     T: fmt::Display,
     R: Resolve<T> + Send + Clone + 'static,
     R::Error: Into<Error>,
-    R::Endpoint: fmt::Debug + Clone + PartialEq + Send,
+    R::Endpoint: fmt::Debug + Clone + PartialEq + Send + 'static,
     R::Resolution: Send + 'static,
     R::Future: Send + 'static,
     M: tower::Service<R::Endpoint> + Clone + Send + 'static,
     M::Error: Into<Error>,
     M::Response: Send + 'static,
     M::Future: Send + 'static,
 {
-    type Service = Buffer<MakeEndpoint<FromResolve<R>, M>>;
+    type Service = Buffer<MakeEndpoint<FromResolve<R, R::Endpoint>, M>>;
 
     fn layer(&self, make_endpoint: M) -> Self::Service {
         let make_discover =
diff --git a/linkerd/proxy/discover/src/make_endpoint.rs b/linkerd/proxy/discover/src/make_endpoint.rs
@@ -195,18 +195,17 @@ where
             // services. Don't process any updates until we can do so.
             ready!(this.make_endpoint.poll_ready(cx)).map_err(Into::into)?;
 
-            match ready!(this.discover.poll_discover(cx))
-                .expect("XXX(eliza): can this ever be none???")
-                .map_err(Into::into)?
-            {
-                Change::Insert(key, target) => {
-                    // Start building the service and continue. If a pending
-                    // service exists for this addr, it will be canceled.
-                    let fut = this.make_endpoint.call(target);
-                    this.make_futures.push(key, fut);
-                }
-                Change::Remove(key) => {
-                    this.pending_removals.push(key);
+            if let Some(change) = ready!(this.discover.poll_discover(cx)) {
+                match change.map_err(Into::into)? {
+                    Change::Insert(key, target) => {
+                        // Start building the service and continue. If a pending
+                        // service exists for this addr, it will be canceled.
+                        let fut = this.make_endpoint.call(target);
+                        this.make_futures.push(key, fut);
+                    }
+                    Change::Remove(key) => {
+                        this.pending_removals.push(key);
+                    }
                 }
             }
         }
diff --git a/linkerd/proxy/resolve/src/make_unpin.rs b/linkerd/proxy/resolve/src/make_unpin.rs
@@ -1,5 +1,6 @@
+use futures::stream::{Stream, TryStream};
 use futures::TryFuture;
-use linkerd2_proxy_core::resolve::{self, Resolution, Update};
+use linkerd2_proxy_core::resolve;
 use pin_project::pin_project;
 use std::future::Future;
 use std::pin::Pin;
@@ -38,15 +39,11 @@ impl<T> MakeUnpin<T> {
     }
 }
 
-impl<T: Resolution> Resolution for MakeUnpin<T> {
-    type Endpoint = T::Endpoint;
-    type Error = T::Error;
+impl<T: TryStream> Stream for MakeUnpin<T> {
+    type Item = Result<T::Ok, T::Error>;
 
-    fn poll(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Update<Self::Endpoint>, Self::Error>> {
-        self.project().0.as_mut().as_mut().poll(cx)
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.project().0.as_mut().as_mut().try_poll_next(cx)
     }
 }
 
diff --git a/linkerd/proxy/resolve/src/map_endpoint.rs b/linkerd/proxy/resolve/src/map_endpoint.rs
diff --git a/linkerd/proxy/resolve/src/recover.rs b/linkerd/proxy/resolve/src/recover.rs

Original file line number	Diff line number	Diff line change
`@@ -104,14 +104,10 @@ where`
`104`	`104`	`}`
`105`	`105`	`}`
`106`	`106`
`107`		`-impl resolve::Resolution for Resolution {`
`108`		`- type Endpoint = Metadata;`
`109`		`- type Error = grpc::Status;`
	`107`	`+impl Stream for Resolution {`
	`108`	`+ type Item = Result<resolve::Update<Metadata>, grpc::Status>;`
`110`	`109`
`111`		`- fn poll(`
`112`		`- self: Pin<&mut Self>,`
`113`		`- cx: &mut Context<'_>,`
`114`		`- ) -> Poll<Result<Update<Self::Endpoint>, Self::Error>> {`
	`110`	`+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {`
`115`	`111`	`let mut this = self.project();`
`116`	`112`	`loop {`
`117`	`113`	`match ready!(this.inner.as_mut().poll_next(cx)) {`
`@@ -126,7 +122,7 @@ impl resolve::Resolution for Resolution {`
`126`	`122`	`.collect::<Vec<_>>();`
`127`	`123`	`if !addr_metas.is_empty() {`
`128`	`124`	`debug!(endpoints = %addr_metas.len(), "Add");`
`129`		`- return Poll::Ready(Ok(Update::Add(addr_metas)));`
	`125`	`+ return Poll::Ready(Some(Ok(Update::Add(addr_metas))));`
`130`	`126`	`}`
`131`	`127`	`}`
`132`	`128`
`@@ -137,7 +133,7 @@ impl resolve::Resolution for Resolution {`
`137`	`133`	`.collect::<Vec<_>>();`
`138`	`134`	`if !sock_addrs.is_empty() {`
`139`	`135`	`debug!(endpoints = %sock_addrs.len(), "Remove");`
`140`		`- return Poll::Ready(Ok(Update::Remove(sock_addrs)));`
	`136`	`+ return Poll::Ready(Some(Ok(Update::Remove(sock_addrs))));`
`141`	`137`	`}`
`142`	`138`	`}`
`143`	`139`
`@@ -148,14 +144,12 @@ impl resolve::Resolution for Resolution {`
`148`	`144`	`} else {`
`149`	`145`	`Update::DoesNotExist`
`150`	`146`	`};`
`151`		`- return Poll::Ready(Ok(update.into()));`
	`147`	`+ return Poll::Ready(Some(Ok(update.into())));`
`152`	`148`	`}`
`153`	`149`
`154`	`150`	`None => {} // continue`
`155`	`151`	`},`
`156`		`- None => {`
`157`		`- return Poll::Ready(Err(grpc::Status::new(grpc::Code::Ok, "end of stream")))`
`158`		`- }`
	`152`	`+ None => return Poll::Ready(None),`
`159`	`153`	`};`
`160`	`154`	`}`
`161`	`155`	`}`
Original file line number	Diff line number	Diff line change
`@@ -2,4 +2,4 @@`
`2`	`2`
`3`	`3`	`pub mod resolve;`
`4`	`4`
`5`		`-pub use self::resolve::{Resolution, Resolve};`
	`5`	`+pub use self::{resolve::Resolve, resolve::Update};`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,6 @@`
	`1`	`+use futures::stream::{Stream, TryStream};`
`1`	`2`	`use futures::TryFuture;`
`2`		`-use linkerd2_proxy_core::resolve::{self, Resolution, Update};`
	`3`	`+use linkerd2_proxy_core::resolve;`
`3`	`4`	`use pin_project::pin_project;`
`4`	`5`	`use std::future::Future;`
`5`	`6`	`use std::pin::Pin;`
`@@ -38,15 +39,11 @@ impl<T> MakeUnpin<T> {`
`38`	`39`	`}`
`39`	`40`	`}`
`40`	`41`
`41`		`-impl<T: Resolution> Resolution for MakeUnpin<T> {`
`42`		`- type Endpoint = T::Endpoint;`
`43`		`- type Error = T::Error;`
	`42`	`+impl<T: TryStream> Stream for MakeUnpin<T> {`
	`43`	`+ type Item = Result<T::Ok, T::Error>;`
`44`	`44`
`45`		`- fn poll(`
`46`		`- self: Pin<&mut Self>,`
`47`		`- cx: &mut Context<'_>,`
`48`		`- ) -> Poll<Result<Update<Self::Endpoint>, Self::Error>> {`
`49`		`- self.project().0.as_mut().as_mut().poll(cx)`
	`45`	`+ fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {`
	`46`	`+ self.project().0.as_mut().as_mut().try_poll_next(cx)`
`50`	`47`	`}`
`51`	`48`	`}`
`52`	`49`