fix(outbound): apply opaq filters in connection handler (linkerd#3378)

zaharidichev · olix0r · web-flow · commit 7b782128a9fe · 2024-11-22T09:48:35.000-08:00
The initial implementation of opaq filters uses `push_filter` so that
errors originate from `Service::poll_ready` rather than `Service::call`.
This causes the transport metrics middleware to not count failed
connections.

By changing the filter contract so that filters are applied in `Service::call`,
we ensure that middlewares see response failures rather than readiness
failures.

Signed-off-by: Zahari Dichev &lt;zaharidichev@gmail.com&gt;
Co-authored-by: Oliver Gould &lt;ver@buoyant.io&gt;
diff --git a/linkerd/app/outbound/src/opaq/logical/route.rs b/linkerd/app/outbound/src/opaq/logical/route.rs
@@ -91,15 +91,15 @@ where
                 .push_map_target(|t| t)
                 .push_map_target(|b: Backend<T>| b.concrete)
                 // apply backend filters
-                .push_filter(filters::apply)
+                .push(filters::NewApplyFilters::layer())
                 .lift_new()
                 .push(NewDistribute::layer())
                 // The router does not take the backend's availability into
                 // consideration, so we must eagerly fail requests to prevent
                 // leaking tasks onto the runtime.
                 .push_on_service(svc::LoadShed::layer())
                 // apply route level filters
-                .push_filter(filters::apply)
+                .push(filters::NewApplyFilters::layer())
                 .push(svc::NewMapErr::layer_with(|rt: &Self| {
                     let route = rt.params.route_ref.clone();
                     move |source| RouteError {
diff --git a/linkerd/app/outbound/src/opaq/logical/route/filters.rs b/linkerd/app/outbound/src/opaq/logical/route/filters.rs
@@ -1,30 +1,97 @@
-use linkerd_app_core::{svc, Error};
+use futures::{future, TryFutureExt};
+use linkerd_app_core::{io, svc, Error};
 use linkerd_proxy_client_policy::opaq;
-use std::{fmt::Debug, sync::Arc};
+use std::{
+    fmt::Debug,
+    sync::Arc,
+    task::{Context, Poll},
+};
 
-pub(crate) fn apply<T>(t: T) -> Result<T, Error>
+#[derive(Clone, Debug)]
+pub struct NewApplyFilters<N> {
+    inner: N,
+}
+
+#[derive(Clone, Debug)]
+pub struct ApplyFilters<S> {
+    inner: S,
+    filters: Arc<[opaq::Filter]>,
+}
+
+// === impl NewApplyFilters ===
+
+impl<N> NewApplyFilters<N> {
+    pub fn layer() -> impl svc::layer::Layer<N, Service = Self> + Clone {
+        svc::layer::mk(move |inner| Self { inner })
+    }
+}
+
+impl<T, N, S> svc::NewService<T> for NewApplyFilters<N>
 where
-    T: Clone,
+    N: svc::NewService<T, Service = S>,
     T: svc::Param<Arc<[opaq::Filter]>>,
 {
-    let filters: &[opaq::Filter] = &t.param();
-    if let Some(filter) = filters.iter().next() {
-        match filter {
-            opaq::Filter::Forbidden => {
-                return Err(errors::TCPForbiddenRoute.into());
-            }
+    type Service = ApplyFilters<S>;
 
-            opaq::Filter::Invalid(message) => {
-                return Err(errors::TCPInvalidBackend(message.clone()).into());
-            }
+    fn new_service(&self, target: T) -> Self::Service {
+        let filters: Arc<[opaq::Filter]> = target.param();
+        let svc = self.inner.new_service(target);
+        ApplyFilters {
+            inner: svc,
+            filters,
+        }
+    }
+}
 
-            opaq::Filter::InternalError(message) => {
-                return Err(errors::TCPInvalidPolicy(message).into());
+// === impl ApplyFilters ===
+
+impl<S> ApplyFilters<S> {
+    fn apply_filters(&self) -> Result<(), Error> {
+        if let Some(filter) = self.filters.iter().next() {
+            match filter {
+                opaq::Filter::Forbidden => {
+                    return Err(errors::TCPForbiddenRoute.into());
+                }
+
+                opaq::Filter::Invalid(message) => {
+                    return Err(errors::TCPInvalidBackend(message.clone()).into());
+                }
+
+                opaq::Filter::InternalError(message) => {
+                    return Err(errors::TCPInvalidPolicy(message).into());
+                }
             }
         }
+
+        Ok(())
+    }
+}
+
+impl<I, S> svc::Service<I> for ApplyFilters<S>
+where
+    I: io::AsyncRead + io::AsyncWrite + Send + 'static,
+    S: svc::Service<I> + Send + Clone + 'static,
+    S::Error: Into<Error>,
+    S::Future: Send,
+{
+    type Response = S::Response;
+    type Error = Error;
+    type Future = future::Either<
+        future::ErrInto<S::Future, Error>,
+        future::Ready<Result<S::Response, Error>>,
+    >;
+
+    #[inline]
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx).map_err(Into::into)
     }
 
-    Ok(t)
+    fn call(&mut self, io: I) -> Self::Future {
+        if let Err(e) = self.apply_filters() {
+            return future::Either::Right(future::err(e));
+        }
+        future::Either::Left(self.inner.call(io).err_into())
+    }
 }
 
 pub mod errors {
diff --git a/linkerd/app/outbound/src/tls/logical/route.rs b/linkerd/app/outbound/src/tls/logical/route.rs
@@ -95,15 +95,15 @@ where
                 .push_map_target(|t| t)
                 .push_map_target(|b: Backend<T>| b.concrete)
                 // apply backend filters
-                .push_filter(filters::apply)
+                .push(filters::NewApplyFilters::layer())
                 .lift_new()
                 .push(NewDistribute::layer())
                 // The router does not take the backend's availability into
                 // consideration, so we must eagerly fail requests to prevent
                 // leaking tasks onto the runtime.
                 .push_on_service(svc::LoadShed::layer())
                 // apply route level filters
-                .push_filter(filters::apply)
+                .push(filters::NewApplyFilters::layer())
                 .push(svc::NewMapErr::layer_with(|rt: &Self| {
                     let route = rt.params.route_ref.clone();
                     move |source| RouteError {
diff --git a/linkerd/app/outbound/src/tls/logical/route/filters.rs b/linkerd/app/outbound/src/tls/logical/route/filters.rs
@@ -1,30 +1,97 @@
-use linkerd_app_core::{svc, Error};
+use futures::{future, TryFutureExt};
+use linkerd_app_core::{io, svc, Error};
 use linkerd_proxy_client_policy::tls;
-use std::{fmt::Debug, sync::Arc};
+use std::{
+    fmt::Debug,
+    sync::Arc,
+    task::{Context, Poll},
+};
 
-pub(crate) fn apply<T>(t: T) -> Result<T, Error>
+#[derive(Clone, Debug)]
+pub struct NewApplyFilters<N> {
+    inner: N,
+}
+
+#[derive(Clone, Debug)]
+pub struct ApplyFilters<S> {
+    inner: S,
+    filters: Arc<[tls::Filter]>,
+}
+
+// === impl NewApplyFilters ===
+
+impl<N> NewApplyFilters<N> {
+    pub fn layer() -> impl svc::layer::Layer<N, Service = Self> + Clone {
+        svc::layer::mk(move |inner| Self { inner })
+    }
+}
+
+impl<T, N, S> svc::NewService<T> for NewApplyFilters<N>
 where
-    T: Clone,
+    N: svc::NewService<T, Service = S>,
     T: svc::Param<Arc<[tls::Filter]>>,
 {
-    let filters: &[tls::Filter] = &t.param();
-    if let Some(filter) = filters.iter().next() {
-        match filter {
-            tls::Filter::Forbidden => {
-                return Err(errors::TLSForbiddenRoute.into());
-            }
+    type Service = ApplyFilters<S>;
 
-            tls::Filter::Invalid(message) => {
-                return Err(errors::TLSInvalidBackend(message.clone()).into());
-            }
+    fn new_service(&self, target: T) -> Self::Service {
+        let filters: Arc<[tls::Filter]> = target.param();
+        let svc = self.inner.new_service(target);
+        ApplyFilters {
+            inner: svc,
+            filters,
+        }
+    }
+}
 
-            tls::Filter::InternalError(message) => {
-                return Err(errors::TLSInvalidPolicy(message).into());
+// === impl ApplyFilters ===
+
+impl<S> ApplyFilters<S> {
+    fn apply_filters(&self) -> Result<(), Error> {
+        if let Some(filter) = self.filters.iter().next() {
+            match filter {
+                tls::Filter::Forbidden => {
+                    return Err(errors::TLSForbiddenRoute.into());
+                }
+
+                tls::Filter::Invalid(message) => {
+                    return Err(errors::TLSInvalidBackend(message.clone()).into());
+                }
+
+                tls::Filter::InternalError(message) => {
+                    return Err(errors::TLSInvalidPolicy(message).into());
+                }
             }
         }
+
+        Ok(())
+    }
+}
+
+impl<I, S> svc::Service<I> for ApplyFilters<S>
+where
+    I: io::AsyncRead + io::AsyncWrite + Send + 'static,
+    S: svc::Service<I> + Send + Clone + 'static,
+    S::Error: Into<Error>,
+    S::Future: Send,
+{
+    type Response = S::Response;
+    type Error = Error;
+    type Future = future::Either<
+        future::ErrInto<S::Future, Error>,
+        future::Ready<Result<S::Response, Error>>,
+    >;
+
+    #[inline]
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx).map_err(Into::into)
     }
 
-    Ok(t)
+    fn call(&mut self, io: I) -> Self::Future {
+        if let Err(e) = self.apply_filters() {
+            return future::Either::Right(future::err(e));
+        }
+        future::Either::Left(self.inner.call(io).err_into())
+    }
 }
 
 pub mod errors {
