outbound: implement OutboundPolicies backend request timeouts (#2419)

Depends on #2418 

The latest proxy-api release, v0.10.0, adds fields to the
`OutboundPolicies` API for configuring HTTP request timeouts, based on
the proposed changes to HTTPRoute in kubernetes-sigs/gateway-api#1997.
PR #2418 updates the proxy to depend on the new proxy-api release, and
implements the `Rule.request_timeout` field added to the API. However,
that branch does *not* add a timeout for the
`RouteBackend.request_timeout` field. This branch changes the proxy to
apply the backend request timeout when configured by the policy
controller.

This branch implements `RouteBackend.request_timeout` by adding an
additional timeout layer in the `MatchedBackend` stack. This applies the
per-backend timeout once a backend is selected for a route. I've also
added stack tests for the interaction between the request and backend
request timeouts.

Note that once retries are added to client policy stacks, it may be
necessary to move the backend request timeout to ensure it occurs
"below" retries, depending on where the retry middleware ends up being
located in the proxy stack.

Author: hawkw

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

@@ -16,6 +16,7 @@ pub(crate) struct Backend<T, F> {
     pub(crate) route_ref: RouteRef,
     pub(crate) concrete: Concrete<T>,
     pub(crate) filters: Arc<[F]>,
+    pub(crate) request_timeout: Option<std::time::Duration>,
 }
 
 pub(crate) type MatchedBackend<T, M, F> = super::Matched<M, Backend<T, F>>;
@@ -37,6 +38,7 @@ impl<T: Clone, F> Clone for Backend<T, F> {
             route_ref: self.route_ref.clone(),
             filters: self.filters.clone(),
             concrete: self.concrete.clone(),
+            request_timeout: self.request_timeout,
         }
     }
 }
@@ -107,6 +109,7 @@ where
                      }| concrete,
                 )
                 .push(filters::NewApplyFilters::<Self, _, _>::layer())
+                .push(http::NewTimeout::layer())
                 .push(count_reqs::NewCountRequests::layer_via(ExtractMetrics {
                     metrics: metrics.clone(),
                 }))
@@ -116,6 +119,12 @@ where
     }
 }
 
+impl<T, M, F> svc::Param<http::ResponseTimeout> for MatchedBackend<T, M, F> {
+    fn param(&self) -> http::ResponseTimeout {
+        http::ResponseTimeout(self.params.request_timeout)
+    }
+}
+
 impl<T> filters::Apply for Http<T> {
     #[inline]
     fn apply<B>(&self, req: &mut ::http::Request<B>) -> Result<()> {

linkerd/app/outbound/src/http/logical/policy/router.rs

@@ -179,6 +179,7 @@ where
                 route_ref: route_ref.clone(),
                 filters,
                 concrete,
+                request_timeout: rb.request_timeout,
             }
         };
 

linkerd/app/outbound/src/http/logical/tests.rs

@@ -306,9 +306,9 @@ async fn route_request_timeout() {
 
     let (_route_tx, routes) = {
         let backend = default_backend(&dest);
-        let mut route = default_route(backend.clone());
-        // set the request timeout on the route.
-        route.rules[0].policy.request_timeout = Some(REQUEST_TIMEOUT);
+        // Set a request timeout for the route, and no backend request timeout
+        // on the backend.
+        let route = timeout_route(backend.clone(), Some(REQUEST_TIMEOUT), None);
         watch::channel(Routes::Policy(policy::Params::Http(policy::HttpParams {
             addr: dest.into(),
             meta: ParentRef(client_policy::Meta::new_default("parent")),
@@ -345,6 +345,88 @@ async fn route_request_timeout() {
     ));
 }
 
+#[tokio::test(flavor = "current_thread")]
+async fn backend_request_timeout() {
+    tokio::time::pause();
+    let _trace = trace::test::trace_init();
+    // must be less than the `default_config` failfast timeout, or we'll hit
+    // that instead.
+    const ROUTE_REQUEST_TIMEOUT: Duration = std::time::Duration::from_secs(2);
+    const BACKEND_REQUEST_TIMEOUT: Duration = std::time::Duration::from_secs(1);
+
+    let addr = SocketAddr::new([192, 0, 2, 41].into(), PORT);
+    let dest: NameAddr = format!("{AUTHORITY}:{PORT}")
+        .parse::<NameAddr>()
+        .expect("dest addr is valid");
+    let (svc, mut handle) = tower_test::mock::pair();
+    let connect = HttpConnect::default().service(addr, svc);
+    let resolve = support::resolver().endpoint_exists(dest.clone(), addr, Default::default());
+    let (rt, _shutdown) = runtime();
+    let stack = Outbound::new(default_config(), rt)
+        .with_stack(connect)
+        .push_http_cached(resolve)
+        .into_inner();
+
+    let (_route_tx, routes) = {
+        let backend = default_backend(&dest);
+        // Set both a route request timeout and a backend request timeout.
+        let route = timeout_route(
+            backend.clone(),
+            Some(ROUTE_REQUEST_TIMEOUT),
+            Some(BACKEND_REQUEST_TIMEOUT),
+        );
+        watch::channel(Routes::Policy(policy::Params::Http(policy::HttpParams {
+            addr: dest.into(),
+            meta: ParentRef(client_policy::Meta::new_default("parent")),
+            backends: Arc::new([backend]),
+            routes: Arc::new([route]),
+            failure_accrual: client_policy::FailureAccrual::None,
+        })))
+    };
+    let target = Target {
+        num: 1,
+        version: http::Version::H2,
+        routes,
+    };
+    let svc = stack.new_service(target);
+
+    handle.allow(1);
+    let rsp = send_req(svc.clone(), http::Request::get("/"));
+    serve_req(&mut handle, mk_rsp(StatusCode::OK, "good")).await;
+    assert_eq!(
+        rsp.await.expect("request must succeed").status(),
+        http::StatusCode::OK
+    );
+
+    // Now, time out...
+    let rsp = send_req(svc.clone(), http::Request::get("/"));
+    // Wait until we actually get the request --- this timeout only starts once
+    // the service has been acquired.
+    handle.allow(1);
+    let (_, send_rsp) = handle
+        .next_request()
+        .await
+        .expect("service must receive request");
+    tokio::time::sleep(BACKEND_REQUEST_TIMEOUT + Duration::from_millis(1)).await;
+    // Still send a response, so that if we didn't hit the backend timeout
+    // timeout, we don't hit the route timeout and succeed incorrectly.
+    send_rsp.send_response(mk_rsp(StatusCode::OK, "good"));
+    let error = rsp.await.expect_err("request must fail with a timeout");
+    assert!(errors::is_caused_by::<http::timeout::ResponseTimeoutError>(
+        error.as_ref()
+    ));
+
+    // The route request timeout should still apply to time spent before
+    // the backend is acquired.
+    let rsp = send_req(svc.clone(), http::Request::get("/"));
+    tokio::time::sleep(ROUTE_REQUEST_TIMEOUT + Duration::from_millis(1)).await;
+    handle.allow(1);
+    let error = rsp.await.expect_err("request must fail with a timeout");
+    assert!(errors::is_caused_by::<http::timeout::ResponseTimeoutError>(
+        error.as_ref()
+    ));
+}
+
 #[derive(Clone, Debug)]
 struct Target {
     num: usize,
@@ -518,3 +600,33 @@ fn default_route(backend: client_policy::Backend) -> client_policy::http::Route
         }],
     }
 }
+
+fn timeout_route(
+    backend: client_policy::Backend,
+    route_timeout: Option<Duration>,
+    backend_timeout: Option<Duration>,
+) -> client_policy::http::Route {
+    use client_policy::{
+        http::{self, Filter, Policy, Route, Rule},
+        Meta, RouteBackend, RouteDistribution,
+    };
+    use once_cell::sync::Lazy;
+    static NO_FILTERS: Lazy<Arc<[Filter]>> = Lazy::new(|| Arc::new([]));
+    Route {
+        hosts: vec![],
+        rules: vec![Rule {
+            matches: vec![http::r#match::MatchRequest::default()],
+            policy: Policy {
+                meta: Meta::new_default("test_route"),
+                filters: NO_FILTERS.clone(),
+                failure_policy: Default::default(),
+                request_timeout: route_timeout,
+                distribution: RouteDistribution::FirstAvailable(Arc::new([RouteBackend {
+                    filters: NO_FILTERS.clone(),
+                    backend,
+                    request_timeout: backend_timeout,
+                }])),
+            },
+        }],
+    }
+}