Merge branch 'ver/route-reqs-mtx'

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};
@@ -45,11 +45,17 @@ where
     // Parent target type.
     T: Debug + Eq + Hash,
     T: Clone + Send + Sync + 'static,
+    route::backend::ExtractMetrics:
+        svc::ExtractParam<route::backend::RequestCount, route::backend::Http<T>>,
+    // route::backend::ExtractMetrics:
+    //     svc::ExtractParam<route::backend::RequestCount, route::backend::Grpc<T>>,
 {
     /// 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 +80,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,19 @@
+use crate::RouteRef;
+
 use super::{super::Concrete, filters};
 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 +24,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 +65,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 +107,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 +129,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,94 @@
+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, Eq, Hash, PartialEq)]
+struct Labels(ParentRef, RouteRef, BackendRef);
+
+#[derive(Clone, Debug, Default)]
+pub struct RouteBackendMetrics {
+    http: Arc<Mutex<AHashMap<Labels, Arc<Counter>>>>,
+    grpc: Arc<Mutex<AHashMap<Labels, Arc<Counter>>>>,
+}
+
+// === 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 Self(parent, route, backend) = self;
+
+        Self::write_meta("parent", parent, f)?;
+        f.write_char(',')?;
+
+        Self::write_meta("route", route, f)?;
+        f.write_char(',')?;
+
+        Self::write_meta("backend", backend, f)?;
+        Ok(())
+    }
+}
+
+impl Labels {
+    fn write_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(())
+    }
+}