ingress: Fall back to normal TCP forwarding

Currently, ingress-mode proxies ONLY support HTTP traffic. This is an
unfortunate tradeoff, as ingress may make arbitrary outbound connections
(e.g., TLS calls to external services). To avoid requiring that these
connections be completely skipped from the proxy, we can fallback to
transporting these connections after HTTP detection fails.

Ingress-mode proxies DO NOT fully honor port opacity configurations:
HTTP detection is always performed before discovery is attempted.

Signed-off-by: Oliver Gould <[email protected]>

Author: olix0r

linkerd/app/core/src/svc.rs

@@ -10,7 +10,7 @@ pub use linkerd_stack::{
     ExtractParam, Fail, FailFast, Filter, InsertParam, MakeConnection, MapErr, MapTargetLayer,
     NewRouter, NewService, Param, Predicate, UnwrapOr,
 };
-pub use linkerd_stack_tracing::{NewInstrument, NewInstrumentLayer};
+pub use linkerd_stack_tracing::{GetSpan, NewInstrument, NewInstrumentLayer};
 use std::{
     task::{Context, Poll},
     time::Duration,
@@ -262,6 +262,14 @@ impl<S> Stack<S> {
         self
     }
 
+    pub fn check_new_accept<T, I>(self) -> Self
+    where
+        S: NewService<T>,
+        S::Service: Service<I, Response = ()>,
+    {
+        self
+    }
+
     /// Validates that this stack serves T-typed targets.
     pub fn check_clone_new_service<T, Req>(self) -> Self
     where

linkerd/app/outbound/src/discover.rs

@@ -1,12 +1,11 @@
-use crate::{tcp, Outbound};
+use crate::Outbound;
 use linkerd_app_core::{
     io, profiles,
     svc::{self, stack::Param},
-    transport::{self, metrics::SensorIo, OrigDstAddr},
+    transport::OrigDstAddr,
     Error,
 };
-use std::convert::TryFrom;
-use tracing::{debug, debug_span, info_span};
+use tracing::debug;
 
 impl<N> Outbound<N> {
     /// Discovers the profile for a TCP endpoint.
@@ -23,41 +22,34 @@ impl<N> Outbound<N> {
     >
     where
         T: Param<OrigDstAddr>,
+        T: Clone + Eq + std::fmt::Debug + std::hash::Hash + Send + Sync + 'static,
         I: io::AsyncRead + io::AsyncWrite + io::PeerAddr + std::fmt::Debug + Send + Unpin + 'static,
-        N: svc::NewService<(Option<profiles::Receiver>, tcp::Accept), Service = NSvc>
-            + Clone
-            + Send
-            + Sync
-            + 'static,
-        NSvc: svc::Service<SensorIo<I>, Response = (), Error = Error> + Send + 'static,
+        N: svc::NewService<(Option<profiles::Receiver>, T), Service = NSvc>,
+        N: Clone + Send + Sync + 'static,
+        NSvc: svc::Service<I, Response = (), Error = Error> + Send + 'static,
         NSvc::Future: Send,
         P: profiles::GetProfile<profiles::LookupAddr> + Clone + Send + Sync + 'static,
         P::Future: Send,
         P::Error: Send,
     {
-        self.map_stack(|config, rt, accept| {
+        self.map_stack(|config, rt, inner| {
             let allow = config.allow_discovery.clone();
-            accept
-                .push(profiles::discover::layer(
-                    profiles,
-                    move |a: tcp::Accept| {
-                        let OrigDstAddr(addr) = a.orig_dst;
-                        if allow.matches_ip(addr.ip()) {
-                            debug!("Allowing profile lookup");
-                            Ok(profiles::LookupAddr(addr.into()))
-                        } else {
-                            tracing::debug!(
-                                %addr,
-                                networks = %allow.nets(),
-                                "Profile discovery not in configured search networks",
-                            );
-                            Err(profiles::DiscoveryRejected::new(
-                                "not in configured search networks",
-                            ))
-                        }
-                    },
-                ))
-                .instrument(|_: &_| debug_span!("profile"))
+            inner
+                .push(profiles::discover::layer(profiles, move |t: T| {
+                    let OrigDstAddr(addr) = t.param();
+                    if allow.matches_ip(addr.ip()) {
+                        debug!("Allowing profile lookup");
+                        return Ok(profiles::LookupAddr(addr.into()));
+                    }
+                    tracing::debug!(
+                        %addr,
+                        networks = %allow.nets(),
+                        "Profile discovery not in configured search networks",
+                    );
+                    Err(profiles::DiscoveryRejected::new(
+                        "not in configured search networks",
+                    ))
+                }))
                 .push_on_service(
                     svc::layers()
                         // If the traffic split is empty/unavailable, eagerly fail
@@ -77,13 +69,7 @@ impl<N> Outbound<N> {
                         ))
                         .push_spawn_buffer(config.proxy.buffer_capacity),
                 )
-                .push(transport::metrics::NewServer::layer(
-                    rt.metrics.proxy.transport.clone(),
-                ))
                 .push_cache(config.proxy.cache_max_idle_age)
-                .instrument(|a: &tcp::Accept| info_span!("server", orig_dst = %a.orig_dst))
-                .push_request_filter(|t: T| tcp::Accept::try_from(t.param()))
-                .push(rt.metrics.tcp_errors.to_layer())
                 .push(svc::ArcNewService::layer())
                 .check_new_service::<T, I>()
         })