detect: Make protocol detection more robust (#744)

The existing protocol detection implementation is somewhat naive: we
simply perform a single read on the socket and use that initial buffer
to try to determine the protocol. This means that if the initial read
only returns a few bytes, we won't actually have enough information to
do protocol detection. Furthermore, because we don't actually read until
a `\r\n`, we can incorrectly infer that a non-HTTP protocol is HTTP/1.1.

This change fixes this by replacing the `Prefix` middleware with a
`Detect` service and trait that does its own buffering as it tries to
read from the socket.

When detection times out, an error is thrown to the server. This change
requires modifying test data to include a trailing newline (since it's
not feasible to close these connections given the current test
infrastructure).

This change increases the default dispatch (and therefore detection)
timeouts to 5s.

Author: olix0r

Cargo.lock

@@ -1360,6 +1360,7 @@ dependencies = [
 name = "linkerd2-proxy-http"
 version = "0.1.0"
 dependencies = [
+ "async-trait",
  "bytes 0.6.0",
  "futures 0.3.5",
  "h2 0.3.0",
@@ -1375,14 +1376,17 @@ dependencies = [
  "linkerd2-http-box",
  "linkerd2-identity",
  "linkerd2-io",
+ "linkerd2-proxy-transport",
  "linkerd2-stack",
  "linkerd2-timeout",
  "pin-project 1.0.2",
  "rand 0.7.2",
  "tokio 0.3.5",
+ "tokio-test 0.3.0",
  "tower",
  "tracing",
  "tracing-futures",
+ "tracing-subscriber",
  "try-lock",
 ]
 
@@ -1459,6 +1463,7 @@ name = "linkerd2-proxy-transport"
 version = "0.1.0"
 dependencies = [
  "async-stream 0.2.1",
+ "async-trait",
  "bytes 0.6.0",
  "futures 0.3.5",
  "indexmap",

linkerd/app/inbound/src/lib.rs

@@ -317,10 +317,10 @@ impl Config {
     where
         I: io::AsyncRead + io::AsyncWrite + io::PeerAddr + Unpin + Send + 'static,
         F: svc::NewService<TcpEndpoint, Service = A> + Unpin + Clone + Send + 'static,
-        A: tower::Service<io::PrefixedIo<I>, Response = ()> + Clone + Send + 'static,
+        A: tower::Service<io::PrefixedIo<I>, Response = ()> + Clone + Send + Sync + 'static,
         A::Error: Into<Error>,
         A::Future: Send,
-        H: svc::NewService<Target, Service = S> + Unpin + Clone + Send + 'static,
+        H: svc::NewService<Target, Service = S> + Unpin + Clone + Send + Sync + 'static,
         S: tower::Service<
                 http::Request<http::boxed::BoxBody>,
                 Response = http::Response<http::boxed::BoxBody>,
@@ -335,7 +335,7 @@ impl Config {
             dispatch_timeout,
             max_in_flight_requests,
             detect_protocol_timeout,
-            buffer_capacity,
+            cache_max_idle_age,
             ..
         } = self.proxy.clone();
 
@@ -389,10 +389,12 @@ impl Config {
                 .into_inner(),
             drain.clone(),
         ))
-        .push_on_response(svc::layers().push_spawn_buffer(buffer_capacity).push(
-            transport::Prefix::layer(
-                http::Version::DETECT_BUFFER_CAPACITY,
+        .check_new_clone::<(Option<http::Version>, TcpAccept)>()
+        .push_cache(cache_max_idle_age)
+        .push(transport::NewDetectService::layer(
+            transport::detect::DetectTimeout::new(
                 detect_protocol_timeout,
+                http::DetectHttp::default(),
             ),
         ))
         .into_inner()

linkerd/app/integration/src/tests/identity.rs

@@ -25,7 +25,7 @@ async fn nonblocking_identity_detection() {
         .await;
     let proxy = proxy::new().identity(id_svc);
 
-    let msg1 = "custom tcp hello";
+    let msg1 = "custom tcp hello\n";
     let msg2 = "custom tcp bye";
     let srv = server::tcp()
         .accept(move |read| {

linkerd/app/integration/src/tests/shutdown.rs

@@ -100,7 +100,7 @@ async fn tcp_waits_for_proxies_to_close() {
     let _trace = trace_init();
 
     let (shdn, rx) = shutdown_signal();
-    let msg1 = "custom tcp hello";
+    let msg1 = "custom tcp hello\n";
     let msg2 = "custom tcp bye";
 
     let srv = server::tcp()

linkerd/app/integration/src/tests/telemetry.rs

@@ -76,7 +76,7 @@ impl Fixture {
 }
 
 impl TcpFixture {
-    const HELLO_MSG: &'static str = "custom tcp hello";
+    const HELLO_MSG: &'static str = "custom tcp hello\n";
     const BYE_MSG: &'static str = "custom tcp bye";
 
     async fn server() -> server::Listening {

linkerd/app/integration/src/tests/transparency.rs

@@ -50,7 +50,7 @@ async fn inbound_http1() {
 async fn outbound_tcp() {
     let _trace = trace_init();
 
-    let msg1 = "custom tcp hello";
+    let msg1 = "custom tcp hello\n";
     let msg2 = "custom tcp bye";
 
     let srv = server::tcp()
@@ -88,7 +88,7 @@ async fn outbound_tcp() {
 async fn outbound_tcp_external() {
     let _trace = trace_init();
 
-    let msg1 = "custom tcp hello";
+    let msg1 = "custom tcp hello\n";
     let msg2 = "custom tcp bye";
 
     let srv = server::tcp()
@@ -127,7 +127,7 @@ async fn outbound_tcp_external() {
 async fn inbound_tcp() {
     let _trace = trace_init();
 
-    let msg1 = "custom tcp hello";
+    let msg1 = "custom tcp hello\n";
     let msg2 = "custom tcp bye";
 
     let srv = server::tcp()
@@ -296,7 +296,7 @@ async fn tcp_server_first_tls() {
 async fn tcp_connections_close_if_client_closes() {
     let _trace = trace_init();
 
-    let msg1 = "custom tcp hello";
+    let msg1 = "custom tcp hello\n";
     let msg2 = "custom tcp bye";
 
     let (mut tx, mut rx) = mpsc::channel(1);

linkerd/app/outbound/src/ingress.rs

@@ -46,6 +46,7 @@ where
             Error = Error,
         > + Clone
         + Send
+        + Sync
         + 'static,
     TSvc::Future: Send,
     H: svc::NewService<http::Logical, Service = HSvc> + Unpin + Clone + Send + Sync + 'static,
@@ -130,11 +131,13 @@ where
         .into_inner();
 
     svc::stack(http::NewServeHttp::new(h2_settings, http, tcp, drain))
-        .check_new_service::<tcp::Accept, io::PrefixedIo<transport::metrics::SensorIo<I>>>()
-        .push_on_response(svc::layers().push_spawn_buffer(buffer_capacity).push(
-            transport::Prefix::layer(
-                http::Version::DETECT_BUFFER_CAPACITY,
+        .check_new_service::<(Option<http::Version>, tcp::Accept), io::PrefixedIo<transport::metrics::SensorIo<I>>>()
+        .check_new_clone::<(Option<http::Version>, tcp::Accept)>()
+        .push_cache(cache_max_idle_age)
+        .push(transport::NewDetectService::layer(
+            transport::detect::DetectTimeout::new(
                 detect_protocol_timeout,
+                http::DetectHttp::default(),
             ),
         ))
         .check_new_service::<tcp::Accept, transport::metrics::SensorIo<I>>()

linkerd/app/outbound/src/server.rs

@@ -156,6 +156,7 @@ where
         max_in_flight_requests,
         detect_protocol_timeout,
         buffer_capacity,
+        cache_max_idle_age,
         ..
     } = config.proxy.clone();
 
@@ -222,11 +223,13 @@ where
         tcp_balance,
         drain.clone(),
     ))
-    .check_new_service::<tcp::Logical, transport::io::PrefixedIo<transport::metrics::SensorIo<I>>>()
-    .push_on_response(svc::layers().push_spawn_buffer(buffer_capacity).push(
-        transport::Prefix::layer(
-            http::Version::DETECT_BUFFER_CAPACITY,
+    .check_new_clone::<(Option<http::Version>, tcp::Logical)>()
+    .check_new_service::<(Option<http::Version>, tcp::Logical), transport::io::PrefixedIo<transport::metrics::SensorIo<I>>>()
+    .push_cache(cache_max_idle_age)
+    .push(transport::NewDetectService::layer(
+        transport::detect::DetectTimeout::new(
             detect_protocol_timeout,
+            http::DetectHttp::default(),
         ),
     ))
     .check_new_service::<tcp::Logical, transport::metrics::SensorIo<I>>()

linkerd/app/outbound/src/target.rs

@@ -127,6 +127,21 @@ impl<P> Logical<P> {
     }
 }
 
+impl<P: PartialEq> PartialEq<Logical<P>> for Logical<P> {
+    fn eq(&self, other: &Logical<P>) -> bool {
+        self.orig_dst == other.orig_dst && self.protocol == other.protocol
+    }
+}
+
+impl<P: Eq> Eq for Logical<P> {}
+
+impl<P: std::hash::Hash> std::hash::Hash for Logical<P> {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.orig_dst.hash(state);
+        self.protocol.hash(state);
+    }
+}
+
 impl<P: std::fmt::Debug> std::fmt::Debug for Logical<P> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Logical")

linkerd/app/outbound/src/tcp/tests.rs

@@ -865,7 +865,7 @@ where
         svc
     };
     async move {
-        let io = support::io().read(b"hello\r\n").write(b"world").build();
+        let io = support::io().read(b"hello\n").write(b"world").build();
         let res = svc.oneshot(io).err_into::<Error>().await;
         tracing::trace!(?res);
         if let Err(err) = res {