Extract zone locality into separate TCP metric

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 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 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 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() {

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>>,

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

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(),
         }

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;

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 }
     }
 }
 

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>>,

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,

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(),
+                ))
         })
     }
 }

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())
+    }
+}

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
     }