outbound: Avoid redundant TCP endpoint resolution (#742)

This change extends the changes in #736 to the
TCP-forwarding stack so that load balancer resolutions are not created
when the control plane does not indicate that the target is a service.

Author: olix0r

linkerd/app/outbound/src/server.rs

@@ -93,6 +93,18 @@ where
         .check_new_service::<(http::Version, tcp::Logical), http::Request<_>>()
         .into_inner();
 
+    let tcp_forward = svc::stack(tcp_connect.clone())
+        .push_make_thunk()
+        .check_make_service::<tcp::Endpoint, ()>()
+        .push_on_response(svc::layer::mk(tcp::Forward::new))
+        .into_new_service()
+        .check_new_service::<tcp::Endpoint, transport::io::PrefixedIo<transport::metrics::SensorIo<I>>>()
+        .push_map_target(tcp::Endpoint::from_logical(
+            tls::ReasonForNoPeerName::NotProvidedByServiceDiscovery,
+        ))
+        .check_new_service::<tcp::Logical, transport::io::PrefixedIo<transport::metrics::SensorIo<I>>>()
+        .into_inner();
+
     // Load balances TCP streams that cannot be decoded as HTTP.
     let tcp_balance = svc::stack(tcp::balance::stack(
         &config.proxy,
@@ -101,6 +113,8 @@ where
     ))
     .push_map_target(tcp::Concrete::from)
     .push(profiles::split::layer())
+    .check_new_service::<tcp::Logical, transport::io::PrefixedIo<transport::metrics::SensorIo<I>>>()
+    .push_switch(tcp::Logical::should_resolve, tcp_forward)
     .push_on_response(
         svc::layers()
             .push_failfast(dispatch_timeout)

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

@@ -172,6 +172,7 @@ async fn resolutions_are_reused() {
 
     let addr = SocketAddr::new([0, 0, 0, 0].into(), 5550);
     let cfg = default_config(addr);
+    let svc_name = profile::Name::from_str("foo.ns1.svc.example.com").unwrap();
     let id_name =
         linkerd2_identity::Name::from_str("foo.ns1.serviceaccount.identity.linkerd.cluster.local")
             .expect("hostname is valid");
@@ -195,10 +196,16 @@ async fn resolutions_are_reused() {
 
     // Configure the mock destination resolver to just give us a single endpoint
     // for the target, which always exists and has no metadata.
-    let resolver = support::resolver().endpoint_exists(Addr::from(addr), addr, meta);
+    let resolver = support::resolver().endpoint_exists((svc_name.clone(), addr.port()), addr, meta);
     let resolve_state = resolver.handle();
 
-    let profiles = support::profiles().profile(addr, Default::default());
+    let profiles = support::profiles().profile(
+        addr,
+        profile::Profile {
+            name: Some(svc_name),
+            ..profile::Profile::default()
+        },
+    );
     let profile_state = profiles.handle();
 
     // Build the outbound server
@@ -252,6 +259,7 @@ async fn load_balances() {
     ];
 
     let cfg = default_config(svc_addr);
+    let svc_name = profile::Name::from_str("foo.ns1.svc.example.com").unwrap();
     let id_name =
         linkerd2_identity::Name::from_str("foo.ns1.serviceaccount.identity.linkerd.cluster.local")
             .expect("hostname is valid");
@@ -269,7 +277,13 @@ async fn load_balances() {
         );
     }
 
-    let profiles = support::profile::resolver().profile(svc_addr, Default::default());
+    let profiles = support::profile::resolver().profile(
+        svc_addr,
+        profile::Profile {
+            name: Some(svc_name.clone()),
+            ..Default::default()
+        },
+    );
     let profile_state = profiles.handle();
 
     let meta = support::resolver::Metadata::new(
@@ -281,7 +295,7 @@ async fn load_balances() {
     );
 
     let resolver = support::resolver();
-    let mut dst = resolver.endpoint_tx(Addr::Socket(svc_addr));
+    let mut dst = resolver.endpoint_tx((svc_name, svc_addr.port()));
     dst.add(endpoints.iter().map(|&(addr, _)| (addr, meta.clone())))
         .expect("still listening");
     let resolve_state = resolver.handle();
@@ -342,6 +356,7 @@ async fn load_balancer_add_endpoints() {
     ];
 
     let cfg = default_config(svc_addr);
+    let svc_name = profile::Name::from_str("foo.ns1.svc.example.com").unwrap();
     let id_name =
         linkerd2_identity::Name::from_str("foo.ns1.serviceaccount.identity.linkerd.cluster.local")
             .expect("hostname is valid");
@@ -358,7 +373,13 @@ async fn load_balancer_add_endpoints() {
         );
     }
 
-    let profiles = support::profile::resolver().profile(svc_addr, Default::default());
+    let profiles = support::profile::resolver().profile(
+        svc_addr,
+        profile::Profile {
+            name: Some(svc_name.clone()),
+            ..Default::default()
+        },
+    );
 
     let meta = support::resolver::Metadata::new(
         Default::default(),
@@ -369,7 +390,7 @@ async fn load_balancer_add_endpoints() {
     );
 
     let resolver = support::resolver();
-    let mut dst = resolver.endpoint_tx(Addr::Socket(svc_addr));
+    let mut dst = resolver.endpoint_tx((svc_name, svc_addr.port()));
     dst.add(Some((endpoints[0].0, meta.clone())))
         .expect("still listening");
 
@@ -450,6 +471,7 @@ async fn load_balancer_remove_endpoints() {
     ];
 
     let cfg = default_config(svc_addr);
+    let svc_name = profile::Name::from_str("foo.ns1.svc.example.com").unwrap();
     let id_name =
         linkerd2_identity::Name::from_str("foo.ns1.serviceaccount.identity.linkerd.cluster.local")
             .expect("hostname is valid");
@@ -466,7 +488,13 @@ async fn load_balancer_remove_endpoints() {
         );
     }
 
-    let profiles = support::profile::resolver().profile(svc_addr, Default::default());
+    let profiles = support::profile::resolver().profile(
+        svc_addr,
+        profile::Profile {
+            name: Some(svc_name.clone()),
+            ..Default::default()
+        },
+    );
 
     let meta = support::resolver::Metadata::new(
         Default::default(),
@@ -477,7 +505,7 @@ async fn load_balancer_remove_endpoints() {
     );
 
     let resolver = support::resolver();
-    let mut dst = resolver.endpoint_tx(Addr::Socket(svc_addr));
+    let mut dst = resolver.endpoint_tx((svc_name, svc_addr.port()));
     dst.add(Some((endpoints[0].0, meta.clone())))
         .expect("still listening");
 
@@ -539,6 +567,7 @@ async fn no_profiles_when_outside_search_nets() {
         allow_discovery: IpMatch::new(Some(IpNet::from_str("10.0.0.0/8").unwrap())).into(),
         ..default_config(profile_addr)
     };
+    let svc_name = profile::Name::from_str("foo.ns1.svc.example.com").unwrap();
     let id_name =
         linkerd2_identity::Name::from_str("foo.ns1.serviceaccount.identity.linkerd.cluster.local")
             .expect("hostname is invalid");
@@ -578,11 +607,20 @@ async fn no_profiles_when_outside_search_nets() {
 
     // Configure the mock destination resolver to just give us a single endpoint
     // for the target, which always exists and has no metadata.
-    let resolver =
-        support::resolver().endpoint_exists(Addr::from(profile_addr), profile_addr, meta);
+    let resolver = support::resolver().endpoint_exists(
+        (svc_name.clone(), profile_addr.port()),
+        profile_addr,
+        meta,
+    );
     let resolve_state = resolver.handle();
 
-    let profiles = support::profiles().profile(profile_addr, Default::default());
+    let profiles = support::profiles().profile(
+        profile_addr,
+        profile::Profile {
+            name: Some(svc_name),
+            ..Default::default()
+        },
+    );
     let profile_state = profiles.handle();
 
     // Build the outbound server

linkerd/app/test/src/connect.rs

@@ -35,9 +35,14 @@ where
         let span = tracing::info_span!("connect", %addr);
         let f = span.in_scope(|| {
             tracing::trace!("connecting...");
-            match self.endpoints.lock().unwrap().get_mut(&addr) {
+            let mut endpoints = self.endpoints.lock().unwrap();
+            match endpoints.get_mut(&addr) {
                 Some(f) => (f)(endpoint),
-                None => panic!("did not expect to connect to the endpoint {:?}", endpoint),
+                None => panic!(
+                    "did not expect to connect to the endpoint {} not in {:?}",
+                    addr,
+                    endpoints.keys().collect::<Vec<_>>()
+                ),
             }
         });
         f.instrument(span)

linkerd/app/test/src/resolver.rs

@@ -81,14 +81,14 @@ impl<T, E> Dst<T, E>
 where
     T: Hash + Eq,
 {
-    pub fn endpoint_tx(&self, endpoint: T) -> DstSender<E> {
+    pub fn endpoint_tx(&self, t: impl Into<T>) -> DstSender<E> {
         let (tx, rx) = mpsc::unbounded_channel();
-        self.state.endpoints.lock().unwrap().insert(endpoint, rx);
+        self.state.endpoints.lock().unwrap().insert(t.into(), rx);
         DstSender(tx)
     }
 
-    pub fn endpoint_exists(self, endpoint: T, addr: SocketAddr, meta: E) -> Self {
-        let mut tx = self.endpoint_tx(endpoint);
+    pub fn endpoint_exists(self, t: impl Into<T>, addr: SocketAddr, meta: E) -> Self {
+        let mut tx = self.endpoint_tx(t);
         tx.add(vec![(addr, meta)]).unwrap();
         self
     }

linkerd/service-profiles/src/lib.rs

@@ -1,7 +1,7 @@
 #![deny(warnings, rust_2018_idioms)]
 
 use linkerd2_addr::Addr;
-use linkerd2_dns_name::Name;
+pub use linkerd2_dns_name::Name;
 use linkerd2_error::Error;
 use linkerd2_proxy_api_resolve::Metadata;
 use std::{