Add a protocol label to stack metrics (#773)

Now that TCP stacks are getting more complex, it's worth differentiating
them with a protocol label. This change also renames 'source' stacks to
'server' stacks.

Author: olix0r

linkerd/app/core/src/metrics.rs

@@ -56,6 +56,7 @@ pub struct EndpointLabels {
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct StackLabels {
     pub direction: Direction,
+    pub protocol: &'static str,
     pub name: &'static str,
 }
 
@@ -297,24 +298,26 @@ impl FmtLabels for TlsId {
 // === impl StackLabels ===
 
 impl StackLabels {
-    pub fn inbound(name: &'static str) -> Self {
+    pub fn inbound(protocol: &'static str, name: &'static str) -> Self {
         Self {
-            direction: Direction::In,
             name,
+            protocol,
+            direction: Direction::In,
         }
     }
 
-    pub fn outbound(name: &'static str) -> Self {
+    pub fn outbound(protocol: &'static str, name: &'static str) -> Self {
         Self {
-            direction: Direction::Out,
             name,
+            protocol,
+            direction: Direction::Out,
         }
     }
 }
 
 impl FmtLabels for StackLabels {
     fn fmt_labels(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         self.direction.fmt_labels(f)?;
-        write!(f, ",name=\"{}\"", self.name)
+        write!(f, ",protocol=\"{}\",name=\"{}\"", self.protocol, self.name)
     }
 }

linkerd/app/inbound/src/lib.rs

@@ -280,7 +280,7 @@ impl Config {
                 svc::layers()
                     .push_failfast(dispatch_timeout)
                     .push_spawn_buffer(buffer_capacity)
-                    .push(metrics.stack.layer(stack_labels("logical"))),
+                    .push(metrics.stack.layer(stack_labels("http", "logical"))),
             )
             .push_cache(cache_max_idle_age)
             .push_on_response(
@@ -371,7 +371,7 @@ impl Config {
                     .push(TraceContext::layer(span_sink.map(|span_sink| {
                         SpanConverter::server(span_sink, trace_labels())
                     })))
-                    .push(metrics.stack.layer(stack_labels("source")))
+                    .push(metrics.stack.layer(stack_labels("http", "server")))
                     .box_http_request()
                     .box_http_response(),
             )
@@ -458,8 +458,8 @@ pub fn trace_labels() -> HashMap<String, String> {
     l
 }
 
-fn stack_labels(name: &'static str) -> metrics::StackLabels {
-    metrics::StackLabels::inbound(name)
+fn stack_labels(proto: &'static str, name: &'static str) -> metrics::StackLabels {
+    metrics::StackLabels::inbound(proto, name)
 }
 
 // === impl SkipByPort ===

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

@@ -55,7 +55,11 @@ where
         .push_on_response(
             svc::layers()
                 .push(svc::layer::mk(svc::SpawnReady::new))
-                .push(metrics.stack.layer(stack_labels("balance.endpoint")))
+                .push(
+                    metrics
+                        .stack
+                        .layer(stack_labels("http", "balance.endpoint")),
+                )
                 .box_http_request(),
         )
         .check_new_service::<Endpoint, http::Request<_>>()
@@ -71,7 +75,7 @@ where
                 // If the balancer has been empty/unavailable for 10s, eagerly fail
                 // requests.
                 .push_failfast(dispatch_timeout)
-                .push(metrics.stack.layer(stack_labels("concrete"))),
+                .push(metrics.stack.layer(stack_labels("http", "concrete"))),
         )
         .into_new_service()
         .check_new_service::<Concrete, http::Request<_>>()

linkerd/app/outbound/src/ingress.rs

@@ -110,7 +110,7 @@ where
                 .push(TraceContext::layer(span_sink.clone().map(|span_sink| {
                     SpanConverter::server(span_sink, trace_labels())
                 })))
-                .push(metrics.stack.layer(stack_labels("source")))
+                .push(metrics.stack.layer(stack_labels("http", "server")))
                 .push_failfast(dispatch_timeout)
                 .push_spawn_buffer(buffer_capacity)
                 .box_http_response(),

linkerd/app/outbound/src/lib.rs

@@ -26,8 +26,8 @@ pub struct Config {
     pub allow_discovery: AddrMatch,
 }
 
-fn stack_labels(name: &'static str) -> metrics::StackLabels {
-    metrics::StackLabels::outbound(name)
+fn stack_labels(proto: &'static str, name: &'static str) -> metrics::StackLabels {
+    metrics::StackLabels::outbound(proto, name)
 }
 
 pub fn trace_labels() -> HashMap<String, String> {

linkerd/app/outbound/src/server.rs

@@ -176,7 +176,7 @@ where
                 .push(TraceContext::layer(span_sink.clone().map(|span_sink| {
                     SpanConverter::server(span_sink, trace_labels())
                 })))
-                .push(metrics.stack.layer(stack_labels("source")))
+                .push(metrics.stack.layer(stack_labels("http", "server")))
                 .push_failfast(dispatch_timeout)
                 .push_spawn_buffer(buffer_capacity)
                 .box_http_response(),
@@ -193,6 +193,7 @@ where
         .check_make_service::<tcp::Endpoint, ()>()
         .push_on_response(svc::layer::mk(tcp::Forward::new))
         .into_new_service()
+        .push_on_response(metrics.stack.layer(stack_labels("tcp", "forward")))
         .check_new_service::<tcp::Endpoint, transport::io::PrefixedIo<transport::metrics::SensorIo<I>>>()
         .push_map_target(tcp::Endpoint::from_logical(
             tls::ReasonForNoPeerName::NotProvidedByServiceDiscovery,
@@ -235,6 +236,7 @@ where
         .push_map_target(tcp::Endpoint::from_logical(
             tls::ReasonForNoPeerName::PortSkipped,
         ))
+        .push_on_response(metrics.stack.layer(stack_labels("tcp", "opaque")))
         .check_new_service::<tcp::Logical, transport::metrics::SensorIo<I>>()
         .into_inner();