Extract zone locality into separate TCP metric

Currently, the destination controller populates a zone locality field in the metric labels for an endpoint. This works fine, but can be expensive to query against in prometheus due to the high cardinality of the metrics it is attached to. This introduces a separate `outbound_tcp_bytes_total` metric with only the zone locality as a label, which is the most common use of this label. This includes some extra infrastructure to wire the new metric through as we currently don't use prometheus-client for transport metrics. Eventually, the existing metrics could and/or should be migrated, but that is saved for a future change. Signed-off-by: Scott Fleener <[email protected]>
linkerd · Oct 9, 2024 · 4e359ae · 4e359ae
1 parent 53b528a
commit 4e359ae
Show file tree

Hide file tree

Showing 16 changed files with 350 additions and 12 deletions.
diff --git a/linkerd/app/core/src/metrics.rs b/linkerd/app/core/src/metrics.rs
@@ -9,14 +9,13 @@
 pub use crate::transport::labels::{TargetAddr, TlsAccept};
 use crate::{
     classify::Class,
-    control, http_metrics, opencensus, opentelemetry, profiles, stack_metrics,
-    svc::Param,
-    tls,
+    control, http_metrics, opencensus, opentelemetry, profiles, proxy, stack_metrics, svc, tls,
     transport::{self, labels::TlsConnect},
 };
 use linkerd_addr::Addr;
 pub use linkerd_metrics::*;
 use linkerd_proxy_server_policy as policy;
+use prometheus_client::encoding::EncodeLabelValue;
 use std::{
     fmt::{self, Write},
     net::SocketAddr,
@@ -102,9 +101,32 @@ pub struct OutboundEndpointLabels {
     pub server_id: tls::ConditionalClientTls,
     pub authority: Option<http::uri::Authority>,
     pub labels: Option<String>,
+    pub zone_locality: OutboundZoneLocality,
     pub target_addr: SocketAddr,
 }
 
+#[derive(Debug, Copy, Clone, Default, Hash, Eq, PartialEq, EncodeLabelValue)]
+pub enum OutboundZoneLocality {
+    #[default]
+    Unknown,
+    Local,
+    Remote,
+}
+
+impl OutboundZoneLocality {
+    pub fn new(metadata: &proxy::api_resolve::Metadata) -> Self {
+        if let Some(is_zone_local) = metadata.is_zone_local() {
+            if is_zone_local {
+                OutboundZoneLocality::Local
+            } else {
+                OutboundZoneLocality::Remote
+            }
+        } else {
+            OutboundZoneLocality::Unknown
+        }
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct StackLabels {
     pub direction: Direction,
@@ -217,7 +239,7 @@ impl Metrics {
 
 // === impl CtlLabels ===
 
-impl Param<ControlLabels> for control::ControlAddr {
+impl svc::Param<ControlLabels> for control::ControlAddr {
     fn param(&self) -> ControlLabels {
         ControlLabels {
             addr: self.addr.clone(),
@@ -361,6 +383,12 @@ impl FmtLabels for RouteAuthzLabels {
     }
 }
 
+impl svc::Param<OutboundZoneLocality> for OutboundEndpointLabels {
+    fn param(&self) -> OutboundZoneLocality {
+        self.zone_locality
+    }
+}
+
 impl FmtLabels for OutboundEndpointLabels {
     fn fmt_labels(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         if let Some(a) = self.authority.as_ref() {

diff --git a/linkerd/app/outbound/src/http/concrete.rs b/linkerd/app/outbound/src/http/concrete.rs
@@ -5,6 +5,7 @@ use super::{balance::EwmaConfig, client, handle_proxy_error_headers};
 use crate::{http, stack_labels, BackendRef, Outbound, ParentRef};
 use linkerd_app_core::{
     config::{ConnectConfig, QueueConfig},
+    metrics::OutboundZoneLocality,
     metrics::{prefix_labels, EndpointLabels, OutboundEndpointLabels},
     profiles,
     proxy::{
@@ -212,12 +213,19 @@ where
         OutboundEndpointLabels {
             authority: self.parent.param(),
             labels: prefix_labels("dst", self.metadata.labels().iter()),
+            zone_locality: self.param(),
             server_id: self.param(),
             target_addr: self.addr.into(),
         }
     }
 }
 
+impl<T> svc::Param<OutboundZoneLocality> for Endpoint<T> {
+    fn param(&self) -> OutboundZoneLocality {
+        OutboundZoneLocality::new(&self.metadata)
+    }
+}
+
 impl<T> svc::Param<EndpointLabels> for Endpoint<T>
 where
     T: svc::Param<Option<http::uri::Authority>>,

diff --git a/linkerd/app/outbound/src/http/endpoint.rs b/linkerd/app/outbound/src/http/endpoint.rs
@@ -7,6 +7,7 @@ use super::{
 use crate::{tcp::tagged_transport, Outbound};
 use linkerd_app_core::{
     classify, config, errors, http_tracing, metrics,
+    metrics::OutboundZoneLocality,
     proxy::{api_resolve::ProtocolHint, http, tap},
     svc::{self, ExtractParam},
     tls,
@@ -238,6 +239,13 @@ impl<T: svc::Param<transport::labels::Key>> svc::Param<transport::labels::Key> f
     }
 }
 
+impl<T: svc::Param<OutboundZoneLocality>> svc::Param<OutboundZoneLocality> for Connect<T> {
+    #[inline]
+    fn param(&self) -> OutboundZoneLocality {
+        self.inner.param()
+    }
+}
+
 // === impl EndpointError ===
 
 impl<T> From<(&T, Error)> for EndpointError

diff --git a/linkerd/app/outbound/src/http/endpoint/tests.rs b/linkerd/app/outbound/src/http/endpoint/tests.rs
@@ -1,6 +1,7 @@
 use super::*;
 use crate::{http, tcp, test_util::*};
 use ::http::header::{CONNECTION, UPGRADE};
+use linkerd_app_core::metrics::OutboundZoneLocality;
 use linkerd_app_core::{
     io,
     proxy::api_resolve::ProtocolHint,
@@ -300,6 +301,7 @@ impl svc::Param<metrics::OutboundEndpointLabels> for Endpoint {
         metrics::OutboundEndpointLabels {
             authority: None,
             labels: None,
+            zone_locality: OutboundZoneLocality::Unknown,
             server_id: self.param(),
             target_addr: self.addr.into(),
         }

diff --git a/linkerd/app/outbound/src/lib.rs b/linkerd/app/outbound/src/lib.rs
@@ -46,6 +46,7 @@ mod sidecar;
 pub mod tcp;
 #[cfg(any(test, feature = "test-util"))]
 pub mod test_util;
+mod zone;
 
 pub use self::discover::{spawn_synthesized_profile_policy, synthesize_forward_policy, Discovery};
 use self::metrics::OutboundMetrics;

diff --git a/linkerd/app/outbound/src/metrics.rs b/linkerd/app/outbound/src/metrics.rs
@@ -37,6 +37,7 @@ pub struct OutboundMetrics {
 pub(crate) struct PromMetrics {
     pub(crate) http: crate::http::HttpMetrics,
     pub(crate) opaq: crate::opaq::OpaqMetrics,
+    pub(crate) zone: crate::zone::TcpZoneMetricsParams,
 }
 
 #[derive(Clone, Debug, Hash, PartialEq, Eq)]
@@ -47,7 +48,7 @@ pub struct BalancerMetricsParams<K>(balance::MetricFamilies<K>);
 
 struct ScopedKey<'a, 'b>(&'a str, &'b str);
 
-// === impl BalancerMetricsPaarams ===
+// === impl BalancerMetricsParams ===
 
 impl<K> BalancerMetricsParams<K>
 where
@@ -80,6 +81,7 @@ where
         Self(balance::MetricFamilies::default())
     }
 }
+
 // === impl PromMetrics ===
 
 impl PromMetrics {
@@ -89,8 +91,10 @@ impl PromMetrics {
         let http = crate::http::HttpMetrics::register(registry);
 
         let opaq = crate::opaq::OpaqMetrics::register(registry.sub_registry_with_prefix("tcp"));
+        let zone =
+            crate::zone::TcpZoneMetricsParams::register(registry.sub_registry_with_prefix("tcp"));
 
-        Self { http, opaq }
+        Self { http, opaq, zone }
     }
 }
 

diff --git a/linkerd/app/outbound/src/opaq/concrete.rs b/linkerd/app/outbound/src/opaq/concrete.rs
@@ -2,6 +2,7 @@ use crate::{metrics::BalancerMetricsParams, stack_labels, BackendRef, Outbound,
 use linkerd_app_core::{
     config::QueueConfig,
     drain, io,
+    metrics::OutboundZoneLocality,
     metrics::{
         self,
         prom::{self, EncodeLabelSetMut},
@@ -323,12 +324,19 @@ where
         metrics::OutboundEndpointLabels {
             authority,
             labels: metrics::prefix_labels("dst", self.metadata.labels().iter()),
+            zone_locality: self.param(),
             server_id: self.param(),
             target_addr: self.addr.into(),
         }
     }
 }
 
+impl<T> svc::Param<OutboundZoneLocality> for Endpoint<T> {
+    fn param(&self) -> OutboundZoneLocality {
+        OutboundZoneLocality::new(&self.metadata)
+    }
+}
+
 impl<T> svc::Param<metrics::EndpointLabels> for Endpoint<T>
 where
     T: svc::Param<Option<profiles::LogicalAddr>>,

diff --git a/linkerd/app/outbound/src/tcp.rs b/linkerd/app/outbound/src/tcp.rs
@@ -1,6 +1,9 @@
+pub use self::connect::Connect;
 use crate::Outbound;
 use linkerd_app_core::{
-    io, svc,
+    io,
+    metrics::OutboundZoneLocality,
+    svc,
     transport::{self, addrs::*, metrics},
     Error, Infallible,
 };
@@ -9,9 +12,6 @@ mod connect;
 mod endpoint;
 pub mod tagged_transport;
 
-pub use self::connect::Connect;
-pub use linkerd_app_core::proxy::tcp::Forward;
-
 #[derive(Copy, Clone, Debug, Eq, PartialEq, Hash)]
 pub struct Accept {
     orig_dst: OrigDstAddr,

diff --git a/linkerd/app/outbound/src/tcp/endpoint.rs b/linkerd/app/outbound/src/tcp/endpoint.rs
@@ -22,6 +22,7 @@ impl<C> Outbound<C> {
         T: svc::Param<Option<http::AuthorityOverride>>,
         T: svc::Param<Option<SessionProtocol>>,
         T: svc::Param<transport::labels::Key>,
+        T: svc::Param<OutboundZoneLocality>,
         // Connector stack.
         C: svc::MakeConnection<Connect, Metadata = Local<ClientAddr>, Error = io::Error>,
         C: Clone + Send + 'static,
@@ -45,6 +46,9 @@ impl<C> Outbound<C> {
                 .push(transport::metrics::Client::layer(
                     rt.metrics.proxy.transport.clone(),
                 ))
+                .push(transport::metrics::zone::client::ZoneMetricsClient::layer(
+                    rt.metrics.prom.zone.clone(),
+                ))
         })
     }
 }
diff --git a/linkerd/app/outbound/src/zone.rs b/linkerd/app/outbound/src/zone.rs
@@ -0,0 +1,79 @@
+use crate::metrics::prom;
+use linkerd_app_core::{
+    metrics::OutboundZoneLocality, svc, transport::metrics::zone::TcpZoneMetrics,
+};
+use prometheus_client::{
+    encoding::{EncodeLabelSet, EncodeLabelValue},
+    registry::Unit,
+};
+
+#[derive(Clone, Debug, Default)]
+pub struct TcpZoneMetricsParams {
+    transfer_cost: prom::Family<TcpZoneLabels, prom::Counter>,
+}
+
+impl TcpZoneMetricsParams {
+    pub fn register(registry: &mut prom::Registry) -> Self {
+        let transfer_cost = prom::Family::default();
+        registry.register_with_unit(
+            "transfer_cost",
+            "The total cost of data transferred in and out of this proxy in bytes",
+            Unit::Bytes,
+            transfer_cost.clone(),
+        );
+
+        TcpZoneMetricsParams { transfer_cost }
+    }
+
+    pub fn metrics(&self, zone_locality: OutboundZoneLocality) -> TcpZoneMetrics {
+        let recv_bytes = self
+            .transfer_cost
+            .get_or_create(&TcpZoneLabels::new_recv_labels(zone_locality))
+            .clone();
+        let send_bytes = self
+            .transfer_cost
+            .get_or_create(&TcpZoneLabels::new_send_labels(zone_locality))
+            .clone();
+        TcpZoneMetrics {
+            recv_bytes,
+            send_bytes,
+        }
+    }
+}
+
+#[derive(Debug, Copy, Clone, Hash, Eq, PartialEq, EncodeLabelSet)]
+pub struct TcpZoneLabels {
+    op: TcpZoneOp,
+    zone_locality: OutboundZoneLocality,
+}
+
+impl TcpZoneLabels {
+    fn new_send_labels(zone_locality: OutboundZoneLocality) -> Self {
+        Self {
+            op: TcpZoneOp::Send,
+            zone_locality,
+        }
+    }
+
+    fn new_recv_labels(zone_locality: OutboundZoneLocality) -> Self {
+        Self {
+            op: TcpZoneOp::Recv,
+            zone_locality,
+        }
+    }
+}
+
+#[derive(Debug, Copy, Clone, Hash, Eq, PartialEq, EncodeLabelValue)]
+enum TcpZoneOp {
+    Send,
+    Recv,
+}
+
+impl<T> svc::ExtractParam<TcpZoneMetrics, T> for TcpZoneMetricsParams
+where
+    T: svc::Param<OutboundZoneLocality>,
+{
+    fn extract_param(&self, target: &T) -> TcpZoneMetrics {
+        self.metrics(target.param())
+    }
+}
diff --git a/linkerd/proxy/api-resolve/src/metadata.rs b/linkerd/proxy/api-resolve/src/metadata.rs
@@ -27,6 +27,7 @@ pub struct Metadata {
     authority_override: Option<Authority>,
 
     http2: HTTP2ClientParams,
+    is_zone_local: Option<bool>,
 }
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
@@ -53,11 +54,13 @@ impl Default for Metadata {
             tagged_transport_port: None,
             protocol_hint: ProtocolHint::Unknown,
             http2: HTTP2ClientParams::default(),
+            is_zone_local: None,
         }
     }
 }
 
 impl Metadata {
+    #[allow(clippy::too_many_arguments)]
     pub(crate) fn new(
         labels: impl IntoIterator<Item = (String, String)>,
         protocol_hint: ProtocolHint,
@@ -66,6 +69,7 @@ impl Metadata {
         authority_override: Option<Authority>,
         weight: u32,
         http2: HTTP2ClientParams,
+        is_zone_local: Option<bool>,
     ) -> Self {
         Self {
             labels: labels.into_iter().collect::<BTreeMap<_, _>>().into(),
@@ -75,6 +79,7 @@ impl Metadata {
             authority_override,
             weight,
             http2,
+            is_zone_local,
         }
     }
 
@@ -87,6 +92,10 @@ impl Metadata {
         self.labels.clone()
     }
 
+    pub fn is_zone_local(&self) -> Option<bool> {
+        self.is_zone_local
+    }
+
     pub fn protocol_hint(&self) -> ProtocolHint {
         self.protocol_hint
     }