outbound: Report per-route-backend request count metrics (#2377)

When performing policy-based routing, proxies may dispatch requests
through per-route backend configurations. In order to illustrate how
routing rules apply and how backend distributions are being honored,
this change adds two new metrics:

* `outbound_http_route_backend_requests_total`; and
* `outbound_grpc_route_backend_requests_total`

Each of these metrics includes labels that identify a routes parent
(i.e. a Service), the route resource being used, and the backend
resource being used.

This implementation does NOT implement any form of metrics eviction for
these new metrics. This is tolerable for the short term as the
cardinality of services and routes is generally much less than the
cardinality of individual endpoints (where we do require
timeout/eviction for metrics).

Author: olix0r

Cargo.lock

@@ -1032,6 +1032,7 @@ dependencies = [
 name = "linkerd-app-outbound"
 version = "0.1.0"
 dependencies = [
+ "ahash",
  "bytes",
  "futures",
  "http",

linkerd/app/outbound/Cargo.toml

@@ -16,6 +16,7 @@ test-subscriber = []
 test-util = ["linkerd-app-test", "linkerd-meshtls-rustls/test-util"]
 
 [dependencies]
+ahash = "0.8"
 bytes = "1"
 http = "0.2"
 futures = { version = "0.3", default-features = false }

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

@@ -162,7 +162,8 @@ where
         S::Future: Send,
     {
         svc::layer::mk(move |concrete: N| {
-            let policy = svc::stack(concrete.clone()).push(policy::Policy::layer());
+            let policy = svc::stack(concrete.clone())
+                .push(policy::Policy::layer(metrics.http_route_backends.clone()));
             let profile =
                 svc::stack(concrete.clone()).push(profile::Params::layer(metrics.proxy.clone()));
             svc::stack(concrete)

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

@@ -8,7 +8,7 @@ mod router;
 mod tests;
 
 pub use self::{
-    route::errors,
+    route::{backend::RouteBackendMetrics, errors},
     router::{GrpcParams, HttpParams},
 };
 pub use linkerd_proxy_client_policy::{ClientPolicy, FailureAccrual};
@@ -49,7 +49,9 @@ where
     /// Builds a stack that dynamically updates and applies HTTP or gRPC policy
     /// routing configurations to route requests over cached inner backend
     /// services.
-    pub(super) fn layer<N, S>() -> impl svc::Layer<
+    pub(super) fn layer<N, S>(
+        route_backend_metrics: RouteBackendMetrics,
+    ) -> impl svc::Layer<
         N,
         Service = svc::ArcNewService<
             Self,
@@ -74,8 +76,9 @@ where
         S::Future: Send,
     {
         svc::layer::mk(move |inner: N| {
-            let http = svc::stack(inner.clone()).push(router::Http::layer());
-            let grpc = svc::stack(inner).push(router::Grpc::layer());
+            let http =
+                svc::stack(inner.clone()).push(router::Http::layer(route_backend_metrics.clone()));
+            let grpc = svc::stack(inner).push(router::Grpc::layer(route_backend_metrics.clone()));
 
             http.push_switch(
                 |pp: Policy<T>| {

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

@@ -1,4 +1,6 @@
-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;
@@ -25,7 +27,7 @@ pub(crate) struct Matched<M, P> {
 pub(crate) struct Route<T, F, E> {
     pub(super) parent: T,
     pub(super) addr: Addr,
-    pub(super) meta: Arc<policy::Meta>,
+    pub(super) route_ref: RouteRef,
     pub(super) filters: Arc<[F]>,
     pub(super) distribution: BackendDistribution<T, F>,
     pub(super) failure_policy: E,
@@ -66,11 +68,14 @@ where
     Self: filters::Apply,
     Self: svc::Param<classify::Request>,
     MatchedBackend<T, M, F>: filters::Apply,
+    backend::ExtractMetrics: svc::ExtractParam<backend::RequestCount, MatchedBackend<T, M, F>>,
 {
     /// Builds a route stack that applies policy filters to requests and
     /// 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>() -> impl svc::Layer<
+    pub(crate) fn layer<N, S>(
+        backend_metrics: RouteBackendMetrics,
+    ) -> impl svc::Layer<
         N,
         Service = svc::ArcNewService<
             Self,
@@ -98,7 +103,7 @@ where
             svc::stack(inner)
                 // Distribute requests across route backends, applying policies
                 // and filters for each of the route-backends.
-                .push(MatchedBackend::layer())
+                .push(MatchedBackend::layer(backend_metrics.clone()))
                 .lift_new_with_target()
                 .push(NewDistribute::layer())
                 // The router does not take the backend's availability into

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

@@ -1,11 +1,18 @@
 use super::{super::Concrete, filters};
+use crate::RouteRef;
 use linkerd_app_core::{proxy::http, svc, Error, Result};
 use linkerd_http_route as http_route;
 use linkerd_proxy_client_policy as policy;
 use std::{fmt::Debug, hash::Hash, sync::Arc};
 
+mod count_reqs;
+mod metrics;
+
+pub use self::{count_reqs::RequestCount, metrics::RouteBackendMetrics};
+
 #[derive(Debug, PartialEq, Eq, Hash)]
 pub(crate) struct Backend<T, F> {
+    pub(crate) route_ref: RouteRef,
     pub(crate) concrete: Concrete<T>,
     pub(crate) filters: Arc<[F]>,
 }
@@ -16,11 +23,17 @@ pub(crate) type Http<T> =
 pub(crate) type Grpc<T> =
     MatchedBackend<T, http_route::grpc::r#match::RouteMatch, policy::grpc::Filter>;
 
+#[derive(Clone, Debug)]
+pub struct ExtractMetrics {
+    metrics: RouteBackendMetrics,
+}
+
 // === impl Backend ===
 
 impl<T: Clone, F> Clone for Backend<T, F> {
     fn clone(&self) -> Self {
         Self {
+            route_ref: self.route_ref.clone(),
             filters: self.filters.clone(),
             concrete: self.concrete.clone(),
         }
@@ -51,13 +64,16 @@ where
     F: Clone + Send + Sync + 'static,
     // Assert that filters can be applied.
     Self: filters::Apply,
+    ExtractMetrics: svc::ExtractParam<RequestCount, Self>,
 {
     /// Builds a stack that applies per-route-backend policy filters over an
     /// inner [`Concrete`] stack.
     ///
     /// This [`MatchedBackend`] must implement [`filters::Apply`] to apply these
     /// filters.
-    pub(crate) fn layer<N, S>() -> impl svc::Layer<
+    pub(crate) fn layer<N, S>(
+        metrics: RouteBackendMetrics,
+    ) -> impl svc::Layer<
         N,
         Service = svc::ArcNewService<
             Self,
@@ -90,6 +106,9 @@ where
                      }| concrete,
                 )
                 .push(filters::NewApplyFilters::<Self, _, _>::layer())
+                .push(count_reqs::NewCountRequests::layer_via(ExtractMetrics {
+                    metrics: metrics.clone(),
+                }))
                 .push(svc::ArcNewService::layer())
                 .into_inner()
         })
@@ -109,3 +128,23 @@ impl<T> filters::Apply for Grpc<T> {
         filters::apply_grpc(&self.r#match, &self.params.filters, req)
     }
 }
+
+impl<T> svc::ExtractParam<RequestCount, Http<T>> for ExtractMetrics {
+    fn extract_param(&self, params: &Http<T>) -> RequestCount {
+        RequestCount(self.metrics.http_requests_total(
+            params.params.concrete.parent_ref.clone(),
+            params.params.route_ref.clone(),
+            params.params.concrete.backend_ref.clone(),
+        ))
+    }
+}
+
+impl<T> svc::ExtractParam<RequestCount, Grpc<T>> for ExtractMetrics {
+    fn extract_param(&self, params: &Grpc<T>) -> RequestCount {
+        RequestCount(self.metrics.grpc_requests_total(
+            params.params.concrete.parent_ref.clone(),
+            params.params.route_ref.clone(),
+            params.params.concrete.backend_ref.clone(),
+        ))
+    }
+}

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

@@ -0,0 +1,73 @@
+use linkerd_app_core::{metrics::Counter, svc};
+use std::{
+    sync::Arc,
+    task::{Context, Poll},
+};
+
+#[derive(Clone, Debug, Default)]
+pub struct RequestCount(pub Arc<Counter>);
+
+#[derive(Clone, Debug)]
+pub struct NewCountRequests<X, N> {
+    inner: N,
+    extract: X,
+}
+
+#[derive(Clone, Debug)]
+pub struct CountRequests<S> {
+    inner: S,
+    requests: Arc<Counter>,
+}
+
+// === impl NewCountRequests ===
+
+impl<X: Clone, N> NewCountRequests<X, N> {
+    pub fn new(extract: X, inner: N) -> Self {
+        Self { extract, inner }
+    }
+
+    pub fn layer_via(extract: X) -> impl svc::Layer<N, Service = Self> + Clone {
+        svc::layer::mk(move |inner| Self::new(extract.clone(), inner))
+    }
+}
+
+impl<T, X, N> svc::NewService<T> for NewCountRequests<X, N>
+where
+    X: svc::ExtractParam<RequestCount, T>,
+    N: svc::NewService<T>,
+{
+    type Service = CountRequests<N::Service>;
+
+    fn new_service(&self, target: T) -> Self::Service {
+        let RequestCount(counter) = self.extract.extract_param(&target);
+        let inner = self.inner.new_service(target);
+        CountRequests::new(counter, inner)
+    }
+}
+
+// === impl CountRequests ===
+
+impl<S> CountRequests<S> {
+    fn new(requests: Arc<Counter>, inner: S) -> Self {
+        Self { requests, inner }
+    }
+}
+
+impl<B, S> svc::Service<http::Request<B>> for CountRequests<S>
+where
+    S: svc::Service<http::Request<B>>,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = S::Future;
+
+    #[inline]
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), S::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: http::Request<B>) -> Self::Future {
+        self.requests.incr();
+        self.inner.call(req)
+    }
+}

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

@@ -0,0 +1,110 @@
+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"
+    },
+    outbound_grpc_route_backend_requests_total: Counter {
+        "The total number of outbound requests dispatched to a gRPC route backend"
+    }
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct RouteBackendMetrics {
+    http: Arc<Mutex<AHashMap<Labels, Arc<Counter>>>>,
+    grpc: Arc<Mutex<AHashMap<Labels, Arc<Counter>>>>,
+}
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+struct Labels(ParentRef, RouteRef, BackendRef);
+
+// === impl RouteBackendMetrics ===
+
+impl RouteBackendMetrics {
+    pub fn http_requests_total(&self, pr: ParentRef, rr: RouteRef, br: BackendRef) -> Arc<Counter> {
+        self.http
+            .lock()
+            .entry(Labels(pr, rr, br))
+            .or_default()
+            .clone()
+    }
+
+    pub fn grpc_requests_total(&self, pr: ParentRef, rr: RouteRef, br: BackendRef) -> Arc<Counter> {
+        self.grpc
+            .lock()
+            .entry(Labels(pr, rr, br))
+            .or_default()
+            .clone()
+    }
+}
+
+impl FmtMetrics for RouteBackendMetrics {
+    fn fmt_metrics(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let http = self.http.lock();
+        if !http.is_empty() {
+            outbound_http_route_backend_requests_total.fmt_help(f)?;
+            outbound_http_route_backend_requests_total.fmt_scopes(f, http.iter(), |c| c)?;
+        }
+        drop(http);
+
+        let grpc = self.grpc.lock();
+        if !grpc.is_empty() {
+            outbound_grpc_route_backend_requests_total.fmt_help(f)?;
+            outbound_grpc_route_backend_requests_total.fmt_scopes(f, grpc.iter(), |c| c)?;
+        }
+        drop(grpc);
+
+        Ok(())
+    }
+}
+
+impl FmtLabels for Labels {
+    fn fmt_labels(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let Labels(parent, route, backend) = self;
+
+        Self::write_extended_meta("parent", parent, f)?;
+        f.write_char(',')?;
+        Self::write_basic_meta("route", route, f)?;
+        f.write_char(',')?;
+        Self::write_extended_meta("backend", backend, f)?;
+
+        Ok(())
+    }
+}
+
+impl Labels {
+    fn write_basic_meta(
+        scope: &str,
+        meta: &policy::Meta,
+        f: &mut std::fmt::Formatter<'_>,
+    ) -> std::fmt::Result {
+        write!(f, "{scope}_group=\"{}\"", meta.group())?;
+        write!(f, ",{scope}_kind=\"{}\"", meta.kind())?;
+        write!(f, ",{scope}_namespace=\"{}\"", meta.namespace())?;
+        write!(f, ",{scope}_name=\"{}\"", meta.name())?;
+
+        Ok(())
+    }
+
+    fn write_extended_meta(
+        scope: &str,
+        meta: &policy::Meta,
+        f: &mut std::fmt::Formatter<'_>,
+    ) -> std::fmt::Result {
+        Self::write_basic_meta(scope, meta, f)?;
+
+        match meta.port() {
+            Some(port) => write!(f, ",{scope}_port=\"{port}\"")?,
+            None => write!(f, ",{scope}_port=\"\"")?,
+        }
+        write!(f, ",{scope}_section_name=\"{}\"", meta.section())?;
+
+        Ok(())
+    }
+}