Make SkipDetect more generic as stack::MakeSwitch (#657)

SkipDetect is needlessly coupled to its `listen::Addrs` target type and
the Accept stack's shape. In reality, it's pretty similar to the
fallback module, except that decisions are made on the target rather
than the response future.

This change replaces the `core::proxy::skip_detct` module with
`stack::switch`.

Above all, this allows us to use other targets with this module, which
is needed to do discovery before protoocl detection (and to use the
results of discovery to inform detection).

Author: olix0r

linkerd/app/core/src/config.rs

@@ -1,8 +1,6 @@
 pub use crate::exp_backoff::ExponentialBackoff;
 pub use crate::proxy::http::h2;
 pub use crate::transport::{Bind, DefaultOrigDstAddr, NoOrigDstAddr, OrigDstAddr};
-use indexmap::IndexSet;
-use std::sync::Arc;
 use std::time::Duration;
 
 #[derive(Clone, Debug)]
@@ -25,7 +23,7 @@ pub struct ProxyConfig {
     pub connect: ConnectConfig,
     pub buffer_capacity: usize,
     pub cache_max_idle_age: Duration,
-    pub disable_protocol_detection_for_ports: Arc<IndexSet<u16>>,
+    pub disable_protocol_detection_for_ports: crate::SkipByPort,
     pub dispatch_timeout: Duration,
     pub max_in_flight_requests: usize,
     pub detect_protocol_timeout: Duration,

linkerd/app/core/src/lib.rs

@@ -121,3 +121,21 @@ impl From<Addr> for DiscoveryRejected {
         Self::new()
     }
 }
+
+#[derive(Clone, Debug)]
+pub struct SkipByPort(std::sync::Arc<indexmap::IndexSet<u16>>);
+
+impl From<indexmap::IndexSet<u16>> for SkipByPort {
+    fn from(ports: indexmap::IndexSet<u16>) -> Self {
+        SkipByPort(ports.into())
+    }
+}
+
+impl<T> linkerd2_stack::Switch<T> for SkipByPort
+where
+    for<'t> &'t T: Into<std::net::SocketAddr>,
+{
+    fn use_primary(&self, addrs: &T) -> bool {
+        !self.0.contains(&addrs.into().port())
+    }
+}

linkerd/app/core/src/proxy/mod.rs

@@ -9,7 +9,3 @@ pub use linkerd2_proxy_identity as identity;
 pub use linkerd2_proxy_resolve as resolve;
 pub use linkerd2_proxy_tap as tap;
 pub use linkerd2_proxy_tcp as tcp;
-
-mod skip_detect;
-
-pub use self::skip_detect::SkipDetect;

linkerd/app/core/src/proxy/skip_detect.rs

@@ -17,7 +17,7 @@ use linkerd2_app_core::{
     profiles,
     proxy::{
         http::{self, orig_proto, strip_header, DetectHttp},
-        identity, tap, tcp, SkipDetect,
+        identity, tap, tcp,
     },
     reconnect, router, serve,
     spans::SpanConverter,
@@ -392,7 +392,7 @@ impl Config {
             .push_map_target(TcpEndpoint::from)
             .push(metrics.transport.layer_accept(TransportLabels))
             .into_inner();
-        let accept = SkipDetect::new(skip_detect, tls, accept_fwd);
+        let accept = svc::stack::MakeSwitch::new(skip_detect, tls, accept_fwd);
 
         info!(addr = %listen_addr, "Serving");
         serve::serve(listen, accept, drain.signal()).await

linkerd/app/inbound/src/lib.rs

@@ -16,7 +16,6 @@ use linkerd2_app_core::{
     profiles,
     proxy::{
         self, core::resolve::Resolve, discover, http, identity, resolve::map_endpoint, tap, tcp,
-        SkipDetect,
     },
     reconnect, retry, router, serve,
     spans::SpanConverter,
@@ -500,7 +499,7 @@ impl Config {
             .push(admit::AdmitLayer::new(prevent_loop))
             .push_map_target(TcpEndpoint::from);
 
-        let accept = svc::stack(SkipDetect::new(skip_detect, http, tcp_forward))
+        let accept = svc::stack(svc::stack::MakeSwitch::new(skip_detect, http, tcp_forward))
             .push(metrics.transport.layer_accept(TransportLabels));
 
         info!(addr = %listen_addr, "Serving");

linkerd/app/outbound/src/lib.rs

@@ -13,6 +13,7 @@ pub mod new_service;
 pub mod on_response;
 mod oneshot;
 mod proxy;
+mod switch;
 
 pub use self::fallback::{Fallback, FallbackLayer};
 pub use self::future_service::FutureService;
@@ -24,3 +25,4 @@ pub use self::new_service::NewService;
 pub use self::on_response::{OnResponse, OnResponseLayer};
 pub use self::oneshot::{Oneshot, OneshotLayer};
 pub use self::proxy::{Proxy, ProxyService};
+pub use self::switch::{MakeSwitch, Switch};

linkerd/stack/src/lib.rs

@@ -0,0 +1,84 @@
+//! A middleware that switches between two underlying stacks, depending on the
+//! target type.
+
+use futures::{future, prelude::*};
+use linkerd2_error::Error;
+use std::task::{Context, Poll};
+use tower::util::ServiceExt;
+
+/// Determines whether the primary stack should be used.
+pub trait Switch<T> {
+    fn use_primary(&self, target: &T) -> bool;
+}
+
+/// Makes either the primary or fallback stack, as determined by an `S`-typed
+/// `Switch`.
+pub struct MakeSwitch<S, P, F> {
+    switch: S,
+    primary: P,
+    fallback: F,
+}
+
+impl<S, P, F> MakeSwitch<S, P, F> {
+    pub fn new(switch: S, primary: P, fallback: F) -> Self {
+        MakeSwitch {
+            switch,
+            primary,
+            fallback,
+        }
+    }
+
+    pub fn layer(switch: S, fallback: F) -> impl super::layer::Layer<P, Service = Self> + Clone
+    where
+        S: Clone,
+        F: Clone,
+    {
+        super::layer::mk(move |primary| Self::new(switch.clone(), primary, fallback.clone()))
+    }
+}
+
+impl<T, S, P, F> tower::Service<T> for MakeSwitch<S, P, F>
+where
+    S: Switch<T>,
+    P: tower::Service<T> + Clone,
+    P::Error: Into<Error>,
+    F: tower::Service<T> + Clone,
+    F::Error: Into<Error>,
+{
+    type Response = tower::util::Either<P::Response, F::Response>;
+    type Error = Error;
+    type Future = future::Either<
+        future::MapOk<
+            future::ErrInto<tower::util::Oneshot<P, T>, Error>,
+            fn(P::Response) -> tower::util::Either<P::Response, F::Response>,
+        >,
+        future::MapOk<
+            future::ErrInto<tower::util::Oneshot<F, T>, Error>,
+            fn(F::Response) -> tower::util::Either<P::Response, F::Response>,
+        >,
+    >;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, target: T) -> Self::Future {
+        if self.switch.use_primary(&target) {
+            future::Either::Left(
+                self.primary
+                    .clone()
+                    .oneshot(target)
+                    .err_into::<Error>()
+                    .map_ok(tower::util::Either::A),
+            )
+        } else {
+            future::Either::Right(
+                self.fallback
+                    .clone()
+                    .oneshot(target)
+                    .err_into::<Error>()
+                    .map_ok(tower::util::Either::B),
+            )
+        }
+    }
+}