fix(metrics): compute time before receiving the first response from the stream each time

NEXT_CHANGELOG.md

@@ -91,6 +91,12 @@ By [@o0Ignition0o](https://github.com/@o0Ignition0o) in https://github.com/apoll
 
 ## 🐛 Fixes
 
+### Fix metrics duration for router request ([#1705](https://github.com/apollographql/router/issues/1705))
+
+With the introduction of `BoxStream` for defer we introduced a bug when computing http request duration metrics. Sometimes we didn't wait for the first response in the `BoxStream`.
+
+By [@bnjjj](https://github.com/bnjjj) in https://github.com/apollographql/router/pull/1705
+
 ### Numerous fixes to preview `@defer` query planning ([Issue #1698](https://github.com/apollographql/router/issues/1698))
 
 Updated to [Federation `2.1.2-alpha.0`](https://github.com/apollographql/federation/pull/2132) which brings in a number of fixes for the preview `@defer` support.  These fixes include:

apollo-router/src/plugins/telemetry/metrics/mod.rs

@@ -4,10 +4,8 @@ use std::sync::Arc;
 
 use ::serde::Deserialize;
 use access_json::JSONQuery;
-use futures::future::ready;
-use futures::stream::once;
-use futures::StreamExt;
 use http::header::HeaderName;
+use http::response::Parts;
 use http::HeaderMap;
 use multimap::MultiMap;
 use opentelemetry::metrics::Counter;
@@ -23,14 +21,14 @@ use serde_json::Value;
 use tower::BoxError;
 
 use crate::error::FetchError;
+use crate::graphql;
 use crate::graphql::Request;
 use crate::plugin::serde::deserialize_header_name;
 use crate::plugin::serde::deserialize_json_query;
 use crate::plugin::serde::deserialize_regex;
 use crate::plugins::telemetry::config::MetricsCommon;
 use crate::plugins::telemetry::metrics::apollo::Sender;
 use crate::router_factory::Endpoint;
-use crate::services::SupergraphResponse;
 use crate::Context;
 use crate::ListenAddr;
 
@@ -252,10 +250,12 @@ impl ErrorsForward {
 }
 
 impl AttributesForwardConf {
-    pub(crate) async fn get_attributes_from_router_response(
+    pub(crate) fn get_attributes_from_router_response(
         &self,
-        response: SupergraphResponse,
-    ) -> (SupergraphResponse, HashMap<String, String>) {
+        parts: &Parts,
+        context: &Context,
+        first_response: &Option<graphql::Response>,
+    ) -> HashMap<String, String> {
         let mut attributes = HashMap::new();
 
         // Fill from static
@@ -264,7 +264,6 @@ impl AttributesForwardConf {
                 attributes.insert(name.clone(), value.clone());
             }
         }
-        let context = response.context;
         // Fill from context
         if let Some(from_context) = &self.context {
             for ContextForward {
@@ -288,8 +287,7 @@ impl AttributesForwardConf {
                 };
             }
         }
-        let (parts, stream) = response.response.into_parts();
-        let (first, rest) = stream.into_future().await;
+
         // Fill from response
         if let Some(from_response) = &self.response {
             if let Some(header_forward) = &from_response.header {
@@ -303,7 +301,7 @@ impl AttributesForwardConf {
             }
 
             if let Some(body_forward) = &from_response.body {
-                if let Some(body) = &first {
+                if let Some(body) = &first_response {
                     for body_fw in body_forward {
                         let output = body_fw.path.execute(body).unwrap();
                         if let Some(val) = output {
@@ -320,12 +318,7 @@ impl AttributesForwardConf {
             }
         }
 
-        let response = http::Response::from_parts(
-            parts,
-            once(ready(first.unwrap_or_default())).chain(rest).boxed(),
-        );
-
-        (SupergraphResponse { context, response }, attributes)
+        attributes
     }
 
     /// Get attributes from context

apollo-router/src/plugins/telemetry/mod.rs

@@ -15,7 +15,9 @@ use ::tracing::Span;
 use ::tracing::Subscriber;
 use apollo_spaceport::server::ReportSpaceport;
 use apollo_spaceport::StatsContext;
+use futures::future::ready;
 use futures::future::BoxFuture;
+use futures::stream::once;
 use futures::FutureExt;
 use futures::StreamExt;
 use http::HeaderValue;
@@ -832,6 +834,11 @@ impl Telemetry {
                     "status",
                     response.response.status().as_u16().to_string(),
                 ));
+
+                // Wait for the first response of the stream
+                let (parts, stream) = response.response.into_parts();
+                let (first_response, rest) = stream.into_future().await;
+
                 if let Some(MetricsCommon {
                     attributes:
                         Some(MetricsAttributesConf {
@@ -841,22 +848,29 @@ impl Telemetry {
                     ..
                 }) = &config.metrics.as_ref().and_then(|m| m.common.as_ref())
                 {
-                    let (resp, attributes) = forward_attributes
-                        .get_attributes_from_router_response(response)
-                        .await;
+                    let attributes = forward_attributes.get_attributes_from_router_response(
+                        &parts,
+                        &context,
+                        &first_response,
+                    );
 
                     metric_attrs.extend(attributes.into_iter().map(|(k, v)| KeyValue::new(k, v)));
                     metrics.http_requests_total.add(1, &metric_attrs);
-
-                    Ok(resp)
                 } else {
                     metrics.http_requests_total.add(1, &metric_attrs);
-
-                    Ok(response)
                 }
+
+                let response = http::Response::from_parts(
+                    parts,
+                    once(ready(first_response.unwrap_or_default()))
+                        .chain(rest)
+                        .boxed(),
+                );
+
+                Ok(SupergraphResponse { context, response })
             }
             Err(err) => {
-                metrics.http_requests_error_total.add(1, &[]);
+                metrics.http_requests_error_total.add(1, &metric_attrs);
 
                 Err(err)
             }