refactor(http/prom): StreamLabel::end_response accepts EosRef

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

Author: cratelyn

linkerd/app/outbound/src/http/logical/policy/route/metrics.rs

@@ -10,7 +10,7 @@ use linkerd_http_prom::{
     stream_label::{
         error::LabelError,
         status::{LabelGrpcStatus, LabelHttpStatus},
-        LabelSet, StreamLabel,
+        EosRef, LabelSet, StreamLabel,
     },
 };
 
@@ -337,7 +337,7 @@ where
         error.init_response(rsp);
     }
 
-    fn end_response(&mut self, res: Result<Option<&http::HeaderMap>, &linkerd_app_core::Error>) {
+    fn end_response(&mut self, res: EosRef<'_>) {
         let Self {
             parent: _,
             status,
@@ -395,7 +395,7 @@ where
         error.init_response(rsp);
     }
 
-    fn end_response(&mut self, res: Result<Option<&http::HeaderMap>, &linkerd_app_core::Error>) {
+    fn end_response(&mut self, res: EosRef<'_>) {
         let Self {
             parent: _,
             status,

linkerd/http/prom/src/record_response.rs

@@ -1,6 +1,7 @@
 use crate::stream_label::{LabelSet, MkStreamLabel, StreamLabel};
 use http_body::Body;
 use linkerd_error::Error;
+use linkerd_http_body_eos::EosRef;
 use linkerd_http_box::BoxBody;
 use linkerd_stack as svc;
 use prometheus_client::metrics::{
@@ -180,15 +181,15 @@ where
 
                 let (head, inner) = rsp.into_parts();
                 if inner.is_end_stream() {
-                    end_stream(&mut state, Ok(None));
+                    end_stream(&mut state, EosRef::None);
                 }
                 Poll::Ready(Ok(http::Response::from_parts(
                     head,
                     BoxBody::new(ResponseBody { inner, state }),
                 )))
             }
             Err(error) => {
-                end_stream(&mut state, Err(&error));
+                end_stream(&mut state, EosRef::Error(&error));
                 Poll::Ready(Err(error))
             }
         }
@@ -217,14 +218,14 @@ where
 
         match &frame {
             Some(Ok(frame)) => {
-                if let trls @ Some(_) = frame.trailers_ref() {
-                    end_stream(this.state, Ok(trls));
+                if let Some(trls) = frame.trailers_ref() {
+                    end_stream(this.state, EosRef::Trailers(trls));
                 } else if this.inner.is_end_stream() {
-                    end_stream(this.state, Ok(None));
+                    end_stream(this.state, EosRef::None);
                 }
             }
-            Some(Err(error)) => end_stream(this.state, Err(error)),
-            None => end_stream(this.state, Ok(None)),
+            Some(Err(error)) => end_stream(this.state, EosRef::Error(error)),
+            None => end_stream(this.state, EosRef::None),
         }
 
         Poll::Ready(frame)
@@ -246,15 +247,13 @@ where
     fn drop(self: Pin<&mut Self>) {
         let this = self.project();
         if this.state.is_some() {
-            end_stream(this.state, Err(&RequestCancelled.into()));
+            end_stream(this.state, EosRef::Cancelled);
         }
     }
 }
 
-fn end_stream<L>(
-    state: &mut Option<ResponseState<L>>,
-    res: Result<Option<&http::HeaderMap>, &Error>,
-) where
+fn end_stream<L>(state: &mut Option<ResponseState<L>>, res: EosRef<'_>)
+where
     L: StreamLabel,
     L::DurationLabels: LabelSet,
 {

linkerd/http/prom/src/status.rs

@@ -1,9 +1,6 @@
 //! A tower middleware for counting response status codes.
 
-use crate::{
-    record_response::RequestCancelled,
-    stream_label::{LabelSet, MkStreamLabel, StreamLabel},
-};
+use crate::stream_label::{LabelSet, MkStreamLabel, StreamLabel};
 use http::{Request, Response};
 use http_body::Body;
 use linkerd_error::Error;
@@ -258,16 +255,7 @@ where
     L: Clone + Hash + Eq + Send + Sync + 'static,
 {
     fn on_eos(eos: EosRef<'_, Error>, mut stream_label: SL, metrics: StatusMetrics<L>) {
-        // TODO(kate): a static cancellation error. see linkerd/linkerd2-proxy#4306.
-        let cancelled = RequestCancelled.into();
-
-        stream_label.end_response(match eos {
-            EosRef::None => Ok(None),
-            EosRef::Trailers(trls) => Ok(Some(trls)),
-            EosRef::Error(error) => Err(error),
-            EosRef::Cancelled => Err(&cancelled),
-        });
-
+        stream_label.end_response(eos);
         let labels = stream_label.status_labels();
         let counter = metrics.metric(&labels);
         counter.inc();

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

@@ -484,7 +484,7 @@ mod util {
             http.init_response(rsp);
         }
 
-        fn end_response(&mut self, trailers: Result<Option<&http::HeaderMap>, &Error>) {
+        fn end_response(&mut self, trailers: EosRef<'_>) {
             let Self { grpc, http } = self;
             grpc.end_response(trailers);
             http.end_response(trailers);

linkerd/http/prom/src/stream_label.rs

@@ -1,12 +1,13 @@
 //! Stream labeling facilities.
 
-use linkerd_error::Error;
 use prometheus_client::encoding::EncodeLabelSet;
 
 pub mod error;
 pub mod status;
 pub mod with;
 
+pub use linkerd_http_body_eos::EosRef;
+
 /// A strategy for labeling request/responses streams for status and duration
 /// metrics.
 ///
@@ -30,7 +31,7 @@ pub trait StreamLabel: Send + 'static {
     type StatusLabels;
 
     fn init_response<B>(&mut self, rsp: &http::Response<B>);
-    fn end_response(&mut self, trailers: Result<Option<&http::HeaderMap>, &Error>);
+    fn end_response(&mut self, trailers: EosRef<'_>);
 
     fn status_labels(&self) -> Self::StatusLabels;
     fn duration_labels(&self) -> Self::DurationLabels;

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

@@ -2,6 +2,7 @@
 
 use super::StreamLabel;
 use linkerd_error::Error;
+use linkerd_http_body_eos::EosRef;
 
 /// A [`StreamLabel`] implementation that maps boxed errors to labels.
 #[derive(Clone, Debug, Default)]
@@ -21,8 +22,8 @@ where
 
     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 };
+    fn end_response(&mut self, res: EosRef<'_>) {
+        let EosRef::Error(err) = res else { return };
         let labels = E::from(err);
         self.error = Some(labels);
     }

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

@@ -4,7 +4,7 @@
 
 use crate::stream_label::{MkStreamLabel, StreamLabel};
 use http::{HeaderMap, HeaderValue, Response, StatusCode};
-use linkerd_error::Error;
+use linkerd_http_body_eos::EosRef;
 use tonic::Code;
 
 /// A [`MkStreamLabel`] implementation for gRPC traffic.
@@ -58,8 +58,10 @@ impl StreamLabel for LabelGrpcStatus {
         self.code = Self::get_grpc_status(headers);
     }
 
-    fn end_response(&mut self, trailers: Result<Option<&HeaderMap>, &Error>) {
-        let Ok(Some(trailers)) = trailers else { return };
+    fn end_response(&mut self, trailers: EosRef<'_>) {
+        let EosRef::Trailers(trailers) = trailers else {
+            return;
+        };
         self.code = Self::get_grpc_status(trailers);
     }
 
@@ -101,7 +103,7 @@ impl StreamLabel for LabelHttpStatus {
         self.status = Some(rsp.status());
     }
 
-    fn end_response(&mut self, _: Result<Option<&http::HeaderMap>, &linkerd_error::Error>) {}
+    fn end_response(&mut self, _: EosRef<'_>) {}
 
     fn status_labels(&self) -> Self::StatusLabels {
         self.status

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

@@ -1,6 +1,7 @@
 //! [`StreamLabel`] implementation for labels known in advance.
 
 use super::{MkStreamLabel, StreamLabel};
+use linkerd_http_body_eos::EosRef;
 
 /// A [`MkStreamLabel`] implementation for `L`-typed labels.
 ///
@@ -45,7 +46,7 @@ where
     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 end_response(&mut self, _: EosRef<'_>) {}
 
     fn status_labels(&self) -> Self::StatusLabels {
         self.labels.clone()