Move tls::accept to async/await (#607)

olix0r · web-flow · commit eee08f995cd8 · 2020-07-27T17:31:08.000-07:00
The `tls::accept` module can be simplified by moving to async/await.
This presents a few opportunities for improvement:

1. The recently-introduced `Detect` trait is used. This sets up further
   changes to the accept logic, decoupling protocol detection from the
   `Accept` stack/type.
2. The `tls::ReasonForNoIdentity` type has been removed, as it provided
   no value; and the `tls::ReasonForNoPeerName` has been updated with
   additional states.
3. We now avoid having to use `PrefixedIo` in most cases by using
   `TcpStream::peek` with a smaller buffer. The prior behavior is
   retained as a fallback in case peek doesn't return enough data.

This comes at the cost of ~2 additional Boxes per connection; but these
will probably be shaved off in a follow-up as the detection logic
becomes more flexible/dynamic.
diff --git a/Cargo.lock b/Cargo.lock
@@ -1353,6 +1353,7 @@ name = "linkerd2-proxy-transport"
 version = "0.1.0"
 dependencies = [
  "async-stream",
+ "async-trait",
  "bytes 0.5.4",
  "futures 0.3.5",
  "indexmap",
@@ -1365,6 +1366,7 @@ dependencies = [
  "linkerd2-io",
  "linkerd2-metrics",
  "linkerd2-proxy-core",
+ "linkerd2-proxy-detect",
  "linkerd2-stack",
  "pin-project",
  "ring",
diff --git a/linkerd/app/core/src/metric_labels.rs b/linkerd/app/core/src/metric_labels.rs
@@ -16,7 +16,7 @@ pub struct ControlLabels {
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct EndpointLabels {
     pub direction: Direction,
-    pub tls_id: Conditional<TlsId, tls::ReasonForNoIdentity>,
+    pub tls_id: Conditional<TlsId, tls::ReasonForNoPeerName>,
     pub authority: Option<http::uri::Authority>,
     pub labels: Option<String>,
 }
diff --git a/linkerd/app/core/src/proxy/server.rs b/linkerd/app/core/src/proxy/server.rs
@@ -54,8 +54,9 @@ impl ProtocolDetect {
 }
 
 #[async_trait]
-impl detect::Detect<tls::accept::Meta> for ProtocolDetect {
+impl detect::Detect<tls::accept::Meta, BoxedIo> for ProtocolDetect {
     type Target = Protocol;
+    type Io = BoxedIo;
     type Error = io::Error;
 
     async fn detect(
diff --git a/linkerd/app/core/src/transport/labels.rs b/linkerd/app/core/src/transport/labels.rs
@@ -88,12 +88,6 @@ impl Into<tls::Conditional<()>> for TlsStatus {
     }
 }
 
-impl TlsStatus {
-    pub fn no_tls_reason(&self) -> Option<tls::ReasonForNoIdentity> {
-        self.0.reason()
-    }
-}
-
 impl fmt::Display for TlsStatus {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self.0 {
@@ -105,7 +99,10 @@ impl fmt::Display for TlsStatus {
 
 impl FmtLabels for TlsStatus {
     fn fmt_labels(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        if let Some(tls::ReasonForNoIdentity::NoPeerName(why)) = self.no_tls_reason() {
+        if let Self(Conditional::None(tls::ReasonForNoPeerName::LocalIdentityDisabled)) = self {
+            return write!(f, "tls=\"disabled\"");
+        }
+        if let Self(Conditional::None(why)) = self {
             return write!(f, "tls=\"no_identity\",no_tls_reason=\"{}\"", why);
         }
 
diff --git a/linkerd/app/gateway/src/lib.rs b/linkerd/app/gateway/src/lib.rs
@@ -60,7 +60,7 @@ mod test {
 
     #[tokio::test]
     async fn no_identity() {
-        let peer_id = tls::PeerIdentity::None(tls::ReasonForNoPeerName::NotProvidedByRemote.into());
+        let peer_id = tls::PeerIdentity::None(tls::ReasonForNoPeerName::NoPeerIdFromRemote);
         let test = Test {
             peer_id,
             ..Default::default()
diff --git a/linkerd/app/inbound/src/endpoint.rs b/linkerd/app/inbound/src/endpoint.rs
@@ -189,11 +189,11 @@ impl tap::Inspect for Target {
     fn src_tls<'a, B>(
         &self,
         req: &'a http::Request<B>,
-    ) -> Conditional<&'a identity::Name, tls::ReasonForNoIdentity> {
+    ) -> Conditional<&'a identity::Name, tls::ReasonForNoPeerName> {
         req.extensions()
             .get::<tls::accept::Meta>()
             .map(|s| s.peer_identity.as_ref())
-            .unwrap_or_else(|| Conditional::None(tls::ReasonForNoIdentity::Disabled))
+            .unwrap_or_else(|| Conditional::None(tls::ReasonForNoPeerName::LocalIdentityDisabled))
     }
 
     fn dst_addr<B>(&self, _: &http::Request<B>) -> Option<SocketAddr> {
@@ -207,7 +207,7 @@ impl tap::Inspect for Target {
     fn dst_tls<B>(
         &self,
         _: &http::Request<B>,
-    ) -> Conditional<&identity::Name, tls::ReasonForNoIdentity> {
+    ) -> Conditional<&identity::Name, tls::ReasonForNoPeerName> {
         Conditional::None(tls::ReasonForNoPeerName::Loopback.into())
     }
 
diff --git a/linkerd/app/inbound/src/lib.rs b/linkerd/app/inbound/src/lib.rs
@@ -17,7 +17,7 @@ use linkerd2_app_core::{
     opencensus::proto::trace::v1 as oc,
     profiles,
     proxy::{
-        detect::DetectProtocolLayer,
+        detect,
         http::{self, normalize_uri, orig_proto, strip_header},
         identity,
         server::{Protocol as ServerProtocol, ProtocolDetect, Server},
@@ -412,15 +412,15 @@ impl Config {
         );
 
         let tcp_detect = svc::stack(tcp_server)
-            .push(DetectProtocolLayer::new(ProtocolDetect::new(
+            .push(detect::AcceptLayer::new(ProtocolDetect::new(
                 disable_protocol_detection_for_ports.clone(),
             )))
             .push(admit::AdmitLayer::new(require_identity_for_inbound_ports))
             // Terminates inbound mTLS from other outbound proxies.
-            .push(tls::AcceptTls::layer(
+            .push(detect::AcceptLayer::new(tls::DetectTls::new(
                 local_identity,
                 disable_protocol_detection_for_ports,
-            ))
+            )))
             // Limits the amount of time that the TCP server spends waiting for TLS handshake &
             // protocol detection. Ensures that connections that never emit data are dropped
             // eventually.
diff --git a/linkerd/app/outbound/src/endpoint.rs b/linkerd/app/outbound/src/endpoint.rs
@@ -128,7 +128,7 @@ impl<T: tap::Inspect> tap::Inspect for Target<T> {
     fn src_tls<'a, B>(
         &self,
         req: &'a http::Request<B>,
-    ) -> Conditional<&'a identity::Name, tls::ReasonForNoIdentity> {
+    ) -> Conditional<&'a identity::Name, tls::ReasonForNoPeerName> {
         self.inner.src_tls(req)
     }
 
@@ -143,7 +143,7 @@ impl<T: tap::Inspect> tap::Inspect for Target<T> {
     fn dst_tls<B>(
         &self,
         req: &http::Request<B>,
-    ) -> Conditional<&identity::Name, tls::ReasonForNoIdentity> {
+    ) -> Conditional<&identity::Name, tls::ReasonForNoPeerName> {
         self.inner.dst_tls(req)
     }
 
@@ -236,7 +236,7 @@ impl tap::Inspect for HttpEndpoint {
     fn src_tls<'a, B>(
         &self,
         _: &'a http::Request<B>,
-    ) -> Conditional<&'a identity::Name, tls::ReasonForNoIdentity> {
+    ) -> Conditional<&'a identity::Name, tls::ReasonForNoPeerName> {
         Conditional::None(tls::ReasonForNoPeerName::Loopback.into())
     }
 
@@ -251,7 +251,7 @@ impl tap::Inspect for HttpEndpoint {
     fn dst_tls<B>(
         &self,
         _: &http::Request<B>,
-    ) -> Conditional<&identity::Name, tls::ReasonForNoIdentity> {
+    ) -> Conditional<&identity::Name, tls::ReasonForNoPeerName> {
         self.identity.as_ref()
     }
 
diff --git a/linkerd/app/outbound/src/lib.rs b/linkerd/app/outbound/src/lib.rs
@@ -18,8 +18,8 @@ use linkerd2_app_core::{
     opencensus::proto::trace::v1 as oc,
     profiles,
     proxy::{
-        self, core::resolve::Resolve, detect::DetectProtocolLayer, discover, http, identity,
-        resolve::map_endpoint, server::ProtocolDetect, tap, tcp, Server,
+        self, core::resolve::Resolve, detect, discover, http, identity, resolve::map_endpoint,
+        server::ProtocolDetect, tap, tcp, Server,
     },
     reconnect, retry, router, serve,
     spans::SpanConverter,
@@ -539,18 +539,18 @@ impl Config {
         );
 
         let no_tls: tls::Conditional<identity::Local> =
-            Conditional::None(tls::ReasonForNoPeerName::Loopback.into());
+            Conditional::None(tls::ReasonForNoPeerName::Loopback);
 
         let tcp_detect = svc::stack(tcp_server)
-            .push(DetectProtocolLayer::new(ProtocolDetect::new(
+            .push(detect::AcceptLayer::new(ProtocolDetect::new(
                 disable_protocol_detection_for_ports.clone(),
             )))
             // The local application never establishes mTLS with the proxy, so don't try to
             // terminate TLS, just annotate with the connection with the reason.
-            .push(tls::AcceptTls::layer(
+            .push(detect::AcceptLayer::new(tls::DetectTls::new(
                 no_tls,
                 disable_protocol_detection_for_ports,
-            ))
+            )))
             // Limits the amount of time that the TCP server spends waiting for protocol
             // detection. Ensures that connections that never emit data are dropped eventually.
             .push_timeout(detect_protocol_timeout);
diff --git a/linkerd/app/src/admin.rs b/linkerd/app/src/admin.rs
@@ -1,7 +1,7 @@
 use crate::identity::LocalIdentity;
 use linkerd2_app_core::{
-    admin, config::ServerConfig, drain, metrics::FmtMetrics, serve, trace::LevelHandle,
-    transport::tls, Error,
+    admin, config::ServerConfig, drain, metrics::FmtMetrics, proxy::detect, serve,
+    trace::LevelHandle, transport::tls, Error,
 };
 use std::net::SocketAddr;
 use std::pin::Pin;
@@ -34,7 +34,10 @@ impl Config {
 
         let (ready, latch) = admin::Readiness::new();
         let admin = admin::Admin::new(report, ready, log_level);
-        let accept = tls::AcceptTls::new(identity, admin.into_accept());
+        let accept = detect::Accept::new(
+            tls::DetectTls::new(identity, Default::default()),
+            admin.into_accept(),
+        );
         let serve = Box::pin(serve::serve(listen, accept, drain.signal()));
         Ok(Admin {
             listen_addr,
diff --git a/linkerd/app/src/env.rs b/linkerd/app/src/env.rs
@@ -890,7 +890,7 @@ pub fn parse_control_addr_disable_identity<S: Strings>(
     base: &str,
 ) -> Result<Option<ControlAddr>, EnvError> {
     let a = parse(strings, &format!("{}_ADDR", base), parse_addr)?;
-    let identity = tls::Conditional::None(tls::ReasonForNoIdentity::Disabled);
+    let identity = tls::Conditional::None(tls::ReasonForNoPeerName::LocalIdentityDisabled);
     Ok(a.map(|addr| ControlAddr { addr, identity }))
 }
 
diff --git a/linkerd/app/src/identity.rs b/linkerd/app/src/identity.rs
@@ -79,7 +79,9 @@ impl Config {
 impl Identity {
     pub fn local(&self) -> LocalIdentity {
         match self {
-            Identity::Disabled => tls::Conditional::None(tls::ReasonForNoIdentity::Disabled),
+            Identity::Disabled => {
+                tls::Conditional::None(tls::ReasonForNoPeerName::LocalIdentityDisabled)
+            }
             Identity::Enabled { ref local, .. } => tls::Conditional::Some(local.clone()),
         }
     }
diff --git a/linkerd/app/src/tap.rs b/linkerd/app/src/tap.rs
@@ -2,7 +2,7 @@ use indexmap::IndexSet;
 use linkerd2_app_core::{
     config::ServerConfig,
     drain,
-    proxy::{identity, tap},
+    proxy::{detect, identity, tap},
     serve,
     transport::tls,
     Error,
@@ -49,8 +49,8 @@ impl Config {
             } => {
                 let (listen_addr, listen) = config.bind.bind()?;
 
-                let accept = tls::AcceptTls::new(
-                    identity,
+                let accept = detect::Accept::new(
+                    tls::DetectTls::new(identity, Default::default()),
                     tap::AcceptPermittedClients::new(permitted_peer_identities.into(), server),
                 );
 
diff --git a/linkerd/proxy/detect/src/lib.rs b/linkerd/proxy/detect/src/lib.rs
@@ -1,54 +1,61 @@
 use async_trait::async_trait;
 use futures::prelude::*;
 use linkerd2_error::Error;
-use linkerd2_io::BoxedIo;
+use linkerd2_io::{AsyncRead, AsyncWrite, BoxedIo};
 use linkerd2_proxy_core as core;
-use std::future::Future;
-use std::pin::Pin;
-use std::task::{Context, Poll};
+use std::{
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
 use tower::util::ServiceExt;
 
 /// A strategy for detecting values out of a client transport.
 #[async_trait]
-pub trait Detect<T> {
+pub trait Detect<T, I: AsyncRead + AsyncWrite> {
     type Target;
+    type Io: AsyncRead + AsyncWrite + Send + Unpin;
     type Error: Into<Error>;
 
-    async fn detect(&self, target: T, io: BoxedIo) -> Result<(Self::Target, BoxedIo), Self::Error>;
+    async fn detect(&self, target: T, io: I) -> Result<(Self::Target, Self::Io), Self::Error>;
 }
 
 #[derive(Debug, Clone)]
-pub struct DetectProtocolLayer<D> {
+pub struct AcceptLayer<D> {
     detect: D,
 }
 
 #[derive(Debug, Clone)]
-pub struct DetectProtocol<D, A> {
+pub struct Accept<D, A> {
     detect: D,
     accept: A,
 }
 
-impl<D> DetectProtocolLayer<D> {
+impl<D> AcceptLayer<D> {
     pub fn new(detect: D) -> Self {
         Self { detect }
     }
 }
 
-impl<D: Clone, A> tower::layer::Layer<A> for DetectProtocolLayer<D> {
-    type Service = DetectProtocol<D, A>;
+impl<D: Clone, A> tower::layer::Layer<A> for AcceptLayer<D> {
+    type Service = Accept<D, A>;
 
     fn layer(&self, accept: A) -> Self::Service {
-        Self::Service {
-            detect: self.detect.clone(),
-            accept,
-        }
+        Self::Service::new(self.detect.clone(), accept)
     }
 }
 
-impl<T, D, A> tower::Service<(T, BoxedIo)> for DetectProtocol<D, A>
+impl<D: Clone, A> Accept<D, A> {
+    pub fn new(detect: D, accept: A) -> Self {
+        Self { detect, accept }
+    }
+}
+
+impl<T, I, D, A> tower::Service<(T, I)> for Accept<D, A>
 where
     T: Send + 'static,
-    D: Detect<T> + Clone + Send + 'static,
+    I: AsyncRead + AsyncWrite + Send + 'static,
+    D: Detect<T, I, Io = BoxedIo> + Clone + Send + 'static,
     D::Target: Send,
     A: core::Accept<(D::Target, BoxedIo)> + Send + Clone + 'static,
     A::Future: Send,
@@ -62,7 +69,7 @@ where
         Poll::Ready(Ok(()))
     }
 
-    fn call(&mut self, (target, io): (T, BoxedIo)) -> Self::Future {
+    fn call(&mut self, (target, io): (T, I)) -> Self::Future {
         let detect = self.detect.clone();
         let mut accept = self.accept.clone().into_service();
         Box::pin(async move {
diff --git a/linkerd/proxy/tap/src/accept.rs b/linkerd/proxy/tap/src/accept.rs
@@ -6,9 +6,7 @@ use linkerd2_identity as identity;
 use linkerd2_proxy_api::tap::tap_server::{Tap, TapServer};
 use linkerd2_proxy_http::{trace, HyperServerSvc};
 use linkerd2_proxy_transport::io::BoxedIo;
-use linkerd2_proxy_transport::tls::{
-    accept::Connection, Conditional, ReasonForNoIdentity, ReasonForNoPeerName,
-};
+use linkerd2_proxy_transport::tls::{accept::Connection, Conditional, ReasonForNoPeerName};
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
@@ -74,7 +72,7 @@ impl Service<Connection> for AcceptPermittedClients {
                     Box::pin(self.serve_unauthenticated(io, format!("Unauthorized peer: {}", peer)))
                 }
             }
-            Conditional::None(ReasonForNoIdentity::NoPeerName(ReasonForNoPeerName::Loopback)) => {
+            Conditional::None(ReasonForNoPeerName::Loopback) => {
                 Box::pin(self.serve_authenticated(io))
             }
             Conditional::None(reason) => {
diff --git a/linkerd/proxy/tap/src/lib.rs b/linkerd/proxy/tap/src/lib.rs
@@ -3,7 +3,7 @@ use http;
 use indexmap::IndexMap;
 use linkerd2_conditional::Conditional;
 use linkerd2_identity as identity;
-use linkerd2_proxy_transport::tls::ReasonForNoIdentity;
+use linkerd2_proxy_transport::tls::ReasonForNoPeerName;
 use std::net;
 use std::sync::Arc;
 
@@ -40,7 +40,7 @@ pub trait Inspect {
     fn src_tls<'a, B>(
         &self,
         req: &'a http::Request<B>,
-    ) -> Conditional<&'a identity::Name, ReasonForNoIdentity>;
+    ) -> Conditional<&'a identity::Name, ReasonForNoPeerName>;
 
     fn dst_addr<B>(&self, req: &http::Request<B>) -> Option<net::SocketAddr>;
 
@@ -49,7 +49,7 @@ pub trait Inspect {
     fn dst_tls<B>(
         &self,
         req: &http::Request<B>,
-    ) -> Conditional<&identity::Name, ReasonForNoIdentity>;
+    ) -> Conditional<&identity::Name, ReasonForNoPeerName>;
 
     fn route_labels<B>(&self, req: &http::Request<B>) -> Option<Arc<IndexMap<String, String>>>;
 
diff --git a/linkerd/proxy/transport/Cargo.toml b/linkerd/proxy/transport/Cargo.toml
diff --git a/linkerd/proxy/transport/src/tls/accept.rs b/linkerd/proxy/transport/src/tls/accept.rs
diff --git a/linkerd/proxy/transport/src/tls/mod.rs b/linkerd/proxy/transport/src/tls/mod.rs
diff --git a/linkerd/proxy/transport/tests/tls_accept.rs b/linkerd/proxy/transport/tests/tls_accept.rs

Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,7 @@ pub struct ControlLabels {`
`16`	`16`	`#[derive(Clone, Debug, PartialEq, Eq, Hash)]`
`17`	`17`	`pub struct EndpointLabels {`
`18`	`18`	`pub direction: Direction,`
`19`		`- pub tls_id: Conditional<TlsId, tls::ReasonForNoIdentity>,`
	`19`	`+ pub tls_id: Conditional<TlsId, tls::ReasonForNoPeerName>,`
`20`	`20`	`pub authority: Option<http::uri::Authority>,`
`21`	`21`	`pub labels: Option<String>,`
`22`	`22`	`}`
Original file line number	Diff line number	Diff line change
`@@ -54,8 +54,9 @@ impl ProtocolDetect {`
`54`	`54`	`}`
`55`	`55`
`56`	`56`	`#[async_trait]`
`57`		`-impl detect::Detect<tls::accept::Meta> for ProtocolDetect {`
	`57`	`+impl detect::Detect<tls::accept::Meta, BoxedIo> for ProtocolDetect {`
`58`	`58`	`type Target = Protocol;`
	`59`	`+ type Io = BoxedIo;`
`59`	`60`	`type Error = io::Error;`
`60`	`61`
`61`	`62`	`async fn detect(`