feat(tonic-xds): drive gRPC retry config from RDS RouteAction.retry_policy

tonic-xds/src/client/channel.rs

@@ -46,7 +46,9 @@ use xds_client::{
     ClientConfig, MetricsRecorder, Node, ProstCodec, TokioRuntime, TonicTransportBuilder, XdsClient,
 };
 
-use crate::client::retry::{GrpcRetryPolicy, RetryLayer};
+#[cfg(test)]
+use crate::client::retry::GrpcRetryPolicy;
+use crate::client::retry::{GrpcRetrySharedConfig, RetryLayer};
 
 /// Configuration for building [`XdsChannel`] / [`XdsChannelGrpc`].
 #[derive(Clone, Debug)]
@@ -367,6 +369,12 @@ impl XdsChannelBuilder {
         resource_manager: XdsResourceManager,
     ) -> XdsChannelGrpc {
         let router: Arc<dyn Router> = Arc::new(XdsRouter::new(&cache));
+
+        // Fallback retry config for requests that carry no per-route config
+        // (non-xDS callers, or a route with no retry policy). Per-route configs
+        // come from RDS via the request's `RouteDecision`; see `RetryLayer`.
+        let retry_layer = RetryLayer::new(Arc::new(GrpcRetrySharedConfig::default()));
+
         #[cfg(feature = "_tls-any")]
         let discovery: Arc<
             dyn ClusterDiscovery<EndpointAddress, EndpointChannel<Channel>>,
@@ -378,15 +386,13 @@ impl XdsChannelBuilder {
         let discovery: Arc<
             dyn ClusterDiscovery<EndpointAddress, EndpointChannel<Channel>>,
         > = Arc::new(XdsClusterDiscovery::new(cache, GrpcMakeConnector::new()));
-        let retry_policy = GrpcRetryPolicy::default();
 
         let resources = Arc::new(XdsChannelResources {
             _resource_manager: resource_manager,
             _xds_client: xds_client,
         });
 
         let routing_layer = XdsRoutingLayer::new(router, self.pre_route.clone(), self.authority());
-        let retry_layer = RetryLayer::new(retry_policy);
         let cluster_registry = Arc::new(ClusterClientRegistryGrpc::new());
         let lb_service = XdsLbService::new(cluster_registry, discovery);
         let inner = ServiceBuilder::new()
@@ -422,7 +428,7 @@ impl XdsChannelBuilder {
         interceptor: Option<Arc<dyn PreRouteInterceptor>>,
     ) -> XdsChannelGrpc {
         let routing_layer = XdsRoutingLayer::new(router, interceptor, self.authority());
-        let retry_layer = RetryLayer::new(retry_policy);
+        let retry_layer = RetryLayer::new(retry_policy.into_shared());
         let cluster_registry = Arc::new(ClusterClientRegistryGrpc::new());
         let lb_service = XdsLbService::new(cluster_registry, discovery);
         let inner = ServiceBuilder::new()
@@ -522,11 +528,12 @@ mod tests {
         fn route(
             &self,
             _input: &RouteInput<'_>,
-            _config: &crate::xds::resource::route_config::RouteConfigResource,
+            _config: &crate::client::route::RoutingSnapshot,
         ) -> Result<RouteDecision, crate::xds::routing::RoutingError> {
             Ok(RouteDecision {
                 cluster: "test-cluster".to_string(),
                 request_hash: None,
+                retry_config: None,
             })
         }
     }
@@ -680,9 +687,7 @@ mod tests {
 
         let retry_policy = GrpcRetryPolicy::new(
             RetryConfig::new().num_retries(1),
-            GrpcRetryClassifier {
-                retry_on: vec![tonic::Code::Unavailable],
-            },
+            GrpcRetryClassifier::new(vec![tonic::Code::Unavailable]),
         );
 
         let xds_channel = XdsChannelBuilder::new(test_config()).build_grpc_channel_from_parts(
@@ -734,6 +739,7 @@ mod tests {
                         match_fraction: None,
                     },
                     action: RouteConfigAction::Cluster(cluster_name.to_string()),
+                    retry_config: None,
                 }],
             }],
             metadata: Default::default(),

tonic-xds/src/client/circuit_breaking.rs

@@ -731,6 +731,7 @@ mod tests {
         request.extensions_mut().insert(RouteDecision {
             cluster: CLUSTER.to_string(),
             request_hash: None,
+            retry_config: None,
         });
         request
     }
@@ -984,9 +985,7 @@ mod tests {
                 RetryBackoffConfig::new(Duration::from_millis(1))
                     .max_interval(Duration::from_millis(1)),
             ),
-            GrpcRetryClassifier {
-                retry_on: vec![Code::Unavailable],
-            },
+            GrpcRetryClassifier::new(vec![Code::Unavailable]),
         );
         let calls = Arc::new(AtomicU32::new(0));
         let call_counter = calls.clone();
@@ -998,7 +997,7 @@ mod tests {
             },
         );
         let mut service = tower::ServiceBuilder::new()
-            .layer(RetryLayer::new(policy))
+            .layer(RetryLayer::new(policy.into_shared()))
             .layer(CircuitBreakingLayer::new(breakers.clone()))
             .service(service);
 

tonic-xds/src/client/loadbalance/pickers/ring_hash.rs

@@ -223,6 +223,7 @@ mod tests {
         r.extensions_mut().insert(RouteDecision {
             cluster: "c".to_string(),
             request_hash: hash,
+            retry_config: None,
         });
         r
     }