retry: allow retrying requests without content-length headers (#1341)

Currently, the proxy will retry requests with bodies if and only if they have a `content-length` header with a value <= 64 KB. If the request has a body but no `content-length` header, we currently assume that its body will exceed the maximum buffered size, and skip trying to retry it. However, this means gRPC requests will never be retried, because it turns out gRPC requests don't include a `content-length` header (see linkerd/linkerd2#7141). Whoops! This PR fixes this by changing the retry logic to use `Body::size_hint` to determine if buffering should be attempted. This value will be set from `content-length` when it is set and may be set in additional situations where the body length is known before the request is processed. We are still protected against unbounded buffering because the buffering body will stop buffering and discard any previously buffered data if the buffered length ever exceeds the maximum. Co-authored-by: Oliver Gould <ver@buoyant.io> (cherry picked from commit bfc1e2c) Signed-off-by: Oliver Gould <ver@buoyant.io>
linkerd · Mar 30, 2022 · fb470b3 · fb470b3
1 parent f7ef27f
commit fb470b3
Show file tree

Hide file tree

Showing 11 changed files with 234 additions and 106 deletions.
diff --git a/hyper-balance/src/lib.rs b/hyper-balance/src/lib.rs
@@ -127,6 +127,11 @@ where
 
         this.body.poll_trailers(cx)
     }
+
+    #[inline]
+    fn size_hint(&self) -> hyper::body::SizeHint {
+        self.body.size_hint()
+    }
 }
 
 // ==== PendingUntilEosBody ====
@@ -147,6 +152,7 @@ impl<T: Send + 'static, B: HttpBody> HttpBody for PendingUntilEosBody<T, B> {
     type Data = B::Data;
     type Error = B::Error;
 
+    #[inline]
     fn is_end_stream(&self) -> bool {
         self.body.is_end_stream()
     }
@@ -181,6 +187,11 @@ impl<T: Send + 'static, B: HttpBody> HttpBody for PendingUntilEosBody<T, B> {
 
         Poll::Ready(ret)
     }
+
+    #[inline]
+    fn size_hint(&self) -> hyper::body::SizeHint {
+        self.body.size_hint()
+    }
 }
 
 #[cfg(test)]

diff --git a/linkerd/app/core/src/retry.rs b/linkerd/app/core/src/retry.rs
@@ -58,55 +58,33 @@ impl retry::NewPolicy<Route> for NewRetryPolicy {
 
 // === impl Retry ===
 
-impl RetryPolicy {
-    fn can_retry<A: http_body::Body>(&self, req: &http::Request<A>) -> bool {
-        let content_length = |req: &http::Request<_>| {
-            req.headers()
-                .get(http::header::CONTENT_LENGTH)
-                .and_then(|value| value.to_str().ok()?.parse::<usize>().ok())
-        };
-
-        // Requests without bodies can always be retried, as we will not need to
-        // buffer the body. If the request *does* have a body, retry it if and
-        // only if the request contains a `content-length` header and the
-        // content length is >= 64 kb.
-        let has_body = !req.body().is_end_stream();
-        if has_body && content_length(req).unwrap_or(usize::MAX) > MAX_BUFFERED_BYTES {
-            tracing::trace!(
-                req.has_body = has_body,
-                req.content_length = ?content_length(req),
-                "not retryable",
-            );
-            return false;
-        }
-
-        tracing::trace!(
-            req.has_body = has_body,
-            req.content_length = ?content_length(req),
-            "retryable",
-        );
-        true
-    }
-}
-
-impl<A, B, E> retry::Policy<http::Request<A>, http::Response<B>, E> for RetryPolicy
+impl<A, B, E> retry::Policy<http::Request<ReplayBody<A>>, http::Response<B>, E> for RetryPolicy
 where
-    A: http_body::Body + Clone,
+    A: http_body::Body + Unpin,
+    A::Error: Into<Error>,
 {
     type Future = future::Ready<Self>;
 
     fn retry(
         &self,
-        req: &http::Request<A>,
+        req: &http::Request<ReplayBody<A>>,
         result: Result<&http::Response<B>, &E>,
     ) -> Option<Self::Future> {
         let retryable = match result {
             Err(_) => false,
-            Ok(rsp) => classify::Request::from(self.response_classes.clone())
-                .classify(req)
-                .start(rsp)
-                .eos(None)
-                .is_failure(),
+            Ok(rsp) => {
+                // is the request a failure?
+                let is_failure = classify::Request::from(self.response_classes.clone())
+                    .classify(req)
+                    .start(rsp)
+                    .eos(None)
+                    .is_failure();
+                // did the body exceed the maximum length limit?
+                let exceeded_max_len = req.body().is_capped();
+                let retryable = is_failure && !exceeded_max_len;
+                tracing::trace!(is_failure, exceeded_max_len, retryable);
+                retryable
+            }
         };
 
         if !retryable {
@@ -123,16 +101,12 @@ where
         Some(future::ready(self.clone()))
     }
 
-    fn clone_request(&self, req: &http::Request<A>) -> Option<http::Request<A>> {
-        let can_retry = self.can_retry(req);
-        debug_assert!(
-            can_retry,
-            "The retry policy attempted to clone an un-retryable request. This is unexpected."
-        );
-        if !can_retry {
-            return None;
-        }
-
+    fn clone_request(
+        &self,
+        req: &http::Request<ReplayBody<A>>,
+    ) -> Option<http::Request<ReplayBody<A>>> {
+        // Since the body is already wrapped in a ReplayBody, it must not be obviously too large to
+        // buffer/clone.
         let mut clone = http::Request::new(req.body().clone());
         *clone.method_mut() = req.method().clone();
         *clone.uri_mut() = req.uri().clone();
@@ -160,9 +134,20 @@ where
         &self,
         req: http::Request<A>,
     ) -> Either<Self::RetryRequest, http::Request<A>> {
-        if self.can_retry(&req) {
-            return Either::A(req.map(|body| ReplayBody::new(body, MAX_BUFFERED_BYTES)));
-        }
-        Either::B(req)
+        let (head, body) = req.into_parts();
+        let replay_body = match ReplayBody::try_new(body, MAX_BUFFERED_BYTES) {
+            Ok(body) => body,
+            Err(body) => {
+                tracing::debug!(
+                    size = body.size_hint().lower(),
+                    "Body is too large to buffer"
+                );
+                return Either::B(http::Request::from_parts(head, body));
+            }
+        };
+
+        // The body may still be too large to be buffered if the body's length was not known.
+        // `ReplayBody` handles this gracefully.
+        Either::A(http::Request::from_parts(head, replay_body))
     }
 }
diff --git a/linkerd/app/integration/src/tests/profiles.rs b/linkerd/app/integration/src/tests/profiles.rs
@@ -48,7 +48,8 @@ macro_rules! profile_test {
                 async move {
                     // Read the entire body before responding, so that the
                     // client doesn't fail when writing it out.
-                    let _body = hyper::body::aggregate(req.into_body()).await;
+                    let _body = hyper::body::to_bytes(req.into_body()).await;
+                    tracing::debug!(body = ?_body.as_ref().map(|body| body.len()), "recieved body");
                     Ok::<_, Error>(if fail {
                         Response::builder()
                             .status(533)
@@ -222,6 +223,36 @@ async fn retry_with_small_put_body() {
     }
 }
 
+#[tokio::test]
+async fn retry_without_content_length() {
+    profile_test! {
+        routes: [
+            controller::route()
+                .request_any()
+                .response_failure(500..600)
+                .retryable(true)
+        ],
+        budget: Some(controller::retry_budget(Duration::from_secs(10), 0.1, 1)),
+        with_client: |client: client::Client| async move {
+            let (mut tx, body) = hyper::body::Body::channel();
+            let req = client.request_builder("/0.5")
+                .method("POST")
+                .body(body)
+                .unwrap();
+            let res = tokio::spawn(async move { client.request_body(req).await });
+            tx.send_data(Bytes::from_static(b"hello"))
+                .await
+                .expect("the whole body should be read");
+            tx.send_data(Bytes::from_static(b"world"))
+                .await
+                .expect("the whole body should be read");
+            drop(tx);
+            let res = res.await.unwrap();
+            assert_eq!(res.status(), 200);
+        }
+    }
+}
+
 #[tokio::test]
 async fn does_not_retry_if_request_does_not_match() {
     profile_test! {
@@ -280,7 +311,9 @@ async fn does_not_retry_if_body_is_too_long() {
 }
 
 #[tokio::test]
-async fn does_not_retry_if_body_lacks_known_length() {
+async fn does_not_retry_if_streaming_body_exceeds_max_length() {
+    // TODO(eliza): if we make the max length limit configurable, update this
+    // test to test the configurable max length limit...
     profile_test! {
         routes: [
             controller::route()
@@ -296,10 +329,21 @@ async fn does_not_retry_if_body_lacks_known_length() {
                 .body(body)
                 .unwrap();
             let res = tokio::spawn(async move { client.request_body(req).await });
-            let _ = tx.send_data(Bytes::from_static(b"hello"));
-            let _ = tx.send_data(Bytes::from_static(b"world"));
+            // send a 32k chunk
+            tx.send_data(Bytes::from(&[1u8; 32 * 1024][..]))
+                .await
+                .expect("the whole body should be read");
+            // ...and another one...
+            tx.send_data(Bytes::from(&[1u8; 32 * 1024][..]))
+                .await
+                .expect("the whole body should be read");
+            // ...and a third one (exceeding the max length limit)
+            tx.send_data(Bytes::from(&[1u8; 32 * 1024][..]))
+                .await
+                .expect("the whole body should be read");
             drop(tx);
             let res = res.await.unwrap();
+
             assert_eq!(res.status(), 533);
         }
     }

diff --git a/linkerd/app/integration/src/tests/transparency.rs b/linkerd/app/integration/src/tests/transparency.rs
@@ -843,9 +843,7 @@ macro_rules! http1_tests {
 
             let srv = server::http1()
                 .route_fn("/", |req| {
-                    let has_body_header = req.headers().contains_key("transfer-encoding")
-                        || req.headers().contains_key("content-length");
-                    let status = if has_body_header {
+                    let status = if req.headers().contains_key(http::header::TRANSFER_ENCODING) {
                         StatusCode::BAD_REQUEST
                     } else {
                         StatusCode::OK

diff --git a/linkerd/http-box/src/body.rs b/linkerd/http-box/src/body.rs
@@ -45,24 +45,28 @@ impl Body for BoxBody {
     type Data = Data;
     type Error = Error;
 
+    #[inline]
     fn is_end_stream(&self) -> bool {
         self.inner.is_end_stream()
     }
 
+    #[inline]
     fn poll_data(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
         self.as_mut().inner.as_mut().poll_data(cx)
     }
 
+    #[inline]
     fn poll_trailers(
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<Option<HeaderMap<HeaderValue>>, Self::Error>> {
         self.as_mut().inner.as_mut().poll_trailers(cx)
     }
 
+    #[inline]
     fn size_hint(&self) -> http_body::SizeHint {
         self.inner.size_hint()
     }
@@ -95,6 +99,7 @@ where
     type Data = Data;
     type Error = Error;
 
+    #[inline]
     fn is_end_stream(&self) -> bool {
         self.0.is_end_stream()
     }
@@ -111,13 +116,15 @@ where
         }))
     }
 
+    #[inline]
     fn poll_trailers(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<Option<HeaderMap<HeaderValue>>, Self::Error>> {
         Poll::Ready(futures::ready!(self.project().0.poll_trailers(cx)).map_err(Into::into))
     }
 
+    #[inline]
     fn size_hint(&self) -> http_body::SizeHint {
         self.0.size_hint()
     }

diff --git a/linkerd/http-metrics/src/requests/service.rs b/linkerd/http-metrics/src/requests/service.rs
@@ -283,6 +283,7 @@ where
         self.project().inner.poll_trailers(cx)
     }
 
+    #[inline]
     fn size_hint(&self) -> http_body::SizeHint {
         self.inner.size_hint()
     }
@@ -438,6 +439,11 @@ where
 
         Poll::Ready(Ok(trls))
     }
+
+    #[inline]
+    fn size_hint(&self) -> http_body::SizeHint {
+        self.inner.size_hint()
+    }
 }
 
 #[pinned_drop]