outbound: implement OutboundPolicy retries

Cargo.lock

@@ -1479,6 +1479,7 @@ dependencies = [
  "quickcheck",
  "thiserror",
  "tonic",
+ "tower",
 ]
 
 [[package]]
@@ -1894,8 +1895,7 @@ dependencies = [
 [[package]]
 name = "linkerd2-proxy-api"
 version = "0.11.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2348745f909668e6de2dbd175eeeac374887ffb33989a0e09766f1807b27cdfe"
+source = "git+https://github.com/linkerd/linkerd2-proxy-api?branch=eliza/retry-policy#7790fa05aab1c3c11793852a6d571845b1b3ead9"
 dependencies = [
  "h2",
  "http",

Cargo.toml

@@ -85,3 +85,7 @@ lto = true
 webpki = { git = "https://github.com/linkerd/webpki", branch = "cert-dns-names-0.22" }
 boring = { git = "https://github.com/cloudflare/boring" }
 tokio-boring = { git = "https://github.com/cloudflare/boring" }
+
+[patch.crates-io.linkerd2-proxy-api]
+git = "https://github.com/linkerd/linkerd2-proxy-api"
+branch = "eliza/retry-policy"

linkerd/app/core/src/errors/respond.rs

@@ -40,7 +40,7 @@ pub struct SyntheticHttpResponse {
     location: Option<HeaderValue>,
 }
 
-#[derive(Copy, Clone, Debug)]
+#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
 pub struct EmitHeaders(pub bool);
 
 #[derive(Clone, Debug)]

linkerd/app/integration/src/policy.rs

@@ -120,10 +120,12 @@ pub fn outbound_default(dst: impl ToString) -> outbound::OutboundPolicy {
                 http1: Some(proxy_protocol::Http1 {
                     routes: vec![route.clone()],
                     failure_accrual: None,
+                    retry_budget: None,
                 }),
                 http2: Some(proxy_protocol::Http2 {
                     routes: vec![route],
                     failure_accrual: None,
+                    retry_budget: None,
                 }),
                 opaque: Some(proxy_protocol::Opaque {
                     routes: vec![outbound_default_opaque_route(dst)],
@@ -134,28 +136,33 @@ pub fn outbound_default(dst: impl ToString) -> outbound::OutboundPolicy {
 }
 
 pub fn outbound_default_http_route(dst: impl ToString) -> outbound::HttpRoute {
-    use api::http_route;
     outbound::HttpRoute {
         metadata: Some(api::meta::Metadata {
             kind: Some(api::meta::metadata::Kind::Default("default".to_string())),
         }),
         hosts: Vec::new(),
         rules: vec![outbound::http_route::Rule {
-            matches: vec![http_route::HttpRouteMatch {
-                path: Some(http_route::PathMatch {
-                    kind: Some(http_route::path_match::Kind::Prefix("/".to_string())),
-                }),
-                headers: Vec::new(),
-                query_params: Vec::new(),
-                method: None,
-            }],
+            matches: vec![match_path_prefix("/")],
             filters: Vec::new(),
             backends: Some(http_first_available(std::iter::once(backend(dst)))),
             request_timeout: None,
+            retry_policy: None,
         }],
     }
 }
 
+pub fn match_path_prefix(path: impl ToString) -> api::http_route::HttpRouteMatch {
+    use api::http_route;
+    http_route::HttpRouteMatch {
+        path: Some(http_route::PathMatch {
+            kind: Some(http_route::path_match::Kind::Prefix(path.to_string())),
+        }),
+        headers: Vec::new(),
+        query_params: Vec::new(),
+        method: None,
+    }
+}
+
 pub fn outbound_default_opaque_route(dst: impl ToString) -> outbound::OpaqueRoute {
     use outbound::opaque_route::{self, distribution};
     outbound::OpaqueRoute {

linkerd/app/integration/src/tests/client_policy.rs

@@ -2,6 +2,8 @@ use crate::*;
 use linkerd2_proxy_api::{self as api};
 use policy::outbound::{self, proxy_protocol};
 
+mod retries;
+
 #[tokio::test]
 async fn default_http1_route() {
     let _trace = trace_init();
@@ -64,10 +66,12 @@ async fn empty_http1_route() {
                                 rules: Vec::new(),
                             }],
                             failure_accrual: None,
+                            retry_budget: None,
                         }),
                         http2: Some(proxy_protocol::Http2 {
                             routes: vec![policy::outbound_default_http_route(&dst)],
                             failure_accrual: None,
+                            retry_budget: None,
                         }),
                         opaque: Some(proxy_protocol::Opaque {
                             routes: vec![policy::outbound_default_opaque_route(&dst)],
@@ -149,6 +153,7 @@ async fn empty_http2_route() {
                         http1: Some(proxy_protocol::Http1 {
                             routes: vec![policy::outbound_default_http_route(&dst)],
                             failure_accrual: None,
+                            retry_budget: None,
                         }),
                         http2: Some(proxy_protocol::Http2 {
                             routes: vec![outbound::HttpRoute {
@@ -157,6 +162,7 @@ async fn empty_http2_route() {
                                 rules: Vec::new(),
                             }],
                             failure_accrual: None,
+                            retry_budget: None,
                         }),
                         opaque: Some(proxy_protocol::Opaque {
                             routes: vec![policy::outbound_default_opaque_route(&dst)],
@@ -224,6 +230,7 @@ async fn header_based_routing() {
                 policy::backend(dst),
             ))),
             request_timeout: None,
+            retry_policy: None,
         };
 
     let route = outbound::HttpRoute {
@@ -238,6 +245,7 @@ async fn header_based_routing() {
                     policy::backend(&dst_world),
                 ))),
                 request_timeout: None,
+                retry_policy: None,
             },
             // x-hello-city: sf | x-hello-city: san francisco
             mk_header_rule(
@@ -267,10 +275,12 @@ async fn header_based_routing() {
                         http1: Some(proxy_protocol::Http1 {
                             routes: vec![route.clone()],
                             failure_accrual: None,
+                            retry_budget: None,
                         }),
                         http2: Some(proxy_protocol::Http2 {
                             routes: vec![route],
                             failure_accrual: None,
+                            retry_budget: None,
                         }),
                         opaque: Some(proxy_protocol::Opaque {
                             routes: vec![policy::outbound_default_opaque_route(&dst_world)],
@@ -400,8 +410,8 @@ async fn path_based_routing() {
             backends: Some(policy::http_first_available(std::iter::once(
                 policy::backend(dst),
             ))),
-
             request_timeout: None,
+            retry_policy: None,
         };
 
     let route = outbound::HttpRoute {
@@ -416,6 +426,7 @@ async fn path_based_routing() {
                     policy::backend(&dst_world),
                 ))),
                 request_timeout: None,
+                retry_policy: None,
             },
             // /goodbye/*
             mk_path_rule(
@@ -450,10 +461,12 @@ async fn path_based_routing() {
                         http1: Some(proxy_protocol::Http1 {
                             routes: vec![route.clone()],
                             failure_accrual: None,
+                            retry_budget: None,
                         }),
                         http2: Some(proxy_protocol::Http2 {
                             routes: vec![route],
                             failure_accrual: None,
+                            retry_budget: None,
                         }),
                         opaque: Some(proxy_protocol::Opaque {
                             routes: vec![policy::outbound_default_opaque_route(&dst_world)],