Split the inbound server into multiple modules (#1179)

The `Inbound::push_server` stack builder is responsible for building all
of the inbound stacks: the HTTP stack, the TCP forwarding stack, and the
direct/gateway stack. Furthermore, most of these stacks rely on fixed
target types (which makes them inflexible).

This change refactors the inbound stack to setup for upcoming policy
changes:

1. A 'detect' stack is split out into a `push_detect` helper that only
   deals with protocol detection and dispatching connections to either
   the TCP forwarding stack or the HTTP handling stack.
2. An 'accept' stack is split out into a `push_accept` helper that is
   responsible for extracting relevant connection metadata (client and
   orig dst addrs) and determining the policy to use for the connection.
3. Each stack/module defines its own target types and other modules rely
   on param constraints instead of fixed target types.
4. A `port_policies` module is introduced to model the existing per-port
   policies: whether identity is required and whether protocol detection
   should be skipped. Various auxiliary modules have been eliminated in
   favor of inlined target filters. This helps to make these stacks
   easier to reason about, as relevant logic isn't spread over multiple
   files.

Author: olix0r

linkerd/app/admin/src/server/mod.rs

@@ -19,7 +19,7 @@ use hyper::{
 use linkerd_app_core::{
     metrics::{self as metrics, FmtMetrics},
     proxy::http::ClientHandle,
-    svc, trace, Error,
+    trace, Error,
 };
 use std::{
     future::Future,
@@ -150,13 +150,6 @@ impl<M> Admin<M> {
     }
 }
 
-impl<M: FmtMetrics + Clone, T> svc::NewService<T> for Admin<M> {
-    type Service = Self;
-    fn new_service(&mut self, _: T) -> Self::Service {
-        self.clone()
-    }
-}
-
 impl<M, B> tower::Service<http::Request<B>> for Admin<M>
 where
     M: FmtMetrics,

linkerd/app/admin/src/stack.rs

@@ -7,10 +7,9 @@ use linkerd_app_core::{
     serve,
     svc::{self, ExtractParam, InsertParam, Param},
     tls, trace,
-    transport::{listen::Bind, ClientAddr, Local, Remote, ServerAddr},
+    transport::{self, listen::Bind, ClientAddr, Local, Remote, ServerAddr},
     Error,
 };
-use linkerd_app_inbound::target::{HttpAccept, Target, TcpAccept};
 use std::{pin::Pin, time::Duration};
 use thiserror::Error;
 use tokio::sync::mpsc;
@@ -36,6 +35,19 @@ struct NonHttpClient(Remote<ClientAddr>);
 #[error("Unexpected TLS connection to {} from {}", self.0, self.1)]
 struct UnexpectedSni(tls::ServerId, Remote<ClientAddr>);
 
+#[derive(Clone, Debug)]
+struct Tcp {
+    addr: Local<ServerAddr>,
+    client: Remote<ClientAddr>,
+    tls: tls::ConditionalServerTls,
+}
+
+#[derive(Clone, Debug)]
+struct Http {
+    tcp: Tcp,
+    version: http::Version,
+}
+
 #[derive(Clone)]
 struct TlsParams {
     identity: Option<LocalCrtKey>,
@@ -66,31 +78,29 @@ impl Config {
 
         let (ready, latch) = crate::server::Readiness::new();
         let admin = crate::server::Admin::new(report, ready, shutdown, trace);
-        let admin = svc::stack(admin)
-            .push(metrics.http_endpoint.to_layer::<classify::Response, _, Target>())
+        let admin = svc::stack(move |_| admin.clone())
+            .push(metrics.http_endpoint.to_layer::<classify::Response, _, Http>())
             .push_on_response(
                 svc::layers()
                     .push(metrics.http_errors.clone())
                     .push(errors::layer())
                     .push(http::BoxResponse::layer()),
             )
-            .push_map_target(Target::from)
             .push(http::NewServeHttp::layer(Default::default(), drain.clone()))
             .push_request_filter(
-                |(version, tcp): (
+                |(http, tcp): (
                     Result<Option<http::Version>, detect::DetectTimeoutError<_>>,
-                    TcpAccept,
+                    Tcp,
                 )| {
-                    match version {
-                        Ok(Some(version)) => Ok(HttpAccept::from((version, tcp))),
-                        // If detection timed out, we can make an educated guess
-                        // at the proper behavior:
-                        // - If the connection was meshed, it was most likely
-                        //   transported over HTTP/2.
-                        // - If the connection was unmehsed, it was mostly
-                        //   likely HTTP/1.
-                        // - If we received some unexpected SNI, the client is
-                        //   mostly likely confused/stale.
+                    match http {
+                        Ok(Some(version)) => Ok(Http { version, tcp }),
+                        // If detection timed out, we can make an educated guess at the proper
+                        // behavior:
+                        // - If the connection was meshed, it was most likely transported over
+                        //   HTTP/2.
+                        // - If the connection was unmeshed, it was mostly likely HTTP/1.
+                        // - If we received some unexpected SNI, the client is mostly likely
+                        //   confused/stale.
                         Err(_timeout) => {
                             let version = match tcp.tls.clone() {
                                 tls::ConditionalServerTls::None(_) => http::Version::Http1,
@@ -101,20 +111,19 @@ impl Config {
                                     sni,
                                 }) => {
                                     debug_assert!(false, "If we know the stream is non-mesh TLS, we should be able to prove its not HTTP.");
-                                    return Err(Error::from(UnexpectedSni(sni, tcp.client_addr)));
+                                    return Err(Error::from(UnexpectedSni(sni, tcp.client)));
                                 }
                             };
                             debug!(%version, "HTTP detection timed out; assuming HTTP");
-                            Ok(HttpAccept::from((version, tcp)))
+                            Ok(Http { version, tcp })
                         }
-                        // If the connection failed HTTP detection, check if we
-                        // detected TLS for another target. This might indicate
-                        // that the client is confused/stale.
+                        // If the connection failed HTTP detection, check if we detected TLS for
+                        // another target. This might indicate that the client is confused/stale.
                         Ok(None) => match tcp.tls {
                             tls::ConditionalServerTls::Some(tls::ServerTls::Passthru { sni }) => {
-                                Err(UnexpectedSni(sni, tcp.client_addr).into())
+                                Err(UnexpectedSni(sni, tcp.client).into())
                             }
-                            _ => Err(NonHttpClient(tcp.client_addr).into()),
+                            _ => Err(NonHttpClient(tcp.client).into()),
                         },
                     }
                 },
@@ -123,12 +132,10 @@ impl Config {
             .push(detect::NewDetectService::layer(detect::Config::<http::DetectHttp>::from_timeout(DETECT_TIMEOUT)))
             .push(metrics.transport.layer_accept())
             .push_map_target(|(tls, addrs): (tls::ConditionalServerTls, B::Addrs)| {
-                // TODO this should use an admin-specific target type.
-                let Local(ServerAddr(target_addr)) = addrs.param();
-                TcpAccept {
+                Tcp {
                     tls,
-                    client_addr: addrs.param(),
-                    target_addr,
+                    client: addrs.param(),
+                    addr: addrs.param(),
                 }
             })
             .push(svc::BoxNewService::layer())
@@ -146,6 +153,37 @@ impl Config {
     }
 }
 
+// === impl Tcp ===
+
+impl Param<transport::labels::Key> for Tcp {
+    fn param(&self) -> transport::labels::Key {
+        transport::labels::Key::Accept {
+            direction: transport::labels::Direction::In,
+            tls: self.tls.clone(),
+            target_addr: self.addr.into(),
+        }
+    }
+}
+
+// === impl Http ===
+
+impl Param<http::Version> for Http {
+    fn param(&self) -> http::Version {
+        self.version
+    }
+}
+
+impl Param<metrics::EndpointLabels> for Http {
+    fn param(&self) -> metrics::EndpointLabels {
+        metrics::InboundEndpointLabels {
+            tls: self.tcp.tls.clone(),
+            authority: None,
+            target_addr: self.tcp.addr.into(),
+        }
+        .into()
+    }
+}
+
 // === TlsParams ===
 
 impl<T> ExtractParam<tls::server::Timeout, T> for TlsParams {

linkerd/app/core/src/config.rs

@@ -4,11 +4,7 @@ use crate::{
     svc::Param,
     transport::{Keepalive, ListenAddr},
 };
-use std::{
-    collections::HashSet,
-    hash::{BuildHasherDefault, Hasher},
-    time::Duration,
-};
+use std::time::Duration;
 
 #[derive(Clone, Debug)]
 pub struct ServerConfig {
@@ -37,19 +33,6 @@ pub struct ProxyConfig {
     pub detect_protocol_timeout: Duration,
 }
 
-/// A `HashSet` specialized for ports.
-///
-/// Because ports are `u16` values, this type avoids the overhead of actually
-/// hashing ports.
-pub type PortSet = HashSet<u16, BuildHasherDefault<PortHasher>>;
-
-/// A hasher for ports.
-///
-/// Because ports are single `u16` values, we don't have to hash them; we can just use
-/// the integer values as hashes directly.
-#[derive(Default)]
-pub struct PortHasher(u16);
-
 // === impl ProxyConfig ===
 
 impl ProxyConfig {
@@ -71,43 +54,3 @@ impl Param<Keepalive> for ServerConfig {
         self.keepalive
     }
 }
-
-// === impl PortHasher ===
-
-impl Hasher for PortHasher {
-    fn write(&mut self, _: &[u8]) {
-        unreachable!("hashing a `u16` calls `write_u16`");
-    }
-
-    #[inline]
-    fn write_u16(&mut self, port: u16) {
-        self.0 = port;
-    }
-
-    #[inline]
-    fn finish(&self) -> u64 {
-        self.0 as u64
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use quickcheck::*;
-
-    quickcheck! {
-        fn portset_contains_all_ports(ports: Vec<u16>) -> bool {
-            // Make a port set containing the generated port numbers.
-            let portset = ports.iter().cloned().collect::<PortSet>();
-            for port in ports {
-                // If the port set doesn't contain one of the ports it was
-                // created with, that's bad news!
-                if !portset.contains(&port) {
-                    return false;
-                }
-            }
-
-            true
-        }
-    }
-}

linkerd/app/core/src/dst.rs

@@ -1,14 +1,12 @@
 use super::classify;
 use crate::profiles;
-use linkerd_addr::Addr;
 use linkerd_http_classify::CanClassify;
 use linkerd_proxy_http::timeout;
-use std::fmt;
 use std::time::Duration;
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct Route {
-    pub target: Addr,
+    pub addr: profiles::LogicalAddr,
     pub route: profiles::http::Route,
     pub direction: super::metrics::Direction,
 }
@@ -28,9 +26,3 @@ impl timeout::HasTimeout for Route {
         self.route.timeout()
     }
 }
-
-impl fmt::Display for Route {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        self.target.fmt(f)
-    }
-}

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

@@ -2,7 +2,8 @@ mod tcp_accept_errors;
 
 use crate::{
     classify::{Class, SuccessOrFailure},
-    control, dst, errors, http_metrics, http_metrics as metrics, opencensus, stack_metrics,
+    control, dst, errors, http_metrics, http_metrics as metrics, opencensus, profiles,
+    stack_metrics,
     svc::Param,
     telemetry, tls,
     transport::{
@@ -86,7 +87,7 @@ pub struct StackLabels {
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct RouteLabels {
     direction: Direction,
-    target: Addr,
+    addr: profiles::LogicalAddr,
     labels: Option<String>,
 }
 
@@ -232,7 +233,7 @@ impl FmtLabels for ControlLabels {
 impl Param<RouteLabels> for dst::Route {
     fn param(&self) -> RouteLabels {
         RouteLabels {
-            target: self.target.clone(),
+            addr: self.addr.clone(),
             direction: self.direction,
             labels: prefix_labels("rt", self.route.labels().iter()),
         }
@@ -242,7 +243,7 @@ impl Param<RouteLabels> for dst::Route {
 impl FmtLabels for RouteLabels {
     fn fmt_labels(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         self.direction.fmt_labels(f)?;
-        write!(f, ",dst=\"{}\"", self.target)?;
+        write!(f, ",dst=\"{}\"", self.addr)?;
 
         if let Some(labels) = self.labels.as_ref() {
             write!(f, ",{}", labels)?;

linkerd/app/inbound/fuzz/fuzz_targets/fuzz_target_1.rs

@@ -1,10 +1,10 @@
 #![no_main]
 
 #[cfg(fuzzing)]
-use {libfuzzer_sys::fuzz_target, linkerd_app_inbound::http::fuzz_logic::*};
+use {libfuzzer_sys::fuzz_target, linkerd_app_inbound::http_fuzz};
 
 #[cfg(fuzzing)]
-fuzz_target!(|requests: Vec<HttpRequestSpec>| {
+fuzz_target!(|requests: Vec<http_fuzz::HttpRequestSpec>| {
     // Don't enable tracing in `cluster-fuzz`, since we would emit verbose
     // traces for *every* generated fuzz input...
     let _trace = linkerd_tracing::test::with_default_filter("off");
@@ -15,5 +15,5 @@ fuzz_target!(|requests: Vec<HttpRequestSpec>| {
 
     tokio::runtime::Runtime::new()
         .unwrap()
-        .block_on(fuzz_entry_raw(requests));
+        .block_on(http_fuzz::fuzz_entry_raw(requests));
 });