outbound: Report HTTP balancer endpoint gauges (#2380)

It's not currently possible to know how many endpoints are in a
balancer. This change adds an `outbound_http_balancer_endpoints` gauge
that exposes the number of endpoints a balancer has by their current
readiness status.

Note that in the concrete stack we do not currently differentiate
between gRPC and HTTP backends, so all balancers are exposed under this
single metric.

Author: olix0r

Cargo.lock

@@ -1438,6 +1438,7 @@ dependencies = [
  "futures-util",
  "indexmap",
  "linkerd-error",
+ "linkerd-metrics",
  "linkerd-proxy-core",
  "linkerd-stack",
  "pin-project",

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

@@ -1,9 +1,11 @@
-mod consecutive_failures;
-use consecutive_failures::ConsecutiveFailures;
 use linkerd_app_core::{classify, proxy::http::classify::gate, svc};
 use linkerd_proxy_client_policy::FailureAccrual;
 use tracing::{trace_span, Instrument};
 
+mod consecutive_failures;
+
+use self::consecutive_failures::ConsecutiveFailures;
+
 /// Params configuring a circuit breaker stack.
 #[derive(Copy, Clone, Debug)]
 pub(crate) struct Params {

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

@@ -4,7 +4,9 @@
 use super::{balance, breaker, client, handle_proxy_error_headers};
 use crate::{http, stack_labels, BackendRef, Outbound, ParentRef};
 use linkerd_app_core::{
-    classify, metrics, profiles,
+    classify,
+    metrics::{prefix_labels, EndpointLabels, OutboundEndpointLabels},
+    profiles,
     proxy::{
         api_resolve::{ConcreteAddr, Metadata, ProtocolHint},
         core::Resolve,
@@ -19,6 +21,12 @@ use linkerd_proxy_client_policy::FailureAccrual;
 use std::{fmt::Debug, net::SocketAddr, sync::Arc};
 use tracing::info_span;
 
+mod metrics;
+#[cfg(test)]
+mod tests;
+
+pub use self::metrics::BalancerMetrics;
+
 /// Parameter configuring dispatcher behavior.
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub enum Dispatch {
@@ -182,7 +190,9 @@ where
                 http::Request<http::BoxBody>,
                 Response = http::Response<http::BoxBody>,
                 Error = BalanceError,
-                Future = impl Send,
+                Future = impl std::future::Future<
+                    Output = Result<http::Response<http::BoxBody>, BalanceError>,
+                > + Send,
             >,
         >,
     > + Clone
@@ -234,6 +244,12 @@ where
                         }
                     }),
                 )
+                .push(balance::NewGaugeEndpoints::layer_via({
+                    let metrics = metrics.http_balancer.clone();
+                    move |target: &Self| {
+                        metrics.http_endpoints(target.parent.param(), target.parent.param())
+                    }
+                }))
                 .push_on_service(svc::OnServiceLayer::new(
                     metrics.proxy.stack.layer(stack_labels("http", "endpoint")),
                 ))
@@ -321,26 +337,26 @@ where
     }
 }
 
-impl<T> svc::Param<metrics::OutboundEndpointLabels> for Endpoint<T>
+impl<T> svc::Param<OutboundEndpointLabels> for Endpoint<T>
 where
     T: svc::Param<Option<http::uri::Authority>>,
 {
-    fn param(&self) -> metrics::OutboundEndpointLabels {
-        metrics::OutboundEndpointLabels {
+    fn param(&self) -> OutboundEndpointLabels {
+        OutboundEndpointLabels {
             authority: self.parent.param(),
-            labels: metrics::prefix_labels("dst", self.metadata.labels().iter()),
+            labels: prefix_labels("dst", self.metadata.labels().iter()),
             server_id: self.param(),
             target_addr: self.addr.into(),
         }
     }
 }
 
-impl<T> svc::Param<metrics::EndpointLabels> for Endpoint<T>
+impl<T> svc::Param<EndpointLabels> for Endpoint<T>
 where
     T: svc::Param<Option<http::uri::Authority>>,
 {
-    fn param(&self) -> metrics::EndpointLabels {
-        metrics::EndpointLabels::Outbound(self.param())
+    fn param(&self) -> EndpointLabels {
+        EndpointLabels::Outbound(self.param())
     }
 }
 

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

@@ -0,0 +1,85 @@
+use crate::{BackendRef, ParentRef};
+use ahash::AHashMap;
+use linkerd_app_core::{
+    metrics::{metrics, FmtLabels, FmtMetrics, Gauge},
+    svc::http::balance,
+};
+use parking_lot::Mutex;
+use std::{fmt::Write, sync::Arc};
+
+metrics! {
+    outbound_http_balancer_endpoints: Gauge {
+        "The number of endpoints currently in a HTTP request balancer"
+    }
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct BalancerMetrics {
+    balancers: Arc<Mutex<AHashMap<Labels, balance::EndpointsGauges>>>,
+}
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+struct Labels(ParentRef, BackendRef);
+
+struct Ready<'l>(&'l Labels);
+struct Pending<'l>(&'l Labels);
+
+// === impl RouteBackendMetrics ===
+
+impl BalancerMetrics {
+    pub(super) fn http_endpoints(&self, pr: ParentRef, br: BackendRef) -> balance::EndpointsGauges {
+        self.balancers
+            .lock()
+            .entry(Labels(pr, br))
+            .or_default()
+            .clone()
+    }
+}
+
+impl FmtMetrics for BalancerMetrics {
+    fn fmt_metrics(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let balancers = self.balancers.lock();
+        if !balancers.is_empty() {
+            outbound_http_balancer_endpoints.fmt_help(f)?;
+            outbound_http_balancer_endpoints.fmt_scopes(
+                f,
+                balancers.iter().map(|(l, e)| (Pending(l), e)),
+                |c| &c.pending,
+            )?;
+            outbound_http_balancer_endpoints.fmt_scopes(
+                f,
+                balancers.iter().map(|(l, e)| (Ready(l), e)),
+                |c| &c.ready,
+            )?;
+        }
+        drop(balancers);
+
+        Ok(())
+    }
+}
+
+impl FmtLabels for Labels {
+    fn fmt_labels(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let Labels(parent, backend) = self;
+
+        crate::metrics::write_service_meta_labels("parent", parent, f)?;
+        f.write_char(',')?;
+        crate::metrics::write_service_meta_labels("backend", backend, f)?;
+
+        Ok(())
+    }
+}
+
+impl<'l> FmtLabels for Ready<'l> {
+    fn fmt_labels(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.0.fmt_labels(f)?;
+        write!(f, ",endpoint_state=\"ready\"")
+    }
+}
+
+impl<'l> FmtLabels for Pending<'l> {
+    fn fmt_labels(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.0.fmt_labels(f)?;
+        write!(f, ",endpoint_state=\"pending\"")
+    }
+}

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

@@ -0,0 +1,152 @@
+use super::*;
+use crate::test_util::*;
+use linkerd_app_core::{
+    svc::{http::balance::EwmaConfig, ServiceExt},
+    svc::{NewService, Service},
+    trace,
+};
+use linkerd_proxy_client_policy as policy;
+use std::{net::SocketAddr, num::NonZeroU16, sync::Arc};
+use tokio::time;
+
+#[tokio::test(flavor = "current_thread")]
+async fn gauges_endpoints() {
+    let _trace = trace::test::trace_init();
+    let (rt, _shutdown) = runtime();
+    let outbound = Outbound::new(default_config(), rt);
+
+    let addr = "mysvc.myns.svc.cluster.local:80"
+        .parse::<NameAddr>()
+        .unwrap();
+    let ep0 = SocketAddr::new([192, 0, 2, 41].into(), 8080);
+    let ep1 = SocketAddr::new([192, 0, 2, 42].into(), 8080);
+
+    let resolve = support::resolver::<Metadata>();
+    let mut resolve_tx = resolve.endpoint_tx(addr.clone());
+
+    let (svc0, mut handle0) = tower_test::mock::pair();
+    let (svc1, mut handle1) = tower_test::mock::pair();
+
+    let stk = move |ep: Endpoint<_>| {
+        if *ep.addr == ep0 {
+            return svc0.clone();
+        }
+        if *ep.addr == ep1 {
+            return svc1.clone();
+        }
+        panic!("unexpected endpoint: {:?}", ep)
+    };
+
+    let mut svc = svc::stack(stk)
+        .push(Balance::layer(&outbound.config, &outbound.runtime, resolve))
+        .into_inner()
+        .new_service(Balance {
+            addr,
+            parent: Target,
+            ewma: EwmaConfig {
+                default_rtt: time::Duration::from_millis(100),
+                decay: time::Duration::from_secs(10),
+            },
+        });
+
+    let ready = Arc::new(tokio::sync::Notify::new());
+    let _task = tokio::spawn({
+        let ready = ready.clone();
+        async move {
+            loop {
+                ready.notified().await;
+                svc.ready().await.unwrap();
+                svc.call(http::Request::default()).await.unwrap();
+            }
+        }
+    });
+
+    let gauge = outbound
+        .runtime
+        .metrics
+        .http_balancer
+        .http_endpoints(svc::Param::param(&Target), svc::Param::param(&Target));
+    assert_eq!(gauge.pending.value(), 0);
+    assert_eq!(gauge.ready.value(), 0);
+
+    // Begin with a single endpoint. When the balancer can process requests, the
+    // gauge is accurate.
+    resolve_tx.add(vec![(ep0, Metadata::default())]).unwrap();
+    handle0.allow(1);
+    ready.notify_one();
+    tokio::task::yield_now().await;
+    assert_eq!(gauge.pending.value(), 0);
+    assert_eq!(gauge.ready.value(), 1);
+    let (_, res) = handle0.next_request().await.unwrap();
+    res.send_response(http::Response::default());
+
+    // Add a second endpoint and ensure the gauge is updated.
+    resolve_tx.add(vec![(ep1, Metadata::default())]).unwrap();
+    handle0.allow(0);
+    handle1.allow(1);
+    ready.notify_one();
+    tokio::task::yield_now().await;
+    assert_eq!(gauge.pending.value(), 1);
+    assert_eq!(gauge.ready.value(), 1);
+    let (_, res) = handle1.next_request().await.unwrap();
+    res.send_response(http::Response::default());
+
+    // Remove the first endpoint.
+    resolve_tx.remove(vec![ep0]).unwrap();
+    handle1.allow(2);
+    ready.notify_one();
+    let (_, res) = handle1.next_request().await.unwrap();
+    res.send_response(http::Response::default());
+
+    // The inner endpoint isn't actually dropped until the balancer's subsequent poll.
+    ready.notify_one();
+    tokio::task::yield_now().await;
+    assert_eq!(gauge.pending.value(), 0);
+    assert_eq!(gauge.ready.value(), 1);
+    let (_, res) = handle1.next_request().await.unwrap();
+    res.send_response(http::Response::default());
+
+    // Dropping the remaining endpoint, the gauge is updated.
+    resolve_tx.remove(vec![ep1]).unwrap();
+    ready.notify_one();
+    tokio::task::yield_now().await;
+    assert_eq!(gauge.pending.value(), 0);
+    assert_eq!(gauge.ready.value(), 0);
+}
+
+#[derive(Clone, Debug)]
+struct Target;
+
+// === impl Target ===
+
+impl svc::Param<ParentRef> for Target {
+    fn param(&self) -> ParentRef {
+        ParentRef(Arc::new(policy::Meta::Resource {
+            group: "core".into(),
+            kind: "Service".into(),
+            namespace: "myns".into(),
+            name: "mysvc".into(),
+            port: NonZeroU16::new(80),
+            section: None,
+        }))
+    }
+}
+
+impl svc::Param<BackendRef> for Target {
+    fn param(&self) -> BackendRef {
+        BackendRef(Arc::new(policy::Meta::Resource {
+            group: "core".into(),
+            kind: "Service".into(),
+            namespace: "myns".into(),
+            name: "mysvc".into(),
+            port: NonZeroU16::new(80),
+            section: None,
+        }))
+    }
+}
+
+impl svc::Param<FailureAccrual> for Target {
+    fn param(&self) -> FailureAccrual {
+        FailureAccrual::default()
+    }
+}

linkerd/app/outbound/src/http/logical/policy/route.rs

@@ -1,6 +1,5 @@
+use super::super::Concrete;
 use crate::RouteRef;
-
-use super::{super::Concrete, RouteBackendMetrics};
 use linkerd_app_core::{classify, proxy::http, svc, Addr, Error, Result};
 use linkerd_distribute as distribute;
 use linkerd_http_route as http_route;
@@ -74,7 +73,7 @@ where
     /// distributes requests over each route's backends. These [`Concrete`]
     /// backends are expected to be cached/shared by the inner stack.
     pub(crate) fn layer<N, S>(
-        backend_metrics: RouteBackendMetrics,
+        backend_metrics: backend::RouteBackendMetrics,
     ) -> impl svc::Layer<
         N,
         Service = svc::ArcNewService<

linkerd/app/outbound/src/http/logical/policy/route/backend.rs

@@ -8,7 +8,8 @@ use std::{fmt::Debug, hash::Hash, sync::Arc};
 mod count_reqs;
 mod metrics;
 
-pub use self::{count_reqs::RequestCount, metrics::RouteBackendMetrics};
+pub use self::count_reqs::RequestCount;
+pub use self::metrics::RouteBackendMetrics;
 
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub(crate) struct Backend<T, F> {

linkerd/app/outbound/src/http/logical/policy/route/backend/metrics.rs

@@ -1,11 +1,10 @@
+use crate::{BackendRef, ParentRef, RouteRef};
 use ahash::AHashMap;
 use linkerd_app_core::metrics::{metrics, Counter, FmtLabels, FmtMetrics};
 use linkerd_proxy_client_policy as policy;
 use parking_lot::Mutex;
 use std::{fmt::Write, sync::Arc};
 
-use crate::{BackendRef, ParentRef, RouteRef};
-
 metrics! {
     outbound_http_route_backend_requests_total: Counter {
         "The total number of outbound requests dispatched to a HTTP route backend"