inbound: Simplify protocol-detection skipping (#1031)

The inbound proxy's layer that switches between routed and passthru
stacks uses an unnecessary explict type for its predicate. In order to
setup for further changes to this stack, this change eliminate the
specialized type in favor of a closure.

We also ditch the use of IndexSet as there's no functional need for it.

Author: olix0r

linkerd/app/inbound/src/lib.rs

@@ -28,23 +28,20 @@ use linkerd_app_core::{
     proxy::tcp,
     serve, svc, tls,
     transport::{self, listen::Bind, ClientAddr, Local, OrigDstAddr, Remote, ServerAddr},
-    Error, NameMatch, ProxyRuntime,
+    Error, NameMatch, Never, ProxyRuntime,
 };
-use std::{convert::TryFrom, fmt::Debug, future::Future, time::Duration};
+use std::{collections::HashSet, convert::TryFrom, fmt::Debug, future::Future, time::Duration};
 use tracing::{debug_span, info_span};
 
 #[derive(Clone, Debug)]
 pub struct Config {
     pub allow_discovery: NameMatch,
     pub proxy: ProxyConfig,
     pub require_identity_for_inbound_ports: RequireIdentityForPorts,
-    pub disable_protocol_detection_for_ports: SkipByPort,
+    pub disable_protocol_detection_for_ports: HashSet<u16>,
     pub profile_idle_timeout: Duration,
 }
 
-#[derive(Clone, Debug)]
-pub struct SkipByPort(std::sync::Arc<indexmap::IndexSet<u16>>);
-
 #[derive(Clone, Debug)]
 pub struct Inbound<S> {
     config: Config,
@@ -230,6 +227,7 @@ where
         let disable_detect = self.config.disable_protocol_detection_for_ports.clone();
         let require_id = self.config.require_identity_for_inbound_ports.clone();
         let config = self.config.proxy.clone();
+
         self.clone()
             .push_http_router(profiles)
             .push_http_server()
@@ -262,15 +260,21 @@ where
             ))
             .instrument(|_: &_| debug_span!("proxy"))
             .push_switch(
-                disable_detect,
+                move |t: T| {
+                    let OrigDstAddr(addr) = t.param();
+                    if !disable_detect.contains(&addr.port()) {
+                        Ok::<_, Never>(svc::Either::A(t))
+                    } else {
+                        Ok(svc::Either::B(TcpAccept::port_skipped(t)))
+                    }
+                },
                 self.clone()
                     .push_tcp_forward(server_port)
                     .stack
                     .push_map_target(TcpEndpoint::from)
                     .push(self.runtime.metrics.transport.layer_accept())
-                    .push_map_target(TcpAccept::port_skipped)
-                    .check_new_service::<T, _>()
-                    .instrument(|_: &T| debug_span!("forward"))
+                    .check_new_service::<TcpAccept, _>()
+                    .instrument(|_: &TcpAccept| debug_span!("forward"))
                     .into_inner(),
             )
             .check_new_service::<T, I>()
@@ -294,30 +298,6 @@ where
     }
 }
 
-// === impl SkipByPort ===
-
-impl From<indexmap::IndexSet<u16>> for SkipByPort {
-    fn from(ports: indexmap::IndexSet<u16>) -> Self {
-        SkipByPort(ports.into())
-    }
-}
-
-impl<T> svc::Predicate<T> for SkipByPort
-where
-    T: svc::Param<OrigDstAddr>,
-{
-    type Request = svc::Either<T, T>;
-
-    fn check(&mut self, t: T) -> Result<Self::Request, Error> {
-        let OrigDstAddr(addr) = t.param();
-        if !self.0.contains(&addr.port()) {
-            Ok(svc::Either::A(t))
-        } else {
-            Ok(svc::Either::B(t))
-        }
-    }
-}
-
 fn stack_labels(proto: &'static str, name: &'static str) -> metrics::StackLabels {
     metrics::StackLabels::inbound(proto, name)
 }

linkerd/app/inbound/src/test_util.rs

@@ -1,4 +1,4 @@
-use crate::{Config, RequireIdentityForPorts, SkipByPort};
+use crate::{Config, RequireIdentityForPorts};
 pub use futures::prelude::*;
 use linkerd_app_core::{
     config,
@@ -48,7 +48,7 @@ pub fn default_config() -> Config {
             detect_protocol_timeout: Duration::from_secs(10),
         },
         require_identity_for_inbound_ports: RequireIdentityForPorts::from(None),
-        disable_protocol_detection_for_ports: SkipByPort::from(indexmap::IndexSet::default()),
+        disable_protocol_detection_for_ports: Default::default(),
         profile_idle_timeout: Duration::from_millis(500),
     }
 }

linkerd/app/src/env.rs

@@ -528,7 +528,7 @@ pub fn parse_config<S: Strings>(strings: &S) -> Result<super::Config, EnvError>
             require_identity_for_inbound_ports: require_identity_for_inbound_ports.into(),
             profile_idle_timeout: dst_profile_idle_timeout?
                 .unwrap_or(DEFAULT_DESTINATION_PROFILE_IDLE_TIMEOUT),
-            disable_protocol_detection_for_ports: inbound_opaque_ports.into(),
+            disable_protocol_detection_for_ports: inbound_opaque_ports.into_iter().collect(),
         }
     };