request-filter: Allow altering request type

Cargo.lock

@@ -1462,7 +1462,7 @@ version = "0.1.0"
 dependencies = [
  "futures 0.3.5",
  "linkerd2-error",
- "pin-project",
+ "linkerd2-stack",
  "tower",
  "tracing",
 ]

linkerd/app/core/src/svc.rs

@@ -2,7 +2,7 @@
 
 pub use crate::proxy::http;
 use crate::transport::Connect;
-use crate::{cache, request_filter, Error};
+use crate::{cache, Error};
 pub use linkerd2_buffer as buffer;
 use linkerd2_concurrency_limit as concurrency_limit;
 pub use linkerd2_stack::{self as stack, layer, NewService};
@@ -267,10 +267,6 @@ impl<S> Stack<S> {
         self.push(stack::FallbackLayer::new(fallback).with_predicate(predicate))
     }
 
-    pub fn push_request_filter<F: Clone>(self, filter: F) -> Stack<request_filter::Service<F, S>> {
-        self.push(request_filter::RequestFilterLayer::new(filter))
-    }
-
     // pub fn box_http_request<B>(self) -> Stack<http::boxed::BoxRequest<S, B>>
     // where
     //     B: hyper::body::HttpBody<Data = http::boxed::Data, Error = Error> + 'static,

linkerd/app/src/dst/mod.rs

@@ -3,7 +3,7 @@ mod resolve;
 
 use indexmap::IndexSet;
 use linkerd2_app_core::{
-    control, dns, profiles, proxy::identity, request_filter, svc, transport::tls,
+    control, dns, profiles, proxy::identity, request_filter::RequestFilter, svc, transport::tls,
     ControlHttpMetrics, Error,
 };
 use permit::PermitConfiguredDsts;
@@ -26,12 +26,11 @@ pub struct Config {
 /// The addr is preserved for logging.
 pub struct Dst {
     pub addr: control::ControlAddr,
-    pub profiles: request_filter::Service<
+    pub profiles: RequestFilter<
         PermitConfiguredDsts<profiles::InvalidProfileAddr>,
         profiles::Client<control::Client<BoxBody>, resolve::BackoffUnlessInvalidArgument>,
     >,
-    pub resolve:
-        request_filter::Service<PermitConfiguredDsts, resolve::Resolve<control::Client<BoxBody>>>,
+    pub resolve: RequestFilter<PermitConfiguredDsts, resolve::Resolve<control::Client<BoxBody>>>,
 }
 
 impl Config {
@@ -45,10 +44,10 @@ impl Config {
         let backoff = self.control.connect.backoff.clone();
         let svc = self.control.build(dns, metrics, identity);
         let resolve = svc::stack(resolve::new(svc.clone(), &self.context, backoff))
-            .push_request_filter(PermitConfiguredDsts::new(
+            .push(RequestFilter::layer(PermitConfiguredDsts::new(
                 self.get_suffixes,
                 self.get_networks,
-            ))
+            )))
             .into_inner();
 
         let profiles = svc::stack(profiles::Client::new(
@@ -57,10 +56,10 @@ impl Config {
             self.initial_profile_timeout,
             self.context,
         ))
-        .push_request_filter(
+        .push(RequestFilter::layer(
             PermitConfiguredDsts::new(self.profile_suffixes, self.profile_networks)
                 .with_error::<profiles::InvalidProfileAddr>(),
-        )
+        ))
         .into_inner();
 
         Ok(Dst {

linkerd/app/src/dst/permit.rs

@@ -1,5 +1,7 @@
 use ipnet::{Contains, IpNet};
-use linkerd2_app_core::{dns::Suffix, request_filter, Addr, DiscoveryRejected, Error};
+use linkerd2_app_core::{
+    dns::Suffix, request_filter::FilterRequest, Addr, DiscoveryRejected, Error,
+};
 use std::marker::PhantomData;
 use std::net::IpAddr;
 use std::sync::Arc;
@@ -48,14 +50,14 @@ impl<E> Clone for PermitConfiguredDsts<E> {
     }
 }
 
-impl<T, E> request_filter::RequestFilter<T> for PermitConfiguredDsts<E>
+impl<T, E> FilterRequest<T> for PermitConfiguredDsts<E>
 where
     E: Into<Error> + From<Addr>,
     for<'t> &'t T: Into<Addr>,
 {
-    type Error = E;
+    type Request = T;
 
-    fn filter(&self, t: T) -> Result<T, Self::Error> {
+    fn filter(&self, t: T) -> Result<T, Error> {
         let addr = (&t).into();
         let permitted = match addr {
             Addr::Name(ref name) => self
@@ -72,7 +74,7 @@ where
         if permitted {
             Ok(t)
         } else {
-            Err(E::from(addr.clone()))
+            Err(E::from(addr.clone()).into())
         }
     }
 }

linkerd/request-filter/Cargo.toml

@@ -7,7 +7,7 @@ publish = false
 
 [dependencies]
 futures = "0.3"
+linkerd2-error = { path = "../error" }
+linkerd2-stack = { path = "../stack" }
 tower = { version = "0.3", default-features = false }
 tracing = "0.1.19"
-linkerd2-error = { path = "../error" }
-pin-project = "0.4"

linkerd/request-filter/src/lib.rs

@@ -3,71 +3,51 @@
 
 #![deny(warnings, rust_2018_idioms)]
 
+use futures::{future, prelude::*};
 use linkerd2_error::Error;
-use pin_project::pin_project;
-use std::future::Future;
-use std::pin::Pin;
+use linkerd2_stack::layer;
 use std::task::{Context, Poll};
 
-pub trait RequestFilter<T> {
-    type Error: Into<Error>;
+pub trait FilterRequest<Req> {
+    type Request;
 
-    fn filter(&self, request: T) -> Result<T, Self::Error>;
+    fn filter(&self, request: Req) -> Result<Self::Request, Error>;
 }
 
 #[derive(Clone, Debug)]
-pub struct RequestFilterLayer<T> {
-    filter: T,
-}
-
-#[derive(Clone, Debug)]
-pub struct Service<I, S> {
+pub struct RequestFilter<I, S> {
     filter: I,
     service: S,
 }
 
-#[pin_project(project = ResponseFutureProj)]
-#[derive(Debug)]
-pub enum ResponseFuture<F> {
-    Future(#[pin] F),
-    Rejected(Option<Error>),
-}
-
-// === impl Layer ===
-
-impl<T: Clone> RequestFilterLayer<T> {
-    pub fn new(filter: T) -> Self {
-        Self { filter }
-    }
-}
-
-impl<T: Clone, S> tower::Layer<S> for RequestFilterLayer<T> {
-    type Service = Service<T, S>;
-
-    fn layer(&self, inner: S) -> Self::Service {
-        Service::new(self.filter.clone(), inner)
-    }
-}
+// === impl RequestFilter ===
 
-// === impl Service ===
-
-impl<I, S> Service<I, S> {
+impl<I, S> RequestFilter<I, S> {
     pub fn new(filter: I, service: S) -> Self {
         Self { filter, service }
     }
+
+    pub fn layer(filter: I) -> impl layer::Layer<S, Service = Self> + Clone
+    where
+        I: Clone,
+    {
+        layer::mk(move |inner| Self::new(filter.clone(), inner))
+    }
 }
 
-impl<T, I, S> tower::Service<T> for Service<I, S>
+impl<T, F, S> tower::Service<T> for RequestFilter<F, S>
 where
-    I: RequestFilter<T>,
-    S: tower::Service<T>,
+    F: FilterRequest<T>,
+    S: tower::Service<F::Request>,
     S::Error: Into<Error>,
 {
     type Response = S::Response;
     type Error = Error;
-    type Future = ResponseFuture<S::Future>;
+    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.service.poll_ready(cx).map_err(Into::into)
     }
@@ -76,30 +56,12 @@ where
         match self.filter.filter(request) {
             Ok(req) => {
                 tracing::trace!("accepted");
-                let f = self.service.call(req);
-                ResponseFuture::Future(f)
+                future::Either::Left(self.service.call(req).err_into::<Error>())
             }
             Err(e) => {
                 tracing::trace!("rejected");
-                ResponseFuture::Rejected(Some(e.into()))
+                future::Either::Right(future::err(e))
             }
         }
     }
 }
-
-// === impl ResponseFuture ===
-
-impl<F, T, E> Future for ResponseFuture<F>
-where
-    F: Future<Output = Result<T, E>>,
-    E: Into<Error>,
-{
-    type Output = Result<T, Error>;
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        match self.project() {
-            ResponseFutureProj::Future(f) => f.poll(cx).map(|r| r.map_err(Into::into)),
-            ResponseFutureProj::Rejected(e) => Poll::Ready(Err(e.take().unwrap())),
-        }
-    }
-}