chore(app,outbound)!: Decouple metrics registry from stack building (#2887)

As we introduced the newer prometheus-client metrics registry, we did so by
allowing stacks to register metrics directly. This is incongruent with the
existing legacy metrics registries, which are known by the Inbound and Outbound
stack builders.

This leads to two problems:

1. We cannot build the admin server until all of the proxy stacks are built.
   This ordering dependency is unnecessary and cumbersome if we want to insert
   additional discovery work into the initialization process.
2. It is cumbersome to add new metrics to stacks, as the registry must be passed
   through the stack building process.

To fix this, this change introduces additional Metrics types so that these
metrics may be registered along with the other stack metrics.

BREAKING CHANGE: Gateway-mode proxies now report Balancer metrics with the
"outbound_" prefix instead of the "gateway_" prefix. This metrics scope was
introduced very recently and is incongruent with our other metrics export. We
have no known readers of these metrics, as they are only relevant to new load
balancer behavior, and only on multicluster gateways. Unifying under the
outbound_ prefix reduces noise in metrics in non-gateway use cases and makes it
easier to query proxy data consistently.

This is the only functional change in this commit.

Author: olix0r

Cargo.lock

@@ -2255,6 +2255,7 @@ dependencies = [
  "jemallocator",
  "linkerd-app",
  "linkerd-meshtls",
+ "linkerd-metrics",
  "linkerd-signal",
  "num_cpus",
  "tokio",

linkerd/app/core/src/control.rs

@@ -72,17 +72,30 @@ impl fmt::Display for ControlAddr {
 pub type RspBody =
     linkerd_http_metrics::requests::ResponseBody<http::balance::Body<hyper::Body>, classify::Eos>;
 
+#[derive(Clone, Debug, Default)]
+pub struct Metrics {
+    balance: balance::Metrics,
+}
+
 const EWMA_CONFIG: http::balance::EwmaConfig = http::balance::EwmaConfig {
     default_rtt: time::Duration::from_millis(30),
     decay: time::Duration::from_secs(10),
 };
 
+impl Metrics {
+    pub fn register(registry: &mut prom::Registry) -> Self {
+        Metrics {
+            balance: balance::Metrics::register(registry.sub_registry_with_prefix("balancer")),
+        }
+    }
+}
+
 impl Config {
     pub fn build(
         self,
         dns: dns::Resolver,
-        metrics: metrics::ControlHttp,
-        registry: &mut prom::Registry,
+        legacy_metrics: metrics::ControlHttp,
+        metrics: Metrics,
         identity: identity::NewClient,
     ) -> svc::ArcNewService<
         (),
@@ -133,12 +146,8 @@ impl Config {
 
         let balance = endpoint
             .lift_new()
-            .push(self::balance::layer(
-                registry.sub_registry_with_prefix("balancer"),
-                dns,
-                resolve_backoff,
-            ))
-            .push(metrics.to_layer::<classify::Response, _, _>())
+            .push(self::balance::layer(metrics.balance, dns, resolve_backoff))
+            .push(legacy_metrics.to_layer::<classify::Response, _, _>())
             .push(classify::NewClassify::layer_default());
 
         balance
@@ -233,25 +242,30 @@ mod balance {
     use super::{client::Target, ControlAddr};
     use crate::{
         dns,
-        metrics::prom::{self, encoding::EncodeLabelSet},
+        metrics::prom::encoding::EncodeLabelSet,
         proxy::{dns_resolve::DnsResolve, http, resolve::recover},
         svc, tls,
     };
     use linkerd_stack::ExtractParam;
     use std::net::SocketAddr;
 
+    pub(super) type Metrics = http::balance::MetricFamilies<Labels>;
+
     pub fn layer<B, R: Clone, N>(
-        registry: &mut prom::Registry,
+        metrics: Metrics,
         dns: dns::Resolver,
         recover: R,
     ) -> impl svc::Layer<
         N,
         Service = http::NewBalance<B, Params, recover::Resolve<R, DnsResolve>, NewIntoTarget<N>>,
     > {
         let resolve = recover::Resolve::new(recover, DnsResolve::new(dns));
-        let metrics = Params(http::balance::MetricFamilies::register(registry));
         svc::layer::mk(move |inner| {
-            http::NewBalance::new(NewIntoTarget { inner }, resolve.clone(), metrics.clone())
+            http::NewBalance::new(
+                NewIntoTarget { inner },
+                resolve.clone(),
+                Params(metrics.clone()),
+            )
         })
     }
 
@@ -270,7 +284,7 @@ mod balance {
     }
 
     #[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
-    struct Labels {
+    pub(super) struct Labels {
         addr: String,
     }
 

linkerd/app/gateway/src/http.rs

@@ -1,7 +1,7 @@
 use super::Gateway;
 use inbound::{GatewayAddr, GatewayDomainInvalid};
 use linkerd_app_core::{
-    metrics::{prom, ServerLabel},
+    metrics::ServerLabel,
     profiles,
     proxy::{
         api_resolve::{ConcreteAddr, Metadata},
@@ -47,7 +47,6 @@ impl Gateway {
     /// outbound router.
     pub fn http<T, R>(
         &self,
-        registry: &mut prom::Registry,
         inner: svc::ArcNewHttp<
             outbound::http::concrete::Endpoint<
                 outbound::http::logical::Concrete<outbound::http::Http<Target>>,
@@ -86,7 +85,7 @@ impl Gateway {
             .outbound
             .clone()
             .with_stack(inner)
-            .push_http_cached(outbound::http::HttpMetrics::register(registry), resolve)
+            .push_http_cached(resolve)
             .into_stack()
             // Discard `T` and its associated client-specific metadata.
             .push_map_target(Target::discard_parent)

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

@@ -160,7 +160,6 @@ async fn upgraded_request_remains_relative_form() {
         );
         gateway
             .http(
-                &mut Default::default(),
                 svc::ArcNewHttp::new(move |_: _| svc::BoxHttp::new(inner.clone())),
                 resolve,
             )

linkerd/app/gateway/src/lib.rs

@@ -3,9 +3,7 @@
 #![forbid(unsafe_code)]
 
 use linkerd_app_core::{
-    io,
-    metrics::prom,
-    profiles,
+    io, profiles,
     proxy::{
         api_resolve::{ConcreteAddr, Metadata},
         core::Resolve,
@@ -50,7 +48,6 @@ impl Gateway {
     /// stack.
     pub fn stack<T, I, R>(
         self,
-        registry: &mut prom::Registry,
         resolve: R,
         profiles: impl profiles::GetProfile<Error = Error>,
         policies: impl outbound::policy::GetPolicy,
@@ -73,12 +70,8 @@ impl Gateway {
         R::Resolution: Unpin,
     {
         let opaq = {
-            let registry = registry.sub_registry_with_prefix("tcp");
             let resolve = resolve.clone();
-            let opaq = self
-                .outbound
-                .to_tcp_connect()
-                .push_opaq_cached(registry, resolve);
+            let opaq = self.outbound.to_tcp_connect().push_opaq_cached(resolve);
             self.opaq(opaq.into_inner()).into_inner()
         };
 
@@ -88,7 +81,7 @@ impl Gateway {
                 .to_tcp_connect()
                 .push_tcp_endpoint()
                 .push_http_tcp_client();
-            let http = self.http(registry, http.into_inner(), resolve);
+            let http = self.http(http.into_inner(), resolve);
             self.inbound
                 .clone()
                 .with_stack(http.into_inner())

linkerd/app/integration/src/proxy.rs

@@ -457,7 +457,14 @@ async fn run(proxy: Proxy, mut env: TestEnv, random_ports: bool) -> Listening {
                         let bind_adm = listen::BindTcp::default();
                         let (shutdown_tx, mut shutdown_rx) = tokio::sync::mpsc::unbounded_channel();
                         let main = config
-                            .build(bind_in, bind_out, bind_adm, shutdown_tx, trace_handle)
+                            .build(
+                                bind_in,
+                                bind_out,
+                                bind_adm,
+                                shutdown_tx,
+                                trace_handle,
+                                Default::default(),
+                            )
                             .await
                             .expect("config");
 

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

@@ -37,7 +37,7 @@ async fn errors_propagate() {
 
     // Create a profile stack that uses the tracked inner stack.
     let (rt, _shutdown) = runtime();
-    let stack = Outbound::new(default_config(), rt)
+    let stack = Outbound::new(default_config(), rt, &mut Default::default())
         .with_stack(stack)
         .push_discover(discover)
         .into_inner();
@@ -107,7 +107,7 @@ async fn caches_profiles_until_idle() {
         cfg
     };
     let (rt, _shutdown) = runtime();
-    let stack = Outbound::new(cfg, rt)
+    let stack = Outbound::new(cfg, rt, &mut Default::default())
         .with_stack(stack)
         .push_discover(discover)
         .into_inner();

linkerd/app/outbound/src/http.rs

@@ -86,11 +86,7 @@ impl<T> Outbound<svc::ArcNewHttp<concrete::Endpoint<logical::Concrete<Http<T>>>>
     /// Builds a stack that routes HTTP requests to endpoint stacks.
     ///
     /// Buffered concrete services are cached in and evicted when idle.
-    pub fn push_http_cached<R>(
-        self,
-        metrics: HttpMetrics,
-        resolve: R,
-    ) -> Outbound<svc::ArcNewCloneHttp<T>>
+    pub fn push_http_cached<R>(self, resolve: R) -> Outbound<svc::ArcNewCloneHttp<T>>
     where
         // Logical HTTP target.
         T: svc::Param<http::Version>,
@@ -101,8 +97,8 @@ impl<T> Outbound<svc::ArcNewHttp<concrete::Endpoint<logical::Concrete<Http<T>>>>
         R::Resolution: Unpin,
     {
         self.push_http_endpoint()
-            .push_http_concrete(metrics.balancer, resolve)
-            .push_http_logical(metrics.http_route, metrics.grpc_route)
+            .push_http_concrete(resolve)
+            .push_http_logical()
             .map_stack(move |config, _, stk| {
                 stk.push_new_idle_cached(config.discovery_idle_timeout)
                     .push_map_target(Http)

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

@@ -60,11 +60,7 @@ impl<N> Outbound<N> {
     /// 'failfast'. While in failfast, buffered requests are failed and the
     /// service becomes unavailable so callers may choose alternate concrete
     /// services.
-    pub fn push_http_concrete<T, NSvc, R>(
-        self,
-        balancer_metrics: balance::BalancerMetrics,
-        resolve: R,
-    ) -> Outbound<svc::ArcNewCloneHttp<T>>
+    pub fn push_http_concrete<T, NSvc, R>(self, resolve: R) -> Outbound<svc::ArcNewCloneHttp<T>>
     where
         // Concrete target type.
         T: svc::Param<ParentRef>,
@@ -105,12 +101,7 @@ impl<N> Outbound<N> {
             });
 
             inner
-                .push(balance::Balance::layer(
-                    config,
-                    rt,
-                    balancer_metrics,
-                    resolve,
-                ))
+                .push(balance::Balance::layer(config, rt, resolve))
                 .check_new_clone()
                 .push_switch(Ok::<_, Infallible>, forward.into_inner())
                 .push_switch(

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

@@ -82,7 +82,6 @@ where
     pub(super) fn layer<N, NSvc, R>(
         config: &crate::Config,
         rt: &crate::Runtime,
-        metrics: BalancerMetrics,
         resolve: R,
     ) -> impl svc::Layer<N, Service = svc::ArcNewCloneHttp<Self>> + Clone
     where
@@ -101,6 +100,7 @@ where
         let http_queue = config.http_request_queue;
         let inbound_ips = config.inbound_ips.clone();
         let stack_metrics = rt.metrics.proxy.stack.clone();
+        let balance_metrics = rt.metrics.prom.http.balancer.clone();
 
         let resolve = svc::stack(resolve.into_service())
             .push_map_target(|t: Self| ConcreteAddr(t.addr))
@@ -148,7 +148,10 @@ where
                 .push(svc::ArcNewService::layer());
 
             endpoint
-                .push(http::NewBalance::layer(resolve.clone(), metrics.clone()))
+                .push(http::NewBalance::layer(
+                    resolve.clone(),
+                    balance_metrics.clone(),
+                ))
                 .push_on_service(http::BoxResponse::layer())
                 .push_on_service(stack_metrics.layer(stack_labels("http", "balance")))
                 .push(svc::NewMapErr::layer_from_target::<BalanceError, _>())