fix(policy): respect timeout field on gateway API HTTPRoutes (#14750)

Fixes #14713

The [Linkerd documentation](https://linkerd.io/2-edge/reference/timeouts/#configuring-timeouts) specifies that:

> If the [request timeout](https://gateway-api.sigs.k8s.io/api-types/httproute/#timeouts-optional) field is set on an HTTPRoute resource, it will be used as the timeout.linkerd.io/request timeout. However, if both the field and the annotation are specified, the annotation will take priority.

This is true for policy.linkerd.io/HTTPRoute resources but not for `gateway.networking.k8s.io/HTTPRoute` resources.  

We update the policy controller to consider the request timeout field on `gateway.networking.k8s.io/HTTPRoute` and add API tests.

Signed-off-by: Alex Leong <[email protected]>

Author: adleong

policy-controller/k8s/index/src/outbound/index/http.rs

@@ -159,7 +159,7 @@ fn convert_gateway_rule(
     rule: gateway::HTTPRouteRules,
     cluster: &ClusterInfo,
     resource_info: &HashMap<ResourceRef, ResourceInfo>,
-    timeouts: RouteTimeouts,
+    mut timeouts: RouteTimeouts,
     retry: Option<RouteRetry<HttpRetryCondition>>,
 ) -> Result<OutboundRouteRule<HttpRouteMatch, HttpRetryCondition>> {
     let matches = rule
@@ -183,6 +183,19 @@ fn convert_gateway_rule(
         .map(convert_gateway_filter)
         .collect::<Result<_>>()?;
 
+    timeouts.request = timeouts.request.or_else(|| {
+        rule.timeouts.as_ref().and_then(|timeouts| {
+            let timeout = parse_duration(timeouts.request.as_ref()?).ok()?;
+
+            // zero means "no timeout", per GEP-1742
+            if timeout == time::Duration::ZERO {
+                return None;
+            }
+
+            Some(timeout)
+        })
+    });
+
     Ok(OutboundRouteRule {
         matches,
         backends,

policy-test/tests/outbound_api_http.rs

@@ -1,3 +1,5 @@
+use std::time::Duration;
+
 use futures::StreamExt;
 use linkerd2_proxy_api::{self as api, outbound};
 use linkerd_policy_controller_k8s_api::{self as k8s, gateway, policy};
@@ -701,6 +703,155 @@ async fn http_route_retries_and_timeouts() {
     test::<policy::EgressNetwork, policy::HttpRoute>().await;
 }
 
+#[tokio::test(flavor = "current_thread")]
+async fn http_route_linkerd_timeouts() {
+    async fn test<P: TestParent>() {
+        tracing::debug!(
+            parent = %P::kind(&P::DynamicType::default()),
+        );
+        with_temp_ns(|client, ns| async move {
+            // Create a parent
+            let parent = create(&client, P::make_parent(&ns)).await;
+            let port = 4191;
+            // Create a backend
+            let backend_port = 8888;
+            let backend = match P::make_backend(&ns) {
+                Some(b) => create(&client, b).await,
+                None => parent.clone(),
+            };
+
+            let route = policy::HttpRoute {
+                metadata: k8s::ObjectMeta {
+                    namespace: Some(ns.to_string()),
+                    name: Some("foo-route".to_string()),
+                    ..Default::default()
+                },
+                spec: policy::HttpRouteSpec {
+                    parent_refs: Some(vec![parent.obj_ref()]),
+                    hostnames: None,
+                    rules: Some(vec![policy::httproute::HttpRouteRule {
+                        matches: Some(vec![]),
+                        filters: None,
+                        timeouts: Some(policy::httproute::HttpRouteTimeouts {
+                            request: Some(Duration::from_secs(5).into()),
+                            backend_request: None,
+                        }),
+                        backend_refs: Some(vec![backend.backend_ref(backend_port)]),
+                    }]),
+                },
+                status: None,
+            };
+
+            let route = create(&client, route).await;
+            await_route_accepted(&client, &route).await;
+
+            let mut rx = retry_watch_outbound_policy(&client, &ns, parent.ip(), port).await;
+            let config = rx
+                .next()
+                .await
+                .expect("watch must not fail")
+                .expect("watch must return an initial config");
+            tracing::trace!(?config);
+            assert_resource_meta(&config.metadata, parent.obj_ref(), port);
+
+            policy::HttpRoute::routes(&config, |routes| {
+                let outbound_route = routes.first().expect("route must exist");
+                assert!(route.meta_eq(policy::HttpRoute::extract_meta(outbound_route)));
+                let rule = assert_singleton(&outbound_route.rules);
+                let timeout = rule
+                    .timeouts
+                    .as_ref()
+                    .expect("timeouts expected")
+                    .request
+                    .as_ref()
+                    .expect("request timeout expected");
+                assert_eq!(timeout.seconds, 5);
+            });
+        })
+        .await;
+    }
+
+    test::<k8s::Service>().await;
+    test::<policy::EgressNetwork>().await;
+}
+
+// The timeout field on HTTPRoute is only available in Gateway API v1.2+.
+#[cfg(feature = "gateway-api-experimental")]
+#[tokio::test(flavor = "current_thread")]
+async fn http_route_gateway_timeouts() {
+    async fn test<P: TestParent>() {
+        tracing::debug!(
+            parent = %P::kind(&P::DynamicType::default()),
+        );
+        with_temp_ns(|client, ns| async move {
+            // Create a parent
+            let parent = create(&client, P::make_parent(&ns)).await;
+            let port = 4191;
+            // Create a backend
+            let backend_port = 8888;
+            let backend = match P::make_backend(&ns) {
+                Some(b) => create(&client, b).await,
+                None => parent.clone(),
+            };
+
+            let route = gateway::HTTPRoute {
+                metadata: k8s::ObjectMeta {
+                    namespace: Some(ns.to_string()),
+                    name: Some("foo-route".to_string()),
+                    ..Default::default()
+                },
+                spec: gateway::HTTPRouteSpec {
+                    parent_refs: Some(vec![parent.obj_ref()]),
+                    hostnames: None,
+                    rules: Some(vec![gateway::HTTPRouteRules {
+                        name: None,
+                        matches: Some(vec![]),
+                        filters: None,
+                        timeouts: Some(gateway::HTTPRouteRulesTimeouts {
+                            request: Some("5s".to_string()),
+                            backend_request: None,
+                        }),
+                        retry: None,
+                        backend_refs: Some(vec![backend.backend_ref(backend_port)]),
+                        session_persistence: None,
+                    }]),
+                },
+                status: None,
+            };
+
+            let route = create(&client, route).await;
+            await_route_accepted(&client, &route).await;
+
+            let mut rx = retry_watch_outbound_policy(&client, &ns, parent.ip(), port).await;
+            let config = rx
+                .next()
+                .await
+                .expect("watch must not fail")
+                .expect("watch must return an initial config");
+            tracing::trace!(?config);
+            assert_resource_meta(&config.metadata, parent.obj_ref(), port);
+
+            gateway::HTTPRoute::routes(&config, |routes| {
+                let outbound_route = routes.first().expect("route must exist");
+                assert!(route.meta_eq(gateway::HTTPRoute::extract_meta(outbound_route)));
+                let rule = assert_singleton(&outbound_route.rules);
+                let timeout = rule
+                    .timeouts
+                    .as_ref()
+                    .expect("timeouts expected")
+                    .request
+                    .as_ref()
+                    .expect("request timeout expected");
+                assert_eq!(timeout.seconds, 5);
+            });
+        })
+        .await;
+    }
+
+    test::<k8s::Service>().await;
+    test::<policy::EgressNetwork>().await;
+}
+
 #[tokio::test(flavor = "current_thread")]
 async fn parent_retries_and_timeouts() {
     async fn test<P: TestParent, R: TestRoute<Route = outbound::HttpRoute>>() {

testutil/test_helper.go

@@ -22,7 +22,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 )
 
-const GATEWAY_API_VERSION = "v1.1.1"
+const GATEWAY_API_VERSION = "v1.2.0"
 
 // TestHelper provides helpers for running the linkerd integration tests.
 type TestHelper struct {