stack: Make the router fallible

linkerd/app/inbound/src/target.rs

@@ -8,7 +8,7 @@ use linkerd_app_core::{
     tls,
     transport::{self, listen},
     transport_header::TransportHeader,
-    Addr, Conditional, CANONICAL_DST_HEADER, DST_OVERRIDE_HEADER,
+    Addr, Conditional, Error, CANONICAL_DST_HEADER, DST_OVERRIDE_HEADER,
 };
 use std::{convert::TryInto, net::SocketAddr, str::FromStr, sync::Arc};
 use tracing::debug;
@@ -301,7 +301,7 @@ impl From<HttpAccept> for RequestTarget {
 impl<A> svc::stack::RecognizeRoute<http::Request<A>> for RequestTarget {
     type Key = Target;
 
-    fn recognize(&self, req: &http::Request<A>) -> Self::Key {
+    fn recognize(&self, req: &http::Request<A>) -> Result<Self::Key, Error> {
         let dst = req
             .headers()
             .get(CANONICAL_DST_HEADER)
@@ -325,7 +325,7 @@ impl<A> svc::stack::RecognizeRoute<http::Request<A>> for RequestTarget {
             .or_else(|| http_request_host_addr(req).ok())
             .unwrap_or_else(|| self.accept.tcp.target_addr.into());
 
-        Target {
+        Ok(Target {
             dst,
             target_addr: self.accept.tcp.target_addr,
             tls: self.accept.tcp.tls.clone(),
@@ -335,7 +335,7 @@ impl<A> svc::stack::RecognizeRoute<http::Request<A>> for RequestTarget {
                 .version()
                 .try_into()
                 .expect("HTTP version must be valid"),
-        }
+        })
     }
 }
 

linkerd/app/outbound/src/ingress.rs

@@ -187,10 +187,10 @@ impl TargetPerRequest {
 impl<B> svc::stack::RecognizeRoute<http::Request<B>> for TargetPerRequest {
     type Key = Target;
 
-    fn recognize(&self, req: &http::Request<B>) -> Self::Key {
-        Target {
+    fn recognize(&self, req: &http::Request<B>) -> Result<Self::Key, Error> {
+        Ok(Target {
             accept: self.0,
             dst: http_request_l5d_override_dst_addr(req).unwrap_or_else(|_| self.0.orig_dst.into()),
-        }
+        })
     }
 }

linkerd/stack/src/router.rs

@@ -1,12 +1,14 @@
 use crate::{layer, NewService};
+use futures::{future, TryFutureExt};
+use linkerd_error::Error;
 use std::task::{Context, Poll};
 use tower::util::{Oneshot, ServiceExt};
 
 /// Determines the router key for each `T`-typed target.
 pub trait RecognizeRoute<T> {
     type Key: Clone;
 
-    fn recognize(&self, t: &T) -> Self::Key;
+    fn recognize(&self, t: &T) -> Result<Self::Key, Error>;
 }
 
 #[derive(Clone, Debug)]
@@ -61,29 +63,39 @@ where
     K: RecognizeRoute<Req>,
     N: NewService<K::Key, Service = S>,
     S: tower::Service<Req>,
+    S::Error: Into<Error>,
 {
     type Response = S::Response;
-    type Error = S::Error;
-    type Future = Oneshot<S, Req>;
+    type Error = Error;
+    type Future = future::Either<
+        future::ErrInto<Oneshot<S, Req>, Error>,
+        future::Ready<Result<S::Response, Error>>,
+    >;
 
-    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+    #[inline]
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Error>> {
         Poll::Ready(Ok(()))
     }
 
     fn call(&mut self, req: Req) -> Self::Future {
-        let key = self.recognize.recognize(&req);
-        self.inner.new_service(key).oneshot(req)
+        match self.recognize.recognize(&req) {
+            Ok(key) => {
+                let svc = self.inner.new_service(key);
+                future::Either::Left(svc.oneshot(req).err_into::<Error>())
+            }
+            Err(e) => future::Either::Right(future::err(e)),
+        }
     }
 }
 
 impl<T, K, F> RecognizeRoute<T> for F
 where
     K: Clone,
-    F: Fn(&T) -> K,
+    F: Fn(&T) -> Result<K, Error>,
 {
     type Key = K;
 
-    fn recognize(&self, t: &T) -> Self::Key {
+    fn recognize(&self, t: &T) -> Result<Self::Key, Error> {
         (self)(t)
     }
 }