Introduce a SkipDetect layer to preempt detection (#620)

This change introduces a new SkipDetect layer that configures whether
protocol detection should be attempted. This module will later be
replaced/augmented by discovery.

Furthermore, this change eliminates the `Accept` trait. Instead of
modeling the accept stack as a simple service whose response is a
future, we know model the stack as, effectively, a
`MakeService<Meta, TcpStream>`. This is intended to support caching of
the service that handles the tcp stream (i.e. to hold discovery
responses).

Detection timeouts have been moved from a dedicated layer into the
detection modules.

Author: olix0r

Cargo.lock

@@ -826,7 +826,6 @@ dependencies = [
  "linkerd2-proxy-api",
  "linkerd2-proxy-api-resolve",
  "linkerd2-proxy-core",
- "linkerd2-proxy-detect",
  "linkerd2-proxy-discover",
  "linkerd2-proxy-http",
  "linkerd2-proxy-identity",
@@ -1240,19 +1239,6 @@ dependencies = [
  "tracing-futures",
 ]
 
-[[package]]
-name = "linkerd2-proxy-detect"
-version = "0.1.0"
-dependencies = [
- "async-trait",
- "futures 0.3.5",
- "linkerd2-error",
- "linkerd2-io",
- "linkerd2-proxy-core",
- "tokio",
- "tower",
-]
-
 [[package]]
 name = "linkerd2-proxy-discover"
 version = "0.1.0"
@@ -1274,7 +1260,6 @@ dependencies = [
 name = "linkerd2-proxy-http"
 version = "0.1.0"
 dependencies = [
- "async-trait",
  "bytes 0.5.4",
  "futures 0.3.5",
  "h2 0.2.6",
@@ -1291,8 +1276,7 @@ dependencies = [
  "linkerd2-error",
  "linkerd2-http-box",
  "linkerd2-identity",
- "linkerd2-proxy-core",
- "linkerd2-proxy-detect",
+ "linkerd2-io",
  "linkerd2-proxy-transport",
  "linkerd2-stack",
  "linkerd2-timeout",
@@ -1394,7 +1378,6 @@ dependencies = [
  "linkerd2-io",
  "linkerd2-metrics",
  "linkerd2-proxy-core",
- "linkerd2-proxy-detect",
  "linkerd2-stack",
  "pin-project",
  "ring",

Cargo.toml

@@ -32,7 +32,6 @@ members = [
     "linkerd/opencensus",
     "linkerd/proxy/api-resolve",
     "linkerd/proxy/core",
-    "linkerd/proxy/detect",
     "linkerd/proxy/discover",
     "linkerd/proxy/http",
     "linkerd/proxy/identity",
@@ -63,4 +62,4 @@ debug = false
 
 [patch.crates-io]
 webpki = { git = "https://github.com/linkerd/webpki", branch = "cert-dns-names-0.21" }
-tower = { version = "0.3", git = "https://github.com/tower-rs/tower", rev = "8752a3811788e94670c62dc0acbc9613207931b1"}
+tower = { version = "0.3", git = "https://github.com/tower-rs/tower", rev = "8752a3811788e94670c62dc0acbc9613207931b1"}

linkerd/app/core/Cargo.toml

@@ -43,7 +43,6 @@ linkerd2-opencensus = { path = "../../opencensus" }
 linkerd2-proxy-core = { path = "../../proxy/core" }
 linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", tag = "v0.1.13" }
 linkerd2-proxy-api-resolve = { path = "../../proxy/api-resolve" }
-linkerd2-proxy-detect = { path = "../../proxy/detect" }
 linkerd2-proxy-discover = { path = "../../proxy/discover" }
 linkerd2-proxy-identity = { path = "../../proxy/identity" }
 linkerd2-proxy-http = { path = "../../proxy/http" }

linkerd/app/core/src/accept_error.rs

@@ -3,17 +3,21 @@
 //! * `/metrics` -- reports prometheus-formatted metrics.
 //! * `/ready` -- returns 200 when the proxy is ready to participate in meshed traffic.
 
-use crate::{svc, trace, transport::tls::accept::Connection};
+use crate::{
+    svc, trace,
+    transport::{io, tls},
+};
 use futures::{future, TryFutureExt};
 use http::StatusCode;
 use hyper::{Body, Request, Response};
-use linkerd2_error::Error;
+use linkerd2_error::{Error, Never};
 use linkerd2_metrics::{self as metrics, FmtMetrics};
-use std::future::Future;
-use std::io;
-use std::pin::Pin;
-use std::task::{Context, Poll};
-use tower::{service_fn, Service};
+use std::{
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
+use tower::{service_fn, util::ServiceExt, Service};
 
 mod readiness;
 mod tasks;
@@ -33,6 +37,9 @@ pub struct Admin<M: FmtMetrics> {
 #[derive(Debug, Clone)]
 pub struct Accept<M: FmtMetrics>(Admin<M>, hyper::server::conn::Http);
 
+#[derive(Debug, Clone)]
+pub struct Serve<M: FmtMetrics>(tls::accept::Meta, Accept<M>);
+
 #[derive(Clone, Debug)]
 pub struct ClientAddr(std::net::SocketAddr);
 
@@ -103,28 +110,43 @@ impl<M: FmtMetrics> Service<Request<Body>> for Admin<M> {
     }
 }
 
-impl<M: FmtMetrics + Clone + Send + 'static> svc::Service<Connection> for Accept<M> {
-    type Response = Pin<Box<dyn Future<Output = Result<(), Self::Error>> + Send + 'static>>;
-    type Error = Error;
+impl<M: FmtMetrics + Clone + Send + 'static> svc::Service<tls::accept::Meta> for Accept<M> {
+    type Response = Serve<M>;
+    type Error = Never;
     type Future = future::Ready<Result<Self::Response, Self::Error>>;
 
     fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         Poll::Ready(Ok(()))
     }
 
-    fn call(&mut self, (meta, io): Connection) -> Self::Future {
+    fn call(&mut self, meta: tls::accept::Meta) -> Self::Future {
+        future::ok(Serve(meta, self.clone()))
+    }
+}
+
+impl<M: FmtMetrics + Clone + Send + 'static> svc::Service<io::BoxedIo> for Serve<M> {
+    type Response = ();
+    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'static>>;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, io: io::BoxedIo) -> Self::Future {
+        let Self(ref meta, Accept(ref svc, ref server)) = self;
+
         // Since the `/proxy-log-level` controls access based on the
         // client's IP address, we wrap the service with a new service
         // that adds the remote IP as a request extension.
         let peer = meta.addrs.peer();
-        let mut svc = self.0.clone();
+        let svc = svc.clone();
         let svc = service_fn(move |mut req: Request<Body>| {
             req.extensions_mut().insert(ClientAddr(peer));
-            svc.call(req)
+            svc.clone().oneshot(req)
         });
 
-        let connection_future = self.1.serve_connection(io, svc).map_err(Into::into);
-        future::ok(Box::pin(connection_future))
+        Box::pin(server.serve_connection(io, svc).map_err(Into::into))
     }
 }
 

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

@@ -27,7 +27,6 @@ pub use linkerd2_stack_metrics as stack_metrics;
 pub use linkerd2_stack_tracing as stack_tracing;
 pub use linkerd2_trace_context::TraceContextLayer;
 
-pub mod accept_error;
 pub mod admin;
 pub mod classify;
 pub mod config;

linkerd/app/core/src/lib.rs

@@ -2,10 +2,13 @@
 
 pub use linkerd2_proxy_api_resolve as api_resolve;
 pub use linkerd2_proxy_core as core;
-pub use linkerd2_proxy_detect as detect;
 pub use linkerd2_proxy_discover as discover;
 pub use linkerd2_proxy_http as http;
 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/mod.rs

@@ -0,0 +1,103 @@
+use futures::prelude::*;
+use indexmap::IndexSet;
+use linkerd2_error::Error;
+use linkerd2_proxy_transport::listen::Addrs;
+use std::{
+    future::Future,
+    pin::Pin,
+    sync::Arc,
+    task::{Context, Poll},
+};
+use tower::util::ServiceExt;
+
+pub trait SkipTarget<T> {
+    fn skip_target(&self, target: &T) -> bool;
+}
+
+impl SkipTarget<Addrs> for Arc<IndexSet<u16>> {
+    fn skip_target(&self, addrs: &Addrs) -> bool {
+        self.contains(&addrs.target_addr().port())
+    }
+}
+
+pub struct SkipDetect<S, D, F> {
+    skip: S,
+    detect: D,
+    tcp: F,
+}
+
+pub enum Accept<D, F> {
+    Detect(D),
+    Tcp(F),
+}
+
+impl<S, D, F> SkipDetect<S, D, F> {
+    pub fn new(skip: S, detect: D, tcp: F) -> Self {
+        Self { skip, detect, tcp }
+    }
+}
+
+impl<S, D, F> tower::Service<Addrs> for SkipDetect<S, D, F>
+where
+    S: SkipTarget<Addrs>,
+    D: tower::Service<Addrs> + Clone + Send + 'static,
+    D::Error: Into<Error>,
+    D::Future: Send,
+    F: tower::Service<Addrs> + Clone + Send + 'static,
+    F::Error: Into<Error>,
+    F::Future: Send,
+{
+    type Response = Accept<D::Response, F::Response>;
+    type Error = Error;
+    type Future = Pin<
+        Box<dyn Future<Output = Result<Accept<D::Response, F::Response>, Error>> + Send + 'static>,
+    >;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, addrs: Addrs) -> Self::Future {
+        if self.skip.skip_target(&addrs) {
+            let tcp = self.tcp.clone();
+            Box::pin(async move {
+                let f = tcp.oneshot(addrs).err_into::<Error>().await?;
+                Ok(Accept::Tcp(f))
+            })
+        } else {
+            let detect = self.detect.clone();
+            Box::pin(async move {
+                let d = detect.oneshot(addrs).err_into::<Error>().await?;
+                Ok(Accept::Detect(d))
+            })
+        }
+    }
+}
+
+impl<D, F, T> tower::Service<T> for Accept<D, F>
+where
+    D: tower::Service<T, Response = ()>,
+    D::Error: Into<Error>,
+    D::Future: Send + 'static,
+    F: tower::Service<T, Response = ()>,
+    F::Error: Into<Error>,
+    F::Future: Send + 'static,
+{
+    type Response = ();
+    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'static>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
+        Poll::Ready(match self {
+            Self::Detect(d) => futures::ready!(d.poll_ready(cx)).map_err(Into::into),
+            Self::Tcp(f) => futures::ready!(f.poll_ready(cx)).map_err(Into::into),
+        })
+    }
+
+    fn call(&mut self, io: T) -> Self::Future {
+        match self {
+            Self::Detect(d) => Box::pin(d.call(io).err_into::<Error>()),
+            Self::Tcp(f) => Box::pin(f.call(io).err_into::<Error>()),
+        }
+    }
+}