fix cloning detecthttp blowing away state (#700)

The `AcceptHttp` service is stateful. It builds the underlying HTTP,
HTTP/2, and TCP service stacks lazily the first time they're needed.
This interacts badly with the derived implementation of `Clone`: if it's
cloned before the state is set, the state will only be set for the
clone, rather than for the original instance in the cache. Since
`Prefix` clones and oneshots the underlying service (`AcceptHttp`), this
means that we rebuild the HTTP, HTTP/2, and TCP stacks on every
connection to the same orig dest. This basically breaks caching for
everything other than service profile discovery, causing us to do new
destination discovery for every connection to the same original
destination. Which...isn't great.

This branch fixes this issue by removing the `Clone` impl from
`AcceptHttp` and putting it behind a buffer instead. In the future, it
may be possible to fix this in a nicer way, but this resolves the issue
for now.

Author: hawkw

linkerd/app/inbound/src/lib.rs

@@ -319,6 +319,7 @@ impl Config {
             dispatch_timeout,
             max_in_flight_requests,
             detect_protocol_timeout,
+            buffer_capacity,
             ..
         } = self.proxy.clone();
 
@@ -373,9 +374,11 @@ impl Config {
                 .into_inner(),
             drain.clone(),
         ))
-        .push_on_response(transport::Prefix::layer(
-            http::Version::DETECT_BUFFER_CAPACITY,
-            detect_protocol_timeout,
+        .push_on_response(svc::layers().push_spawn_buffer(buffer_capacity).push(
+            transport::Prefix::layer(
+                http::Version::DETECT_BUFFER_CAPACITY,
+                detect_protocol_timeout,
+            ),
         ))
         .into_inner()
     }

linkerd/app/outbound/src/lib.rs

@@ -403,10 +403,11 @@ impl Config {
             endpoint::TcpLogical,
             transport::io::PrefixedIo<transport::metrics::SensorIo<I>>,
         >()
-        .push_on_response(transport::Prefix::layer(
+        .push_on_response(
+            svc::layers().push_spawn_buffer(buffer_capacity).push(transport::Prefix::layer(
             http::Version::DETECT_BUFFER_CAPACITY,
             detect_protocol_timeout,
-        ))
+        )))
         .check_new_service::<endpoint::TcpLogical, transport::metrics::SensorIo<I>>()
         .into_inner();
 

linkerd/proxy/http/src/detect.rs

@@ -15,7 +15,7 @@ use std::{
     task::{Context, Poll},
 };
 use tower::{util::ServiceExt, Service};
-use tracing::debug;
+use tracing::{debug, trace};
 
 type Server = hyper::server::conn::Http<trace::Executor>;
 
@@ -35,7 +35,7 @@ pub struct DetectHttp<F, H> {
 ///
 /// Otherwise, the `F` type forwarding service is used to handle the TCP
 /// connection.
-#[derive(Clone, Debug)]
+#[derive(Debug)]
 pub struct AcceptHttp<T, F: NewService<T>, H: NewService<(HttpVersion, T)>> {
     target: T,
     new_tcp: F,
@@ -92,6 +92,7 @@ where
     H: NewService<(HttpVersion, T)>,
 {
     pub fn new(target: T, server: Server, new_http: H, new_tcp: F, drain: drain::Watch) -> Self {
+        tracing::trace!("new AcceptHttp");
         Self {
             target,
             server,
@@ -136,8 +137,10 @@ where
             Some(HttpVersion::Http1) => {
                 debug!("Handling as HTTP");
                 let http1 = if let Some(svc) = self.http1.clone() {
+                    trace!("HTTP service already exists");
                     svc
                 } else {
+                    trace!("Building new HTTP service");
                     let svc = self
                         .new_http
                         .new_service((HttpVersion::Http1, self.target.clone()));
@@ -167,8 +170,10 @@ where
             Some(HttpVersion::H2) => {
                 debug!("Handling as H2");
                 let h2 = if let Some(svc) = self.h2.clone() {
+                    trace!("H2 service already exists");
                     svc
                 } else {
+                    trace!("Building new H2 service");
                     let svc = self
                         .new_http
                         .new_service((HttpVersion::H2, self.target.clone()));
@@ -193,8 +198,10 @@ where
             None => {
                 debug!("Forwarding TCP");
                 let tcp = if let Some(svc) = self.tcp.clone() {
+                    trace!("TCP service already exists");
                     svc
                 } else {
+                    trace!("Building new TCP service");
                     let svc = self.new_tcp.new_service(self.target.clone());
                     self.tcp = Some(svc.clone());
                     svc