Split prefix buffering from HTTP detection (#676)

This change modifies `serve` to take a `NewService` instead of a
`MakeService`. Services specific to the accept stack have been
updated as well

`DetectHttp` has been updated to work as either a `MakeService` or
`NewService` -- the asynchronous version is still needed by the outbound
proxy (until caching is changed).

`DetectTls` is now purely a `NewService`.

Author: olix0r

linkerd/app/inbound/src/lib.rs

@@ -16,7 +16,7 @@ use linkerd2_app_core::{
     opencensus::proto::trace::v1 as oc,
     profiles,
     proxy::{
-        http::{self, orig_proto, strip_header, DetectHttp},
+        http::{self, orig_proto, strip_header},
         identity, tap, tcp,
     },
     reconnect, router,
@@ -367,15 +367,19 @@ impl Config {
             .check_new_service::<tls::accept::Meta, http::Request<_>>()
             .into_inner();
 
-        DetectHttp::new(
+        svc::stack(http::DetectHttp::new(
             h2_settings,
-            detect_protocol_timeout,
             http_server,
             svc::stack(tcp_forward)
                 .push_map_target(TcpEndpoint::from)
                 .into_inner(),
             drain.clone(),
-        )
+        ))
+        .push_on_response(transport::Prefix::layer(
+            http::Version::DETECT_BUFFER_CAPACITY,
+            detect_protocol_timeout,
+        ))
+        .into_inner()
     }
 
     pub fn build_tls_accept<D, A, F, B>(

linkerd/app/outbound/src/lib.rs

@@ -518,17 +518,22 @@ impl Config {
             .push_map_target(|a: listen::Addrs| a.target_addr())
             .into_inner();
 
-        let http = http::DetectHttp::new(
+        let http = svc::stack(http::DetectHttp::new(
             h2_settings,
-            detect_protocol_timeout,
             http_server,
             tcp_balance,
             drain.clone(),
-        );
+        ))
+        .push_on_response(transport::Prefix::layer(
+            http::Version::DETECT_BUFFER_CAPACITY,
+            detect_protocol_timeout,
+        ))
+        .into_new_service()
+        .into_inner();
 
         svc::stack(svc::stack::MakeSwitch::new(
             skip_detect.clone(),
-            svc::stack(http).into_new_service().into_inner(),
+            http,
             tcp_forward.push_map_target(TcpEndpoint::from).into_inner(),
         ))
         .push(metrics.transport.layer_accept(TransportLabels))

linkerd/proxy/http/src/detect.rs

@@ -13,22 +13,17 @@ use std::{
     future::Future,
     pin::Pin,
     task::{Context, Poll},
-    time::Duration,
 };
 use tower::{util::ServiceExt, Service};
-use tracing::{debug, info_span, trace};
+use tracing::{debug, info_span};
 use tracing_futures::Instrument;
 
 type Server = hyper::server::conn::Http<trace::Executor>;
 
-#[derive(Copy, Clone, Debug)]
-pub struct DetectTimeout(());
-
 #[derive(Clone, Debug)]
 pub struct DetectHttp<F, H> {
     tcp: F,
     http: H,
-    timeout: Duration,
     server: Server,
     drain: drain::Watch,
 }
@@ -45,7 +40,6 @@ pub struct DetectHttp<F, H> {
 pub struct AcceptHttp<F, H> {
     tcp: F,
     http: H,
-    timeout: Duration,
     server: hyper::server::conn::Http<trace::Executor>,
     drain: drain::Watch,
 }
@@ -54,14 +48,13 @@ pub struct AcceptHttp<F, H> {
 
 impl<F, H> DetectHttp<F, H> {
     /// Creates a new `AcceptHttp`.
-    pub fn new(h2: H2Settings, timeout: Duration, http: H, tcp: F, drain: drain::Watch) -> Self {
+    pub fn new(h2: H2Settings, http: H, tcp: F, drain: drain::Watch) -> Self {
         let mut server = hyper::server::conn::Http::new().with_executor(trace::Executor::new());
         server
             .http2_initial_stream_window_size(h2.initial_stream_window_size)
             .http2_initial_connection_window_size(h2.initial_connection_window_size);
 
         Self {
-            timeout,
             server,
             tcp,
             http,
@@ -81,7 +74,6 @@ where
     fn new_service(&mut self, target: T) -> Self::Service {
         AcceptHttp::new(
             self.server.clone(),
-            self.timeout,
             self.http.new_service(target.clone()),
             self.tcp.new_service(target),
             self.drain.clone(),
@@ -115,34 +107,32 @@ where
         let tcp = self.tcp.clone();
         let http = self.http.clone();
         let server = self.server.clone();
-        let timeout = self.timeout;
 
         Box::pin(async move {
             let (tcp, http) = futures::try_join!(
                 tcp.oneshot(target.clone()).map_err(Into::<Error>::into),
                 http.oneshot(target).map_err(Into::<Error>::into)
             )?;
 
-            Ok(AcceptHttp::new(server, timeout, http, tcp, drain))
+            Ok(AcceptHttp::new(server, http, tcp, drain))
         })
     }
 }
 
 // === impl AcceptHttp ===
 
 impl<F, H> AcceptHttp<F, H> {
-    pub fn new(server: Server, timeout: Duration, http: H, tcp: F, drain: drain::Watch) -> Self {
+    pub fn new(server: Server, http: H, tcp: F, drain: drain::Watch) -> Self {
         Self {
             server,
-            timeout,
             tcp,
             http,
             drain,
         }
     }
 }
 
-impl<I, F, S> Service<I> for AcceptHttp<F, S>
+impl<I, F, S> Service<PrefixedIo<I>> for AcceptHttp<F, S>
 where
     I: io::AsyncRead + io::AsyncWrite + Send + Unpin + 'static,
     F: tower::Service<PrefixedIo<I>, Response = ()> + Clone + Send + 'static,
@@ -164,22 +154,14 @@ where
         Poll::Ready(Ok(().into()))
     }
 
-    fn call(&mut self, io: I) -> Self::Future {
+    fn call(&mut self, io: PrefixedIo<I>) -> Self::Future {
         let drain = self.drain.clone();
         let tcp = self.tcp.clone();
         let http = self.http.clone();
         let mut server = self.server.clone();
 
-        let timeout = tokio::time::delay_for(self.timeout);
+        let version = HttpVersion::from_prefix(io.prefix());
         Box::pin(async move {
-            trace!("Detecting");
-            let (version, io) = tokio::select! {
-                res = HttpVersion::detect(io) => { res? }
-                () = timeout => {
-                    return Err(DetectTimeout(()).into());
-                }
-            };
-
             match version {
                 Some(HttpVersion::Http1) => {
                     debug!("Handling as HTTP");
@@ -220,11 +202,3 @@ where
         })
     }
 }
-
-impl std::fmt::Display for DetectTimeout {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "HTTP detection timeout")
-    }
-}
-
-impl std::error::Error for DetectTimeout {}

linkerd/proxy/http/src/version.rs

@@ -1,4 +1,3 @@
-use linkerd2_proxy_transport::io::{self, Peekable, PrefixedIo};
 use tracing::{debug, trace};
 
 #[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
@@ -22,6 +21,7 @@ impl std::convert::TryFrom<http::Version> for Version {
 }
 
 impl Version {
+    pub const DETECT_BUFFER_CAPACITY: usize = 8192;
     const H2_PREFACE: &'static [u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
 
     /// Tries to detect a known protocol in the peeked bytes.
@@ -60,17 +60,6 @@ impl Version {
         trace!(?bytes);
         None
     }
-
-    pub async fn detect<I>(io: I) -> io::Result<(Option<Self>, PrefixedIo<I>)>
-    where
-        I: io::AsyncRead + io::AsyncWrite + Unpin,
-    {
-        // If we don't find a newline, we consider the stream to be HTTP/1; so
-        // we need enough capacity to prevent false-positives.
-        let io = io.peek(8192).await?;
-        let version = Self::from_prefix(io.prefix());
-        Ok((version, io))
-    }
 }
 
 impl std::fmt::Display for Unsupported {

linkerd/proxy/transport/src/lib.rs

@@ -7,12 +7,14 @@ pub mod connect;
 pub use linkerd2_io as io;
 pub mod listen;
 pub mod metrics;
+pub mod prefix;
 pub mod tls;
 
 pub use self::{
     connect::Connect,
     io::BoxedIo,
     listen::{Bind, DefaultOrigDstAddr, NoOrigDstAddr, OrigDstAddr},
+    prefix::Prefix,
 };
 
 // Misc.

linkerd/proxy/transport/src/prefix.rs

@@ -0,0 +1,76 @@
+use crate::io;
+use futures::prelude::*;
+use linkerd2_error::Error;
+use linkerd2_stack::layer;
+use std::{
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+    time::Duration,
+};
+use tokio::time;
+use tower::util::ServiceExt;
+use tracing::{debug, trace};
+
+#[derive(Copy, Clone)]
+pub struct Prefix<S> {
+    inner: S,
+    capacity: usize,
+    timeout: Duration,
+}
+
+#[derive(Debug)]
+pub struct ReadTimeout(());
+
+impl<S> Prefix<S> {
+    pub fn new(inner: S, capacity: usize, timeout: Duration) -> Self {
+        Self {
+            inner,
+            capacity,
+            timeout,
+        }
+    }
+
+    pub fn layer(
+        capacity: usize,
+        timeout: Duration,
+    ) -> impl layer::Layer<S, Service = Prefix<S>> + Clone {
+        layer::mk(move |inner| Self::new(inner, capacity, timeout))
+    }
+}
+
+impl<S, I> tower::Service<I> for Prefix<S>
+where
+    I: io::Peekable + Send + 'static,
+    S: tower::Service<io::PrefixedIo<I>, Response = ()> + Clone + Send + 'static,
+    S::Error: Into<Error>,
+    S::Future: Send,
+{
+    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(().into()))
+    }
+
+    fn call(&mut self, io: I) -> Self::Future {
+        debug!(capacity = self.capacity, "Buffering prefix");
+        let accept = self.inner.clone();
+        let peek = time::timeout(self.timeout, io.peek(self.capacity)).map_err(|_| ReadTimeout(()));
+        Box::pin(async move {
+            let io = peek.await??;
+            trace!(read = %io.prefix().len());
+            accept.oneshot(io).err_into::<Error>().await?;
+            Ok(())
+        })
+    }
+}
+
+impl std::fmt::Display for ReadTimeout {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Timed out while reading client stream prefix")
+    }
+}
+
+impl std::error::Error for ReadTimeout {}