retry: only buffer request bodies when retries are enabled (#1020)

PR #1017 added support for retries on requests with bodies. This
requires buffering the body data so that it can be sent as part of a
retry request.

In #1017, all requests in the profile HTTP route stack have their bodies
wrapped in a `ReplayBody` type that buffers the request data. Although
the buffering strategy is lazy (the request can be sent before the
client finishes sending the body, so we don't introduce the latency of
waiting to buffer the entire body) and fairly lightweight (the buffered
body data is not copied an additional time, and is cloned only by
increasing the reference count on the buffer it was read into from the
wire), there is still some additional overhead that could be avoided
when retries are not configured on a route.

This branch updates the `Retry` service so that buffering is only added
on routes which have a retry policy. If there is no retry policy, we
avoid wrapping the request's body in a `ReplayBody`.

The `retry` crate is now called `http-retry`.

Author: hawkw

Cargo.lock

@@ -660,6 +660,7 @@ dependencies = [
  "linkerd-exp-backoff",
  "linkerd-http-classify",
  "linkerd-http-metrics",
+ "linkerd-http-retry",
  "linkerd-identity",
  "linkerd-io",
  "linkerd-metrics",
@@ -675,7 +676,6 @@ dependencies = [
  "linkerd-proxy-tcp",
  "linkerd-proxy-transport",
  "linkerd-reconnect",
- "linkerd-retry",
  "linkerd-service-profiles",
  "linkerd-stack",
  "linkerd-stack-metrics",
@@ -781,9 +781,9 @@ dependencies = [
  "ipnet",
  "linkerd-app-core",
  "linkerd-app-test",
+ "linkerd-http-retry",
  "linkerd-identity",
  "linkerd-io",
- "linkerd-retry",
  "linkerd-tracing",
  "pin-project",
  "thiserror",
@@ -995,6 +995,26 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "linkerd-http-retry"
+version = "0.1.0"
+dependencies = [
+ "bytes",
+ "futures",
+ "http",
+ "http-body",
+ "hyper",
+ "linkerd-error",
+ "linkerd-http-box",
+ "linkerd-stack",
+ "linkerd-tracing",
+ "parking_lot",
+ "pin-project",
+ "tokio",
+ "tower",
+ "tracing",
+]
+
 [[package]]
 name = "linkerd-identity"
 version = "0.1.0"
@@ -1258,25 +1278,6 @@ dependencies = [
  "tracing",
 ]
 
-[[package]]
-name = "linkerd-retry"
-version = "0.1.0"
-dependencies = [
- "bytes",
- "futures",
- "http",
- "http-body",
- "hyper",
- "linkerd-error",
- "linkerd-stack",
- "linkerd-tracing",
- "parking_lot",
- "pin-project",
- "tokio",
- "tower",
- "tracing",
-]
-
 [[package]]
 name = "linkerd-service-profiles"
 version = "0.1.0"

Cargo.toml

@@ -30,6 +30,7 @@ members = [
     "linkerd/http-box",
     "linkerd/http-classify",
     "linkerd/http-metrics",
+    "linkerd/http-retry",
     "linkerd/identity",
     "linkerd/io",
     "linkerd/metrics",
@@ -45,7 +46,6 @@ members = [
     "linkerd/proxy/tcp",
     "linkerd/proxy/transport",
     "linkerd/reconnect",
-    "linkerd/retry",
     "linkerd/service-profiles",
     "linkerd/signal",
     "linkerd/stack",

linkerd/app/core/Cargo.toml

@@ -35,6 +35,8 @@ linkerd-error-respond = { path = "../../error-respond" }
 linkerd-exp-backoff = { path = "../../exp-backoff" }
 linkerd-http-classify = { path = "../../http-classify" }
 linkerd-http-metrics = { path = "../../http-metrics" }
+
+linkerd-http-retry = { path = "../../http-retry" }
 linkerd-identity = { path = "../../identity" }
 linkerd-io = { path = "../../io" }
 linkerd-metrics = { path = "../../metrics" }
@@ -52,7 +54,6 @@ linkerd-proxy-tap = { path = "../../proxy/tap" }
 linkerd-proxy-tcp = { path = "../../proxy/tcp" }
 linkerd-proxy-transport = { path = "../../proxy/transport" }
 linkerd-reconnect = { path = "../../reconnect" }
-linkerd-retry = { path = "../../retry" }
 linkerd-timeout = { path = "../../timeout" }
 linkerd-tracing = { path = "../../tracing" }
 linkerd-service-profiles = { path = "../../service-profiles" }

linkerd/app/core/src/retry.rs

@@ -6,13 +6,12 @@ use crate::profiles;
 use futures::future;
 use http_body::Body;
 use linkerd_http_classify::{Classify, ClassifyEos, ClassifyResponse};
-use linkerd_retry::NewRetryLayer;
 use linkerd_stack::Param;
 use std::marker::PhantomData;
 use std::sync::Arc;
 use tower::retry::budget::Budget;
 
-pub use linkerd_retry::*;
+pub use linkerd_http_retry::*;
 
 pub fn layer(metrics: HttpRouteRetry) -> NewRetryLayer<NewRetry> {
     NewRetryLayer::new(NewRetry::new(metrics))
@@ -54,7 +53,7 @@ impl NewRetry {
     }
 }
 
-impl<C> linkerd_retry::NewPolicy<Route> for NewRetry<C> {
+impl<C> NewPolicy<Route> for NewRetry<C> {
     type Policy = Retry<C>;
 
     fn new_policy(&self, route: &Route) -> Option<Self::Policy> {
@@ -70,7 +69,7 @@ impl<C> linkerd_retry::NewPolicy<Route> for NewRetry<C> {
     }
 }
 
-impl<C, A, B, E> linkerd_retry::Policy<http::Request<A>, http::Response<B>, E> for Retry<C>
+impl<C, A, B, E> Policy<http::Request<A>, http::Response<B>, E> for Retry<C>
 where
     C: CloneRequest<http::Request<A>>,
 {

linkerd/app/inbound/fuzz/Cargo.lock

@@ -590,6 +590,7 @@ dependencies = [
  "linkerd-exp-backoff",
  "linkerd-http-classify",
  "linkerd-http-metrics",
+ "linkerd-http-retry",
  "linkerd-identity",
  "linkerd-io",
  "linkerd-metrics",
@@ -605,7 +606,6 @@ dependencies = [
  "linkerd-proxy-tcp",
  "linkerd-proxy-transport",
  "linkerd-reconnect",
- "linkerd-retry",
  "linkerd-service-profiles",
  "linkerd-stack",
  "linkerd-stack-metrics",
@@ -860,6 +860,23 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "linkerd-http-retry"
+version = "0.1.0"
+dependencies = [
+ "bytes",
+ "futures",
+ "http",
+ "http-body",
+ "linkerd-error",
+ "linkerd-http-box",
+ "linkerd-stack",
+ "parking_lot",
+ "pin-project",
+ "tower",
+ "tracing",
+]
+
 [[package]]
 name = "linkerd-identity"
 version = "0.1.0"
@@ -1116,22 +1133,6 @@ dependencies = [
  "tracing",
 ]
 
-[[package]]
-name = "linkerd-retry"
-version = "0.1.0"
-dependencies = [
- "bytes",
- "futures",
- "http",
- "http-body",
- "linkerd-error",
- "linkerd-stack",
- "parking_lot",
- "pin-project",
- "tower",
- "tracing",
-]
-
 [[package]]
 name = "linkerd-service-profiles"
 version = "0.1.0"

linkerd/app/outbound/Cargo.toml

@@ -20,8 +20,8 @@ http = "0.2"
 futures = "0.3.9"
 indexmap = "1.0"
 linkerd-app-core = { path = "../core" }
+linkerd-http-retry = { path = "../../http-retry" }
 linkerd-identity = { path = "../../identity" }
-linkerd-retry = { path = "../../retry" }
 thiserror = "1.0"
 tokio = { version = "1", features = ["sync"] }
 tower = { version = "0.4.7", features = ["util"] }

linkerd/retry/Cargo.toml renamed to linkerd/http-retry/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "linkerd-retry"
+name = "linkerd-http-retry"
 version = "0.1.0"
 authors = ["Linkerd Developers <[email protected]>"]
 license = "Apache-2.0"
@@ -12,6 +12,7 @@ futures = { version = "0.3", default-features = false }
 http-body = "0.4"
 http = "0.2"
 linkerd-error = { path = "../error" }
+linkerd-http-box = { path = "../http-box" }
 linkerd-stack = { path = "../stack" }
 pin-project = "1"
 parking_lot = "0.11"

linkerd/retry/src/lib.rs renamed to linkerd/http-retry/src/lib.rs

@@ -1,8 +1,10 @@
 #![deny(warnings, rust_2018_idioms)]
 #![forbid(unsafe_code)]
 #![allow(clippy::inconsistent_struct_constructor)]
+#![allow(clippy::type_complexity)]
 
 use linkerd_error::Error;
+use linkerd_http_box::BoxBody;
 use linkerd_stack::{NewService, Proxy, ProxyService};
 use pin_project::pin_project;
 use std::future::Future;
@@ -41,17 +43,28 @@ pub struct Retry<P, S> {
     policy: Option<P>,
     inner: S,
 }
-
 #[pin_project(project = ResponseFutureProj)]
-pub enum ResponseFuture<R, P, S, Req>
+pub enum ResponseFuture<R, P, S, B>
 where
-    R: tower::retry::Policy<Req, P::Response, Error> + Clone,
-    P: Proxy<Req, S> + Clone,
+    R: tower::retry::Policy<http::Request<ReplayBody<B>>, P::Response, Error> + Clone,
+    P: Proxy<http::Request<BoxBody>, S> + Clone,
     S: tower::Service<P::Request> + Clone,
     S::Error: Into<Error>,
 {
     Disabled(#[pin] P::Future),
-    Retry(#[pin] Oneshot<tower::retry::Retry<R, ProxyService<P, S>>, Req>),
+    Retry(
+        #[pin]
+        Oneshot<
+            tower::retry::Retry<
+                R,
+                tower::util::MapRequest<
+                    ProxyService<P, S>,
+                    fn(http::Request<ReplayBody<B>>) -> http::Request<BoxBody>,
+                >,
+            >,
+            http::Request<ReplayBody<B>>,
+        >,
+    ),
 }
 
 // === impl NewRetryLayer ===
@@ -93,35 +106,40 @@ where
 
 // === impl Retry ===
 
-impl<R, P, Req, S> Proxy<Req, S> for Retry<R, P>
+impl<R, P, B, S> Proxy<http::Request<B>, S> for Retry<R, P>
 where
-    R: tower::retry::Policy<Req, P::Response, Error> + Clone,
-    P: Proxy<Req, S> + Clone,
+    R: tower::retry::Policy<http::Request<ReplayBody<B>>, P::Response, Error> + Clone,
+    P: Proxy<http::Request<BoxBody>, S> + Clone,
     S: tower::Service<P::Request> + Clone,
     S::Error: Into<Error>,
+    B: http_body::Body + Unpin + Send + 'static,
+    B::Data: Send,
+    B::Error: Into<Error>,
 {
     type Request = P::Request;
     type Response = P::Response;
     type Error = Error;
-    type Future = ResponseFuture<R, P, S, Req>;
+    type Future = ResponseFuture<R, P, S, B>;
 
-    fn proxy(&self, svc: &mut S, req: Req) -> Self::Future {
+    fn proxy(&self, svc: &mut S, req: http::Request<B>) -> Self::Future {
         trace!(retryable = %self.policy.is_some());
 
         if let Some(policy) = self.policy.as_ref() {
-            let inner = self.inner.clone().wrap_service(svc.clone());
+            let inner = self.inner.clone().wrap_service(svc.clone()).map_request(
+                (|req: http::Request<ReplayBody<B>>| req.map(BoxBody::new)) as fn(_) -> _,
+            );
             let retry = tower::retry::Retry::new(policy.clone(), inner);
-            return ResponseFuture::Retry(retry.oneshot(req));
+            return ResponseFuture::Retry(retry.oneshot(req.map(ReplayBody::new)));
         }
 
-        ResponseFuture::Disabled(self.inner.proxy(svc, req))
+        ResponseFuture::Disabled(self.inner.proxy(svc, req.map(BoxBody::new)))
     }
 }
 
-impl<R, P, S, Req> Future for ResponseFuture<R, P, S, Req>
+impl<R, P, S, B> Future for ResponseFuture<R, P, S, B>
 where
-    R: tower::retry::Policy<Req, P::Response, Error> + Clone,
-    P: Proxy<Req, S> + Clone,
+    R: tower::retry::Policy<http::Request<ReplayBody<B>>, P::Response, Error> + Clone,
+    P: Proxy<http::Request<BoxBody>, S> + Clone,
     S: tower::Service<P::Request> + Clone,
     S::Error: Into<Error>,
 {

linkerd/retry/src/replay.rs renamed to linkerd/http-retry/src/replay.rs

@@ -754,7 +754,7 @@ mod tests {
                 tx: Tx(tx),
                 initial,
                 replay,
-                _trace: linkerd_tracing::test::with_default_filter("linkerd_retry=debug"),
+                _trace: linkerd_tracing::test::with_default_filter("linkerd_http_retry=debug"),
             }
         }
     }