fix intermittent circuit breaking test flake by adding CDS settle duration

framework/xds_k8s_testcase.py

@@ -745,6 +745,54 @@ def _check_config() -> bool:
                 f"Timeout waiting for RDS to update to cluster {expected_cluster_name}"
             )
 
+    def assertCdsCircuitBreakerRequestsLimit(
+        self,
+        test_client: XdsTestClient,
+        backend_service_name: str,
+        expected_max_requests: int,
+        *,
+        retry_timeout: datetime.timedelta = datetime.timedelta(minutes=5),
+        retry_wait: datetime.timedelta = datetime.timedelta(seconds=5),
+    ) -> None:
+        logger.info(
+            "Waiting for CDS update to backend service %s to have maxRequests %d",
+            backend_service_name,
+            expected_max_requests,
+        )
+
+        def _check_config() -> bool:
+            config = test_client.csds.fetch_client_status_parsed()
+            if not config or not config.cds:
+                return False
+
+            for cluster in config.cds:
+                cluster_name = cluster.get("name", "")
+                alt_stat_name = cluster.get("altStatName", "")
+                if (
+                    backend_service_name in cluster_name
+                    or backend_service_name in alt_stat_name
+                ):
+                    cb = cluster.get("circuitBreakers", {})
+                    thresholds = cb.get("thresholds", [])
+                    if thresholds:
+                        max_requests = thresholds[0].get("maxRequests")
+                        if max_requests == expected_max_requests:
+                            return True
+            return False
+
+        retryer = retryers.constant_retryer(
+            wait_fixed=retry_wait,
+            timeout=retry_timeout,
+        )
+
+        try:
+            retryer(_check_config)
+        except retryers.RetryError:
+            self.fail(
+                f"Timeout waiting for CDS of {backend_service_name} "
+                f"to update to maxRequests={expected_max_requests}"
+            )
+
     def assertRouteConfigUpdateTrafficHandoff(
         self,
         test_client: XdsTestClient,

tests/circuit_breaking_test.py

@@ -219,7 +219,19 @@ def test_circuit_breaking(self) -> None:
                 test_client, (default_test_server, alternate_test_server)
             )
 
-        with self.subTest("11_configure_client_with_keep_open"):
+        with self.subTest("11_wait_for_circuit_breaker_configs_propagation"):
+            self.assertCdsCircuitBreakerRequestsLimit(
+                test_client,
+                self.td.backend_service.name,
+                _INITIAL_UNARY_MAX_REQUESTS,
+            )
+            self.assertCdsCircuitBreakerRequestsLimit(
+                test_client,
+                self.td.alternative_backend_service.name,
+                _INITIAL_EMPTY_MAX_REQUESTS,
+            )
+
+        with self.subTest("12_configure_client_with_keep_open"):
             test_client.update_config.configure(
                 rpc_types=grpc_testing.RPC_TYPES_BOTH_CALLS,
                 metadata={
@@ -242,7 +254,7 @@ def test_circuit_breaking(self) -> None:
         # batch thereby resulting in a shortfall of QPS number of RPCs inflight.
         after_steady_state_shortfall = _QPS
 
-        with self.subTest("12_client_reaches_target_steady_state"):
+        with self.subTest("13_client_reaches_target_steady_state"):
             self.assertClientEventuallyReachesSteadyState(
                 test_client,
                 rpc_type=grpc_testing.RPC_TYPE_UNARY_CALL,
@@ -256,12 +268,21 @@ def test_circuit_breaking(self) -> None:
                 after_steady_state_allowed_shortfall_count=after_steady_state_shortfall,
             )
 
-        with self.subTest("13_increase_backend_max_requests"):
+        with self.subTest("14_increase_backend_max_requests"):
             self.td.backend_service_patch_backends(
                 circuit_breakers={"maxRequests": _UPDATED_UNARY_MAX_REQUESTS}
             )
 
-        with self.subTest("14_client_reaches_increased_steady_state"):
+        with self.subTest(
+            "15_wait_for_increased_circuit_breaker_config_propagation"
+        ):
+            self.assertCdsCircuitBreakerRequestsLimit(
+                test_client,
+                self.td.backend_service.name,
+                _UPDATED_UNARY_MAX_REQUESTS,
+            )
+
+        with self.subTest("16_client_reaches_increased_steady_state"):
             self.assertClientEventuallyReachesSteadyState(
                 test_client,
                 rpc_type=grpc_testing.RPC_TYPE_UNARY_CALL,