outbound: Strip endpoint identity when disabled (#862)

When the proxy's identity is disabled, we may still mark outbound
endpoints as having a peer identity (even though a TLS connection is
never established.

This change modifies the outbound HTTP and TCP endpoint stacks to
optionally strip server identities before metrics are recorded.

Author: olix0r

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

@@ -386,8 +386,7 @@ mod http2 {
         metrics::metric("tcp_close_total")
             .label("peer", "dst")
             .label("direction", "outbound")
-            .label("tls", "no_identity")
-            .label("no_tls_reason", "not_provided_by_service_discovery")
+            .label("tls", "disabled")
             .label(
                 "authority",
                 format_args!("disco.test.svc.cluster.local:{}", port),

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

@@ -158,8 +158,7 @@ async fn metrics_endpoint_outbound_request_count() {
         let srv_port = proxy.outbound_server.as_ref().unwrap().addr.port();
         metrics::labels()
             .label("direction", "outbound")
-            .label("tls", "no_identity")
-            .label("no_tls_reason", "not_provided_by_service_discovery")
+            .label("tls", "disabled")
             .label(
                 "authority",
                 format_args!("tele.test.svc.cluster.local:{}", srv_port),
@@ -319,13 +318,9 @@ mod response_classification {
     #[tokio::test]
     async fn outbound_http() {
         let fixture = async { Fixture::outbound_with_server(make_test_server().await).await };
-        test_http(
-            fixture,
-            "outbound",
-            "no_identity",
-            Some("not_provided_by_service_discovery"),
-            |proxy| Some(proxy.outbound_server.as_ref().unwrap().addr.port()),
-        )
+        test_http(fixture, "outbound", "disabled", None, |proxy| {
+            Some(proxy.outbound_server.as_ref().unwrap().addr.port())
+        })
         .await
     }
 }
@@ -1135,8 +1130,7 @@ mod transport {
                     format_args!("tele.test.svc.cluster.local:{}", port),
                 )
                 .label("direction", "outbound")
-                .label("tls", "no_identity")
-                .label("no_tls_reason", "not_provided_by_service_discovery")
+                .label("tls", "disabled")
         })
         .await;
     }
@@ -1170,8 +1164,7 @@ mod transport {
             metrics::labels()
                 .label("authority", proxy.outbound_server.as_ref().unwrap().addr)
                 .label("direction", "outbound")
-                .label("tls", "no_identity")
-                .label("no_tls_reason", "not_provided_by_service_discovery")
+                .label("tls", "disabled")
         })
         .await;
     }
@@ -1214,8 +1207,7 @@ mod transport {
         metrics::metric("tcp_close_total")
             .label("peer", "dst")
             .label("direction", "inbound")
-            .label("tls", "no_identity")
-            .label("no_tls_reason", "not_provided_by_service_discovery")
+            .label("tls", "disabled")
             .label("errno", "EXFULL")
             .value(1u64)
             .assert_in(&metrics)
@@ -1256,8 +1248,7 @@ mod transport {
         metrics::metric("tcp_close_total")
             .label("peer", "dst")
             .label("direction", "outbound")
-            .label("tls", "no_identity")
-            .label("no_tls_reason", "not_provided_by_service_discovery")
+            .label("tls", "disabled")
             .label("errno", "EXFULL")
             .value(1u64)
             .assert_in(&metrics)
@@ -1379,8 +1370,7 @@ mod transport {
         let mut dst_count = metrics::metric("tcp_connection_duration_count")
             .label("peer", "dst")
             .label("direction", "outbound")
-            .label("tls", "no_identity")
-            .label("no_tls_reason", "not_provided_by_service_discovery")
+            .label("tls", "disabled")
             .label("errno", "")
             .value(1u64);
 
@@ -1416,8 +1406,7 @@ mod transport {
             |proxy| {
                 metrics::labels()
                     .label("direction", "outbound")
-                    .label("tls", "no_identity")
-                    .label("no_tls_reason", "not_provided_by_service_discovery")
+                    .label("tls", "disabled")
                     .label("authority", proxy.outbound_server.as_ref().unwrap().addr)
             },
         )
@@ -1435,8 +1424,7 @@ mod transport {
             |proxy| {
                 metrics::labels()
                     .label("direction", "outbound")
-                    .label("tls", "no_identity")
-                    .label("no_tls_reason", "not_provided_by_service_discovery")
+                    .label("tls", "disabled")
                     .label("authority", proxy.outbound_server.as_ref().unwrap().addr)
             },
         )

linkerd/app/outbound/src/http/endpoint.rs

@@ -9,13 +9,14 @@ use linkerd_app_core::{
     proxy::{http, tap},
     reconnect,
     spans::SpanConverter,
-    svc, Error, TraceContext, CANONICAL_DST_HEADER, L5D_REQUIRE_ID,
+    svc, tls, Error, TraceContext, CANONICAL_DST_HEADER, L5D_REQUIRE_ID,
 };
 use tokio::{io, sync::mpsc};
 use tracing::debug_span;
 
 pub fn stack<B, C>(
     config: &ConnectConfig,
+    local_id: Option<&tls::LocalId>,
     tcp_connect: C,
     tap: tap::Registry,
     metrics: metrics::Proxy,
@@ -36,6 +37,7 @@ where
     C::Response: io::AsyncRead + io::AsyncWrite + Send + Unpin,
     C::Future: Send + Unpin,
 {
+    let identity_disabled = local_id.is_none();
     svc::stack(tcp_connect)
         // Initiates an HTTP client on the underlying transport. Prior-knowledge HTTP/2
         // is typically used (i.e. when communicating with other proxies); though
@@ -67,5 +69,12 @@ where
         .push_on_response(http::BoxResponse::layer())
         .check_new::<Endpoint>()
         .instrument(|e: &Endpoint| debug_span!("endpoint", peer.addr = %e.addr))
+        .push_map_target(move |e: Endpoint| {
+            if identity_disabled {
+                e.identity_disabled()
+            } else {
+                e
+            }
+        })
         .into_inner()
 }

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

@@ -89,6 +89,7 @@ where
         &cfg.proxy,
         super::endpoint::stack(
             &cfg.proxy.connect,
+            None,
             connect,
             tap,
             metrics.outbound.clone(),
@@ -403,8 +404,7 @@ async fn active_stacks_dont_idle_out() {
         let body = body.take().expect("service only called once in this test");
         Response::new(body)
     })
-    .http2()
-    .expect_identity(id.clone());
+    .http2();
 
     let connect = support::connect().endpoint_fn_boxed(ep1, server.run());
     let profiles = profile::resolver().profile(

linkerd/app/outbound/src/target.rs

@@ -224,6 +224,12 @@ impl<P> Endpoint<P> {
     pub fn from_accept(reason: tls::NoServerId) -> impl (Fn(Accept<P>) -> Self) + Clone {
         move |accept| Self::from_logical(reason)(Logical::from((None, accept)))
     }
+
+    /// Marks identity as disabled.
+    pub fn identity_disabled(mut self) -> Self {
+        self.identity = Conditional::None(tls::NoServerId::Disabled);
+        self
+    }
 }
 
 impl<P> Into<SocketAddr> for Endpoint<P> {

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

@@ -19,6 +19,7 @@ pub fn stack<P>(
     Error = Error,
     Future = impl Send,
 > + Clone {
+    let identity_disabled = local_identity.is_none();
     svc::stack(ConnectTcp::new(config.keepalive))
         // Initiates mTLS if the target is configured with identity.
         .push(tls::Client::layer(local_identity))
@@ -30,6 +31,13 @@ pub fn stack<P>(
         .push_timeout(config.timeout)
         .push(svc::stack::BoxFuture::layer())
         .push(metrics.transport.layer_connect())
+        .push_map_target(move |e: Endpoint<P>| {
+            if identity_disabled {
+                e.identity_disabled()
+            } else {
+                e
+            }
+        })
         .push_request_filter(PreventLoop { port: server_port })
         .into_inner()
 }

linkerd/app/src/lib.rs

@@ -150,6 +150,7 @@ impl Config {
                 &outbound.proxy,
                 outbound::http::endpoint::stack(
                     &outbound.proxy.connect,
+                    local_identity.as_ref().map(|l| l.id()),
                     outbound::tcp::connect::stack(
                         &outbound.proxy.connect,
                         outbound_addr.port(),

linkerd/app/test/src/http_util.rs

@@ -1,4 +1,4 @@
-use crate::app_core::{identity as id, io::BoxedIo, tls, Conditional, Error};
+use crate::app_core::{io::BoxedIo, tls, Error};
 use hyper::{body::HttpBody, Body, Request, Response};
 use std::{
     fmt,
@@ -8,7 +8,6 @@ use tracing::Instrument;
 
 pub struct Server {
     settings: hyper::server::conn::Http,
-    expected_id: Option<tls::ConditionalServerId>,
     f: HandleFuture,
 }
 
@@ -18,7 +17,6 @@ impl Default for Server {
     fn default() -> Self {
         Self {
             settings: hyper::server::conn::Http::new(),
-            expected_id: None,
             f: Box::new(|_| {
                 Ok(Response::builder()
                     .status(http::status::StatusCode::NOT_FOUND)
@@ -53,16 +51,6 @@ impl Server {
         self
     }
 
-    pub fn expect_identity(mut self, id: impl Into<id::Name>) -> Self {
-        self.expected_id = Some(Conditional::Some(tls::ServerId(id.into())));
-        self
-    }
-
-    pub fn no_identity(mut self, reason: tls::NoServerId) -> Self {
-        self.expected_id = Some(Conditional::None(reason));
-        self
-    }
-
     pub fn new(mut f: impl (FnMut(Request<Body>) -> Response<Body>) + Send + 'static) -> Self {
         Self {
             f: Box::new(move |req| Ok::<_, Error>(f(req))),
@@ -75,19 +63,11 @@ impl Server {
         E: std::fmt::Debug,
         for<'e> &'e E: Into<tls::ConditionalServerId>,
     {
-        let Self {
-            f,
-            settings,
-            expected_id,
-        } = self;
+        let Self { f, settings } = self;
         let f = Arc::new(Mutex::new(f));
         move |endpoint| {
             let span = tracing::debug_span!("server::run", ?endpoint);
             let _e = span.enter();
-            if let Some(expected_id) = expected_id.as_ref() {
-                let server_id: tls::ConditionalServerId = (&endpoint).into();
-                assert_eq!(&server_id, expected_id);
-            }
             let f = f.clone();
             let (client_io, server_io) = crate::io::duplex(4096);
             let svc = hyper::service::service_fn(move |request: Request<Body>| {

linkerd/proxy/identity/src/certify.rs

@@ -192,6 +192,10 @@ impl LocalCrtKey {
         crate::metrics::Report::new(self.crt_key.clone(), self.refreshes.clone())
     }
 
+    pub fn id(&self) -> &id::LocalId {
+        &self.id
+    }
+
     pub fn name(&self) -> &id::Name {
         self.id.as_ref()
     }
@@ -209,7 +213,7 @@ impl Into<tls::client::Config> for &'_ LocalCrtKey {
 
 impl Into<id::LocalId> for &'_ LocalCrtKey {
     fn into(self) -> id::LocalId {
-        id::LocalId(self.name().clone())
+        self.id().clone()
     }
 }