detect: Make protocol detection more robust

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/core/src/metrics.rs

@@ -56,6 +56,7 @@ pub struct EndpointLabels {
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct StackLabels {
     pub direction: Direction,
+    pub protocol: &'static str,
     pub name: &'static str,
 }
 
@@ -297,24 +298,26 @@ impl FmtLabels for TlsId {
 // === impl StackLabels ===
 
 impl StackLabels {
-    pub fn inbound(name: &'static str) -> Self {
+    pub fn inbound(protocol: &'static str, name: &'static str) -> Self {
         Self {
-            direction: Direction::In,
             name,
+            protocol,
+            direction: Direction::In,
         }
     }
 
-    pub fn outbound(name: &'static str) -> Self {
+    pub fn outbound(protocol: &'static str, name: &'static str) -> Self {
         Self {
-            direction: Direction::Out,
             name,
+            protocol,
+            direction: Direction::Out,
         }
     }
 }
 
 impl FmtLabels for StackLabels {
     fn fmt_labels(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         self.direction.fmt_labels(f)?;
-        write!(f, ",name=\"{}\"", self.name)
+        write!(f, ",protocol=\"{}\",name=\"{}\"", self.protocol, self.name)
     }
 }

linkerd/app/inbound/src/lib.rs

@@ -280,7 +280,7 @@ impl Config {
                 svc::layers()
                     .push_failfast(dispatch_timeout)
                     .push_spawn_buffer(buffer_capacity)
-                    .push(metrics.stack.layer(stack_labels("logical"))),
+                    .push(metrics.stack.layer(stack_labels("http", "logical"))),
             )
             .push_cache(cache_max_idle_age)
             .push_on_response(
@@ -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();
 
@@ -371,7 +371,7 @@ impl Config {
                     .push(TraceContext::layer(span_sink.map(|span_sink| {
                         SpanConverter::server(span_sink, trace_labels())
                     })))
-                    .push(metrics.stack.layer(stack_labels("source")))
+                    .push(metrics.stack.layer(stack_labels("http", "server")))
                     .box_http_request()
                     .box_http_response(),
             )
@@ -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()
@@ -458,8 +460,8 @@ pub fn trace_labels() -> HashMap<String, String> {
     l
 }
 
-fn stack_labels(name: &'static str) -> metrics::StackLabels {
-    metrics::StackLabels::inbound(name)
+fn stack_labels(proto: &'static str, name: &'static str) -> metrics::StackLabels {
+    metrics::StackLabels::inbound(proto, name)
 }
 
 // === impl SkipByPort ===

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/http/logical.rs

@@ -55,7 +55,11 @@ where
         .push_on_response(
             svc::layers()
                 .push(svc::layer::mk(svc::SpawnReady::new))
-                .push(metrics.stack.layer(stack_labels("balance.endpoint")))
+                .push(
+                    metrics
+                        .stack
+                        .layer(stack_labels("http", "balance.endpoint")),
+                )
                 .box_http_request(),
         )
         .check_new_service::<Endpoint, http::Request<_>>()
@@ -71,7 +75,7 @@ where
                 // If the balancer has been empty/unavailable for 10s, eagerly fail
                 // requests.
                 .push_failfast(dispatch_timeout)
-                .push(metrics.stack.layer(stack_labels("concrete"))),
+                .push(metrics.stack.layer(stack_labels("http", "concrete"))),
         )
         .into_new_service()
         .check_new_service::<Concrete, http::Request<_>>()

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,
@@ -110,7 +111,7 @@ where
                 .push(TraceContext::layer(span_sink.clone().map(|span_sink| {
                     SpanConverter::server(span_sink, trace_labels())
                 })))
-                .push(metrics.stack.layer(stack_labels("source")))
+                .push(metrics.stack.layer(stack_labels("http", "server")))
                 .push_failfast(dispatch_timeout)
                 .push_spawn_buffer(buffer_capacity)
                 .box_http_response(),
@@ -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/lib.rs

@@ -26,8 +26,8 @@ pub struct Config {
     pub allow_discovery: AddrMatch,
 }
 
-fn stack_labels(name: &'static str) -> metrics::StackLabels {
-    metrics::StackLabels::outbound(name)
+fn stack_labels(proto: &'static str, name: &'static str) -> metrics::StackLabels {
+    metrics::StackLabels::outbound(proto, name)
 }
 
 pub fn trace_labels() -> HashMap<String, String> {

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();
 
@@ -176,7 +177,7 @@ where
                 .push(TraceContext::layer(span_sink.clone().map(|span_sink| {
                     SpanConverter::server(span_sink, trace_labels())
                 })))
-                .push(metrics.stack.layer(stack_labels("source")))
+                .push(metrics.stack.layer(stack_labels("http", "server")))
                 .push_failfast(dispatch_timeout)
                 .push_spawn_buffer(buffer_capacity)
                 .box_http_response(),
@@ -193,6 +194,7 @@ where
         .check_make_service::<tcp::Endpoint, ()>()
         .push_on_response(svc::layer::mk(tcp::Forward::new))
         .into_new_service()
+        .push_on_response(metrics.stack.layer(stack_labels("tcp", "forward")))
         .check_new_service::<tcp::Endpoint, transport::io::PrefixedIo<transport::metrics::SensorIo<I>>>()
         .push_map_target(tcp::Endpoint::from_logical(
             tls::ReasonForNoPeerName::NotProvidedByServiceDiscovery,
@@ -221,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>>()
@@ -235,6 +239,7 @@ where
         .push_map_target(tcp::Endpoint::from_logical(
             tls::ReasonForNoPeerName::PortSkipped,
         ))
+        .push_on_response(metrics.stack.layer(stack_labels("tcp", "opaque")))
         .check_new_service::<tcp::Logical, transport::metrics::SensorIo<I>>()
         .into_inner();
 

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 {

linkerd/app/src/env.rs

@@ -170,14 +170,14 @@ pub const DEFAULT_INBOUND_LISTEN_ADDR: &str = "0.0.0.0:4143";
 pub const DEFAULT_CONTROL_LISTEN_ADDR: &str = "0.0.0.0:4190";
 const DEFAULT_ADMIN_LISTEN_ADDR: &str = "127.0.0.1:4191";
 const DEFAULT_METRICS_RETAIN_IDLE: Duration = Duration::from_secs(10 * 60);
-const DEFAULT_INBOUND_DISPATCH_TIMEOUT: Duration = Duration::from_secs(1);
+const DEFAULT_INBOUND_DISPATCH_TIMEOUT: Duration = Duration::from_secs(5);
 const DEFAULT_INBOUND_CONNECT_TIMEOUT: Duration = Duration::from_millis(100);
 const DEFAULT_INBOUND_CONNECT_BACKOFF: ExponentialBackoff = ExponentialBackoff {
     min: Duration::from_millis(100),
     max: Duration::from_millis(500),
     jitter: 0.1,
 };
-const DEFAULT_OUTBOUND_DISPATCH_TIMEOUT: Duration = Duration::from_secs(3);
+const DEFAULT_OUTBOUND_DISPATCH_TIMEOUT: Duration = Duration::from_secs(5);
 const DEFAULT_OUTBOUND_CONNECT_TIMEOUT: Duration = Duration::from_secs(1);
 const DEFAULT_OUTBOUND_CONNECT_BACKOFF: ExponentialBackoff = ExponentialBackoff {
     min: Duration::from_millis(100),

linkerd/concurrency-limit/src/lib.rs

@@ -33,7 +33,7 @@ pub struct ConcurrencyLimit<T> {
 }
 
 enum State {
-    Waiting(Pin<Box<dyn Future<Output = OwnedSemaphorePermit> + Send + 'static>>),
+    Waiting(Pin<Box<dyn Future<Output = OwnedSemaphorePermit> + Send + Sync + 'static>>),
     Ready(OwnedSemaphorePermit),
     Empty,
 }

linkerd/io/src/lib.rs

@@ -1,11 +1,9 @@
 mod boxed;
-mod peek;
 mod prefixed;
 mod sensor;
 
 pub use self::{
     boxed::BoxedIo,
-    peek::{Peek, Peekable},
     prefixed::PrefixedIo,
     sensor::{Sensor, SensorIo},
 };