outbound: Configure balancers with service metadata (#2374)

The OutboundPolicies API includes resource references with its
responses. These references allow us to accurately identify an arbitrary
(Kubernetes style) resource by its group, kind, namespace, and name.

To support new metrics that include these resource coordinates as
labels, we need access to the parent and backend references from the
balancer stack.

This change makes the following changes to accommodate this:

1. We introduce `ParentRef`, `RouteRef`, `BackendRef`, and
   `EndpointRef` new-types to serve as explicit markers for
   different types of metadata we may encounter.
3. The profile stack is updated to parse out metadata from service names
   in the form <name>.<ns>.svc.*.
4. We also support discovering pod metadata from the endpoint labels
   `dst_pod` and `dst_namespace`.
5. When we can't infer any metadata from a profile response, we use an
   "unknown" metadata.
6. When using policy, we use control-plane provided metadata.

In practice, pod metadata won't be surfaced by the existing code path;
but we have a relatively simple path forward to using it.

This change splits the balancer stack into its own layer so it's
distinct from the rest of the concrete stack configuration.

There are only minor functional changes in this branch:

1. The `balance{addr=...}` tracing context is replace by
   `service{ns=..,name=...}` for clarity;
2. Errors returned from the balance stack are now rendered as:
   `Service {name}.{ns}: {error}`

Author: olix0r

Cargo.lock

@@ -1882,7 +1882,7 @@ dependencies = [
 [[package]]
 name = "linkerd2-proxy-api"
 version = "0.8.0"
-source = "git+https://github.com/linkerd/linkerd2-proxy-api?branch=main#afd13fc7d5a592acd31b4ba822f854e5590e9600"
+source = "git+https://github.com/linkerd/linkerd2-proxy-api?branch=main#ad750d839ceb18294406e0f118527122be56cf66"
 dependencies = [
  "h2",
  "http",

linkerd/app/integration/src/policy.rs

@@ -111,6 +111,9 @@ pub fn outbound_default(dst: impl ToString) -> outbound::OutboundPolicy {
     let dst = dst.to_string();
     let route = outbound_default_http_route(dst.clone());
     outbound::OutboundPolicy {
+        metadata: Some(api::meta::Metadata {
+            kind: Some(api::meta::metadata::Kind::Default("default".to_string())),
+        }),
         protocol: Some(outbound::ProxyProtocol {
             kind: Some(proxy_protocol::Kind::Detect(proxy_protocol::Detect {
                 timeout: Some(Duration::from_secs(10).try_into().unwrap()),

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

@@ -51,6 +51,9 @@ async fn empty_http1_route() {
         .outbound(
             srv.addr,
             outbound::OutboundPolicy {
+                metadata: Some(api::meta::Metadata {
+                    kind: Some(api::meta::metadata::Kind::Default("test".to_string())),
+                }),
                 protocol: Some(outbound::ProxyProtocol {
                     kind: Some(proxy_protocol::Kind::Detect(proxy_protocol::Detect {
                         timeout: Some(Duration::from_secs(10).try_into().unwrap()),
@@ -137,6 +140,9 @@ async fn empty_http2_route() {
         .outbound(
             srv.addr,
             outbound::OutboundPolicy {
+                metadata: Some(api::meta::Metadata {
+                    kind: Some(api::meta::metadata::Kind::Default("test".to_string())),
+                }),
                 protocol: Some(outbound::ProxyProtocol {
                     kind: Some(proxy_protocol::Kind::Detect(proxy_protocol::Detect {
                         timeout: Some(Duration::from_secs(10).try_into().unwrap()),
@@ -250,6 +256,9 @@ async fn header_based_routing() {
         .outbound(
             srv.addr,
             outbound::OutboundPolicy {
+                metadata: Some(api::meta::Metadata {
+                    kind: Some(api::meta::metadata::Kind::Default("test".to_string())),
+                }),
                 protocol: Some(outbound::ProxyProtocol {
                     kind: Some(proxy_protocol::Kind::Detect(proxy_protocol::Detect {
                         timeout: Some(Duration::from_secs(10).try_into().unwrap()),
@@ -427,6 +436,9 @@ async fn path_based_routing() {
         .outbound(
             srv.addr,
             outbound::OutboundPolicy {
+                metadata: Some(api::meta::Metadata {
+                    kind: Some(api::meta::metadata::Kind::Default("test".to_string())),
+                }),
                 protocol: Some(outbound::ProxyProtocol {
                     kind: Some(proxy_protocol::Kind::Detect(proxy_protocol::Detect {
                         timeout: Some(Duration::from_secs(10).try_into().unwrap()),
@@ -490,6 +502,7 @@ fn httproute_meta(name: impl ToString) -> api::meta::Metadata {
             name: name.to_string(),
             namespace: "test".to_string(),
             section: "".to_string(),
+            port: 0,
         })),
     }
 }

linkerd/app/outbound/src/discover.rs

@@ -248,6 +248,7 @@ pub fn synthesize_forward_policy(
     };
 
     ClientPolicy {
+        parent: meta.clone(),
         protocol: detect,
         backends: Arc::new([backend]),
     }

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

@@ -2,7 +2,7 @@
 //! and distributes HTTP requests among them.
 
 use super::{balance, breaker, client, handle_proxy_error_headers};
-use crate::{http, stack_labels, Outbound};
+use crate::{http, stack_labels, BackendRef, Outbound, ParentRef};
 use linkerd_app_core::{
     classify, metrics, profiles,
     proxy::{
@@ -34,9 +34,9 @@ pub struct DispatcherFailed(Arc<str>);
 
 /// Wraps errors encountered in this module.
 #[derive(Debug, thiserror::Error)]
-#[error("concrete service {addr}: {source}")]
-pub struct ConcreteError {
-    addr: NameAddr,
+#[error("{} {}.{}: {source}", backend.0.kind(), backend.0.name(), backend.0.namespace())]
+pub struct BalanceError {
+    backend: BackendRef,
     #[source]
     source: Error,
 }
@@ -87,9 +87,9 @@ impl<N> Outbound<N> {
     >
     where
         // Concrete target type.
+        T: svc::Param<ParentRef>,
+        T: svc::Param<BackendRef>,
         T: svc::Param<Dispatch>,
-        // TODO(ver) T: svc::Param<svc::queue::Capacity> + svc::Param<svc::queue::Timeout>,
-        // Failure accrual policy.
         T: svc::Param<FailureAccrual>,
         T: Clone + Debug + Send + Sync + 'static,
         // Endpoint resolution.
@@ -102,10 +102,6 @@ impl<N> Outbound<N> {
         NSvc::Error: Into<Error>,
         NSvc::Future: Send,
     {
-        let resolve =
-            svc::MapTargetLayer::new(|t: Balance<T>| -> ConcreteAddr { ConcreteAddr(t.addr) })
-                .layer(resolve.into_service());
-
         self.map_stack(|config, rt, inner| {
             let inbound_ips = config.inbound_ips.clone();
 
@@ -119,71 +115,15 @@ impl<N> Outbound<N> {
                 )
                 .instrument(|e: &Endpoint<T>| info_span!("forward", addr = %e.addr));
 
-            let endpoint = inner
-                .push_map_target({
-                    let inbound_ips = inbound_ips.clone();
-                    move |((addr, metadata), target): ((SocketAddr, Metadata), Balance<T>)| {
-                        tracing::trace!(%addr, ?metadata, ?target, "Resolved endpoint");
-                        let is_local = inbound_ips.contains(&addr.ip());
-                        Endpoint {
-                            addr: Remote(ServerAddr(addr)),
-                            metadata,
-                            is_local,
-                            parent: target.parent,
-                            // We don't close server-side connections when we
-                            // get `l5d-proxy-connection: close` response headers
-                            // going through the balancer.
-                            close_server_connection_on_remote_proxy_error: false,
-                        }
-                    }
-                })
-                .push_on_service(svc::MapErr::layer_boxed())
-                .lift_new_with_target()
-                .push(
-                    http::NewClassifyGateSet::<classify::Response, _, _, _>::layer_via({
-                        // This sets the capacity of the channel used to send
-                        // response classifications to the failure accrual task.
-                        // Since the number of messages in this channel should
-                        // be roughly the same as the number of requests, size
-                        // it to the request queue capacity.
-                        // TODO(eliza): when queue capacity is configured
-                        // per-target, extract this from the target instead.
-                        let channel_capacity = config.http_request_queue.capacity;
-                        move |target: &Balance<T>| breaker::Params {
-                            accrual: target.parent.param(),
-                            channel_capacity,
-                        }
-                    }),
-                )
-                .push_on_service(svc::OnServiceLayer::new(
-                    rt.metrics
-                        .proxy
-                        .stack
-                        .layer(stack_labels("http", "endpoint")),
-                ))
-                .push_on_service(svc::NewInstrumentLayer::new(
-                    |(addr, _): &(SocketAddr, _)| info_span!("endpoint", %addr),
-                ));
-
-            let balance = endpoint
-                .push(http::NewBalancePeakEwma::layer(resolve))
-                .push(svc::NewMapErr::layer_from_target::<ConcreteError, _>())
-                .push_on_service(http::BoxResponse::layer())
-                .push_on_service(
-                    rt.metrics
-                        .proxy
-                        .stack
-                        .layer(stack_labels("http", "balance")),
-                )
-                .instrument(|t: &Balance<T>| info_span!("balance", addr = %t.addr));
-
             let fail = svc::ArcNewService::new(|message: Arc<str>| {
                 svc::mk(move |_| {
                     let error = DispatcherFailed(message.clone());
                     futures::future::ready(Err(error))
                 })
             });
-            balance
+
+            inner
+                .push(Balance::layer(config, rt, resolve))
                 .push_switch(Ok::<_, Infallible>, forward.into_inner())
                 .push_switch(
                     move |parent: T| -> Result<_, Infallible> {
@@ -206,23 +146,14 @@ impl<N> Outbound<N> {
                     },
                     fail,
                 )
+                // TODO(ver) Configure this queue from the target (i.e. from
+                // discovery).
                 .push(svc::NewQueue::layer_via(config.http_request_queue))
                 .push(svc::ArcNewService::layer())
         })
     }
 }
 
-// === impl ConcreteError ===
-
-impl<T> From<(&Balance<T>, Error)> for ConcreteError {
-    fn from((target, source): (&Balance<T>, Error)) -> Self {
-        Self {
-            addr: target.addr.clone(),
-            source,
-        }
-    }
-}
-
 // === impl Balance ===
 
 impl<T> svc::Param<http::balance::EwmaConfig> for Balance<T> {
@@ -231,6 +162,117 @@ impl<T> svc::Param<http::balance::EwmaConfig> for Balance<T> {
     }
 }
 
+impl<T> Balance<T>
+where
+    // Parent target.
+    T: svc::Param<ParentRef>,
+    T: svc::Param<BackendRef>,
+    T: svc::Param<FailureAccrual>,
+    T: Clone + Debug + Send + Sync + 'static,
+{
+    pub fn layer<N, NSvc, R>(
+        config: &crate::Config,
+        rt: &crate::Runtime,
+        resolve: R,
+    ) -> impl svc::Layer<
+        N,
+        Service = svc::ArcNewService<
+            Self,
+            impl svc::Service<
+                http::Request<http::BoxBody>,
+                Response = http::Response<http::BoxBody>,
+                Error = BalanceError,
+                Future = impl Send,
+            >,
+        >,
+    > + Clone
+    where
+        // Endpoint resolution.
+        R: Resolve<ConcreteAddr, Error = Error, Endpoint = Metadata> + 'static,
+        // Endpoint stack.
+        N: svc::NewService<Endpoint<T>, Service = NSvc> + Clone + Send + Sync + 'static,
+        NSvc: svc::Service<http::Request<http::BoxBody>, Response = http::Response<http::BoxBody>>
+            + Send
+            + 'static,
+        NSvc::Error: Into<Error>,
+        NSvc::Future: Send,
+    {
+        let classify_channel_capacity = config.http_request_queue.capacity;
+        let inbound_ips = config.inbound_ips.clone();
+        let metrics = rt.metrics.clone();
+
+        let resolve = svc::MapTargetLayer::new(|t: Self| -> ConcreteAddr { ConcreteAddr(t.addr) })
+            .layer(resolve.into_service());
+
+        svc::layer::mk(move |inner: N| {
+            let endpoint = svc::stack(inner)
+                .push_map_target({
+                    let inbound_ips = inbound_ips.clone();
+                    move |((addr, metadata), target): ((SocketAddr, Metadata), Self)| {
+                        tracing::trace!(%addr, ?metadata, ?target, "Resolved endpoint");
+                        let is_local = inbound_ips.contains(&addr.ip());
+                        Endpoint {
+                            addr: Remote(ServerAddr(addr)),
+                            metadata,
+                            is_local,
+                            parent: target.parent,
+                            // We don't close server-side connections when we
+                            // get `l5d-proxy-connection: close` response headers
+                            // going through the balancer.
+                            close_server_connection_on_remote_proxy_error: false,
+                        }
+                    }
+                })
+                .push_on_service(svc::MapErr::layer_boxed())
+                .lift_new_with_target()
+                .push(
+                    http::NewClassifyGateSet::<classify::Response, _, _, _>::layer_via({
+                        move |target: &Self| breaker::Params {
+                            accrual: target.parent.param(),
+                            // TODO configure channel capacities from target.
+                            channel_capacity: classify_channel_capacity,
+                        }
+                    }),
+                )
+                .push_on_service(svc::OnServiceLayer::new(
+                    metrics.proxy.stack.layer(stack_labels("http", "endpoint")),
+                ))
+                .push_on_service(svc::NewInstrumentLayer::new(
+                    |(addr, _): &(SocketAddr, _)| info_span!("endpoint", %addr),
+                ));
+
+            endpoint
+                .push(http::NewBalancePeakEwma::layer(resolve.clone()))
+                .push(svc::NewMapErr::layer_from_target::<BalanceError, _>())
+                .push_on_service(http::BoxResponse::layer())
+                .push_on_service(metrics.proxy.stack.layer(stack_labels("http", "balance")))
+                .instrument(|t: &Self| {
+                    let BackendRef(meta) = t.parent.param();
+                    info_span!(
+                        "service",
+                        ns = %meta.namespace(),
+                        name = %meta.name(),
+                        port = %meta.port().map(u16::from).unwrap_or(0),
+                    )
+                })
+                .push(svc::ArcNewService::layer())
+                .into_inner()
+        })
+    }
+}
+
+// === impl BalanceError ===
+
+impl<T> From<(&Balance<T>, Error)> for BalanceError
+where
+    T: svc::Param<BackendRef>,
+{
+    fn from((target, source): (&Balance<T>, Error)) -> Self {
+        let backend = target.parent.param();
+        Self { backend, source }
+    }
+}
+
 // === impl Endpoint ===
 
 impl<T> svc::Param<Remote<ServerAddr>> for Endpoint<T> {