feat(http/prom): introduce StreamLabel components (#4251)

see #4250 (59f90ac) for previous related work.

* feat(http/prom): add http and grpc status labelers

this commit introduces a collection of `MkStreamLabel` and `StreamLabel`
implementations that can be used to inspect status codes for HTTP and
gRPC traffic.

these do not emit duration labels, but can be used by other
implementors of these traits to perform the common logic of checking
HTTP status codes in response front matter, and for inspecting body
trailers in gRPC traffic.

Signed-off-by: katelyn martin <[email protected]>

* feat(http/prom): add an error labeler

this commit introduces a `LabelError<E>` type, which provides a
`StreamLabel` implementation that maps boxed errors to labels.

this is generic across `E`-typed labels that can be constructed `From` a
reference to a boxed `linkerd_error::Error`.

Signed-off-by: katelyn martin <[email protected]>

* feat(http/prom): add an identity labeler

this commit introduces a simple "identity" implementation of
`MkStreamLabel` and `StreamLabel`.

this is useful for metrics such as our duration recording histogram,
that do not inspect the request, response, or outcome for further
labels.

Signed-off-by: katelyn martin <[email protected]>

---------

Signed-off-by: katelyn martin <[email protected]>

Author: cratelyn

Cargo.lock

@@ -1833,6 +1833,7 @@ dependencies = [
  "prometheus-client",
  "thiserror",
  "tokio",
+ "tonic",
 ]
 
 [[package]]

linkerd/http/prom/Cargo.toml

@@ -21,6 +21,7 @@ pin-project = "1"
 prometheus-client = { workspace = true }
 thiserror = "2"
 tokio = { version = "1", features = ["time"] }
+tonic = { workspace = true }
 
 linkerd-error = { path = "../../error" }
 linkerd-http-box = { path = "../box" }

linkerd/http/prom/src/stream_label.rs

@@ -3,6 +3,10 @@
 use linkerd_error::Error;
 use prometheus_client::encoding::EncodeLabelSet;
 
+pub mod error;
+pub mod status;
+pub mod with;
+
 /// A strategy for labeling request/responses streams for status and duration
 /// metrics.
 ///

linkerd/http/prom/src/stream_label/error.rs

@@ -0,0 +1,35 @@
+//! [`StreamLabel`] implementation for labeling errors.
+
+use super::StreamLabel;
+use linkerd_error::Error;
+
+/// A [`StreamLabel`] implementation that maps boxed errors to labels.
+#[derive(Clone, Debug, Default)]
+pub struct LabelError<E> {
+    error: Option<E>,
+}
+
+// === impl LabelError ===
+
+impl<E> StreamLabel for LabelError<E>
+where
+    E: for<'a> From<&'a Error>,
+    E: Clone + Send + 'static,
+{
+    type DurationLabels = ();
+    type StatusLabels = Option<E>;
+
+    fn init_response<B>(&mut self, _: &http::Response<B>) {}
+
+    fn end_response(&mut self, res: Result<Option<&http::HeaderMap>, &Error>) {
+        let Err(err) = res else { return };
+        let labels = E::from(err);
+        self.error = Some(labels);
+    }
+
+    fn status_labels(&self) -> Self::StatusLabels {
+        self.error.clone()
+    }
+
+    fn duration_labels(&self) -> Self::DurationLabels {}
+}

linkerd/http/prom/src/stream_label/status.rs

@@ -0,0 +1,111 @@
+//! [`MkStreamLabel`] and [`StreamLabel`] implementations for status codes.
+//!
+//! This submodule provides [`MkLabelGrpcStatus`] and [`MkLabelHttpStatus`]
+
+use crate::stream_label::{MkStreamLabel, StreamLabel};
+use http::{HeaderMap, HeaderValue, Response, StatusCode};
+use linkerd_error::Error;
+use tonic::Code;
+
+/// A [`MkStreamLabel`] implementation for gRPC traffic.
+///
+/// This generates [`LabelGrpcStatus`] labelers.
+#[derive(Debug)]
+pub struct MkLabelGrpcStatus;
+
+/// A [`StreamLabel`] implementation for gRPC traffic.
+///
+/// This will inspect response headers and trailers for a [`Code`].
+#[derive(Clone, Debug, Default)]
+pub struct LabelGrpcStatus {
+    code: Option<Code>,
+}
+
+/// A [`MkStreamLabel`] implementation for HTTP traffic.
+///
+/// This generates [`LabelHttpStatus`] labelers.
+#[derive(Debug)]
+pub struct MkLabelHttpStatus;
+
+/// A [`StreamLabel`] implementation for HTTP traffic.
+///
+/// This will inspect the response headers for a [`StatusCode`].
+#[derive(Clone, Debug, Default)]
+pub struct LabelHttpStatus {
+    status: Option<StatusCode>,
+}
+
+// === impl MkLabelGrpcStatus ===
+
+impl MkStreamLabel for MkLabelGrpcStatus {
+    type DurationLabels = <Self::StreamLabel as StreamLabel>::DurationLabels;
+    type StatusLabels = <Self::StreamLabel as StreamLabel>::StatusLabels;
+    type StreamLabel = LabelGrpcStatus;
+
+    fn mk_stream_labeler<B>(&self, _: &http::Request<B>) -> Option<Self::StreamLabel> {
+        Some(LabelGrpcStatus::default())
+    }
+}
+
+// === LabelGrpcStatus ===
+
+impl StreamLabel for LabelGrpcStatus {
+    type DurationLabels = ();
+    type StatusLabels = Option<Code>;
+
+    fn init_response<B>(&mut self, rsp: &Response<B>) {
+        let headers = rsp.headers();
+        self.code = Self::get_grpc_status(headers);
+    }
+
+    fn end_response(&mut self, trailers: Result<Option<&HeaderMap>, &Error>) {
+        let Ok(Some(trailers)) = trailers else { return };
+        self.code = Self::get_grpc_status(trailers);
+    }
+
+    fn status_labels(&self) -> Self::StatusLabels {
+        self.code
+    }
+
+    fn duration_labels(&self) -> Self::DurationLabels {}
+}
+
+impl LabelGrpcStatus {
+    fn get_grpc_status(headers: &HeaderMap) -> Option<Code> {
+        headers
+            .get("grpc-status")
+            .map(HeaderValue::as_bytes)
+            .map(tonic::Code::from_bytes)
+    }
+}
+
+// === impl MkLabelHttpStatus ===
+
+impl MkStreamLabel for MkLabelHttpStatus {
+    type DurationLabels = <Self::StreamLabel as StreamLabel>::DurationLabels;
+    type StatusLabels = <Self::StreamLabel as StreamLabel>::StatusLabels;
+    type StreamLabel = LabelHttpStatus;
+
+    fn mk_stream_labeler<B>(&self, _: &http::Request<B>) -> Option<Self::StreamLabel> {
+        Some(LabelHttpStatus::default())
+    }
+}
+
+// === impl LabelHttpStatus ===
+
+impl StreamLabel for LabelHttpStatus {
+    type DurationLabels = ();
+    type StatusLabels = Option<StatusCode>;
+
+    fn init_response<B>(&mut self, rsp: &http::Response<B>) {
+        self.status = Some(rsp.status());
+    }
+
+    fn end_response(&mut self, _: Result<Option<&http::HeaderMap>, &linkerd_error::Error>) {}
+
+    fn status_labels(&self) -> Self::StatusLabels {
+        self.status
+    }
+
+    fn duration_labels(&self) -> Self::DurationLabels {}
+}

linkerd/http/prom/src/stream_label/with.rs

@@ -0,0 +1,57 @@
+//! [`StreamLabel`] implementation for labels known in advance.
+
+use super::{MkStreamLabel, StreamLabel};
+
+/// A [`MkStreamLabel`] implementation for `L`-typed labels.
+///
+/// This is useful for situations in which a tower middleware does not need to insect the request,
+/// response, or the final outcome.
+pub struct MkWithLabels<L> {
+    labels: L,
+}
+
+/// A [`StreamLabel`] implementation for `L`-typed labels.
+///
+/// This is useful for situations in which a tower middleware does not need to insect the request,
+/// response, or the final outcome.
+#[derive(Clone, Debug, Default)]
+pub struct WithLabels<L> {
+    labels: L,
+}
+
+// === impl MkWithLabels ===
+
+impl<L> MkStreamLabel for MkWithLabels<L>
+where
+    L: Clone + Send + 'static,
+{
+    type DurationLabels = L;
+    type StatusLabels = L;
+    type StreamLabel = WithLabels<L>;
+    fn mk_stream_labeler<B>(&self, _: &http::Request<B>) -> Option<Self::StreamLabel> {
+        Some(WithLabels {
+            labels: self.labels.clone(),
+        })
+    }
+}
+
+// === impl WithLabels ===
+
+impl<L> StreamLabel for WithLabels<L>
+where
+    L: Clone + Send + 'static,
+{
+    type DurationLabels = L;
+    type StatusLabels = L;
+
+    fn init_response<B>(&mut self, _: &http::Response<B>) {}
+    fn end_response(&mut self, _: Result<Option<&http::HeaderMap>, &linkerd_error::Error>) {}
+
+    fn status_labels(&self) -> Self::StatusLabels {
+        self.labels.clone()
+    }
+
+    fn duration_labels(&self) -> Self::DurationLabels {
+        self.labels.clone()
+    }
+}