Merge branch 'xds_watcher_api_change' into xds_watcher_api_change2

markdroth · Dec 27, 2024 · 0da1867 · 0da1867
2 parents 425434a + a4badf1
commit 0da1867
Show file tree

Hide file tree

Showing 7 changed files with 55 additions and 67 deletions.
diff --git a/test/cpp/end2end/xds/xds_cluster_end2end_test.cc b/test/cpp/end2end/xds/xds_cluster_end2end_test.cc
@@ -380,14 +380,10 @@ TEST_P(CdsDeletionTest, ClusterDeleted) {
   // Unset CDS resource.
   balancer_->ads_service()->UnsetResource(kCdsTypeUrl, kDefaultClusterName);
   // Wait for RPCs to start failing.
-  SendRpcsUntil(DEBUG_LOCATION, [](const RpcResult& result) {
-    if (result.status.ok()) return true;  // Keep going.
-    EXPECT_EQ(StatusCode::UNAVAILABLE, result.status.error_code());
-    EXPECT_EQ(absl::StrCat("CDS resource ", kDefaultClusterName,
-                           ": does not exist (node ID:xds_end2end_test)"),
-              result.status.error_message());
-    return false;
-  });
+  SendRpcsUntilFailure(
+      DEBUG_LOCATION, StatusCode::UNAVAILABLE,
+      absl::StrCat("CDS resource ", kDefaultClusterName,
+                   ": does not exist \\(node ID:xds_end2end_test\\)"));
   // Make sure we ACK'ed the update.
   auto response_state = balancer_->ads_service()->cds_response_state();
   ASSERT_TRUE(response_state.has_value());
@@ -543,13 +539,7 @@ TEST_P(EdsTest, LocalityBecomesEmptyWithDeactivatedChildStateUpdate) {
       "EDS resource eds_service_name: contains empty localities: "
       "\\[\\{region=\"xds_default_locality_region\", "
       "zone=\"xds_default_locality_zone\", sub_zone=\"locality0\"\\}\\]";
-  SendRpcsUntil(DEBUG_LOCATION, [&](const RpcResult& result) {
-    if (result.status.ok()) return true;
-    EXPECT_EQ(result.status.error_code(), StatusCode::UNAVAILABLE);
-    EXPECT_THAT(result.status.error_message(),
-                ::testing::MatchesRegex(kErrorMessage));
-    return false;
-  });
+  SendRpcsUntilFailure(DEBUG_LOCATION, StatusCode::UNAVAILABLE, kErrorMessage);
   // Shut down backend.  This triggers a connectivity state update from the
   // deactivated child of the weighted_target policy.
   ShutdownAllBackends();

diff --git a/test/cpp/end2end/xds/xds_core_end2end_test.cc b/test/cpp/end2end/xds/xds_core_end2end_test.cc
@@ -508,16 +508,10 @@ TEST_P(TimeoutTest, CdsSecondResourceNotPresentInRequest) {
   balancer_->ads_service()->SetRdsResource(route_config);
   // New cluster times out.
   // May need to wait a bit for the change to propagate to the client.
-  SendRpcsUntil(
-      DEBUG_LOCATION,
-      [&](const RpcResult& result) {
-        if (result.status.ok()) return true;  // Keep going.
-        EXPECT_EQ(StatusCode::UNAVAILABLE, result.status.error_code());
-        EXPECT_EQ(absl::StrCat("CDS resource ", kNewClusterName,
-                               ": does not exist (node ID:xds_end2end_test)"),
-                  result.status.error_message());
-        return false;
-      },
+  SendRpcsUntilFailure(
+      DEBUG_LOCATION, StatusCode::UNAVAILABLE,
+      absl::StrCat("CDS resource ", kNewClusterName,
+                   ": does not exist \\(node ID:xds_end2end_test\\)"),
       /*timeout_ms=*/30000, RpcOptions().set_timeout_ms(4000));
 }
 
@@ -561,17 +555,11 @@ TEST_P(TimeoutTest, EdsSecondResourceNotPresentInRequest) {
   balancer_->ads_service()->SetRdsResource(route_config);
   // New EDS resource times out.
   // May need to wait a bit for the RDS change to propagate to the client.
-  SendRpcsUntil(
-      DEBUG_LOCATION,
-      [](const RpcResult& result) {
-        if (result.status.ok()) return true;  // Keep going.
-        EXPECT_EQ(StatusCode::UNAVAILABLE, result.status.error_code());
-        EXPECT_EQ(result.status.error_message(),
-                  "no children in weighted_target policy: "
-                  "EDS resource eds_service_name_does_not_exist: "
-                  "does not exist (node ID:xds_end2end_test)");
-        return false;
-      },
+  SendRpcsUntilFailure(
+      DEBUG_LOCATION, StatusCode::UNAVAILABLE,
+      "no children in weighted_target policy: "
+      "EDS resource eds_service_name_does_not_exist: "
+      "does not exist \\(node ID:xds_end2end_test\\)",
       /*timeout_ms=*/30000,
       RpcOptions().set_rpc_method(METHOD_ECHO1).set_timeout_ms(4000));
 }

diff --git a/test/cpp/end2end/xds/xds_end2end_test.cc b/test/cpp/end2end/xds/xds_end2end_test.cc
@@ -395,19 +395,11 @@ class XdsSecurityTest : public XdsEnd2endTest {
     balancer_->ads_service()->SetCdsResource(cluster);
     // The updates might take time to have an effect, so use a retry loop.
     if (test_expects_failure) {
-      SendRpcsUntil(
-          DEBUG_LOCATION,
-          [&](const RpcResult& result) {
-            if (result.status.ok()) {
-              LOG(ERROR) << "RPC succeeded. Failure expected. Trying again.";
-              return true;
-            }
-            EXPECT_EQ(result.status.error_code(), StatusCode::UNAVAILABLE);
-            // TODO(yashkt): Change individual test cases to expect the exact
-            // error message here.
-            return false;
-          },
-          /* timeout_ms= */ 20 * 1000, RpcOptions().set_timeout_ms(5000));
+      SendRpcsUntilFailure(DEBUG_LOCATION, StatusCode::UNAVAILABLE,
+                           // TODO(yashkt): Change individual test cases to
+                           // expect the exact error message here.
+                           ".*", /*timeout_ms=*/20 * 1000,
+                           RpcOptions().set_timeout_ms(5000));
     } else {
       backends_[backend_index_]->backend_service()->ResetCounters();
       SendRpcsUntil(

diff --git a/test/cpp/end2end/xds/xds_end2end_test_lib.cc b/test/cpp/end2end/xds/xds_end2end_test_lib.cc
@@ -681,6 +681,26 @@ void XdsEnd2endTest::CheckRpcSendFailure(
       << debug_location.file() << ":" << debug_location.line();
 }
 
+void XdsEnd2endTest::SendRpcsUntilFailure(
+    const grpc_core::DebugLocation& debug_location, StatusCode expected_status,
+    absl::string_view expected_message_regex, int timeout_ms,
+    const RpcOptions& rpc_options) {
+  SendRpcsUntil(
+      debug_location,
+      [&](const RpcResult& result) {
+        // Might still succeed if channel hasn't yet seen the server go down.
+        if (result.status.ok()) return true;
+        // RPC failed.  Make sure the failure status is as expected and stop.
+        EXPECT_EQ(result.status.error_code(), expected_status)
+            << debug_location.file() << ":" << debug_location.line();
+        EXPECT_THAT(result.status.error_message(),
+                    ::testing::MatchesRegex(expected_message_regex))
+            << debug_location.file() << ":" << debug_location.line();
+        return false;
+      },
+      timeout_ms, rpc_options);
+}
+
 size_t XdsEnd2endTest::SendRpcsAndCountFailuresWithMessage(
     const grpc_core::DebugLocation& debug_location, size_t num_rpcs,
     StatusCode expected_status, absl::string_view expected_message_prefix,

diff --git a/test/cpp/end2end/xds/xds_end2end_test_lib.h b/test/cpp/end2end/xds/xds_end2end_test_lib.h
@@ -758,6 +758,14 @@ class XdsEnd2endTest : public ::testing::TestWithParam<XdsTestType>,
                            absl::string_view expected_message_regex,
                            const RpcOptions& rpc_options = RpcOptions());
 
+  // Sends RPCs until either a timeout or an RPC fail, in which case the
+  // failure must match the specified status and message regex.
+  void SendRpcsUntilFailure(const grpc_core::DebugLocation& debug_location,
+                            StatusCode expected_status,
+                            absl::string_view expected_message_regex,
+                            int timeout_ms = 15000,
+                            const RpcOptions& rpc_options = RpcOptions());
+
   // Sends num_rpcs RPCs, counting how many of them fail with a message
   // matching the specified expected_message_prefix.
   // Any failure with a non-matching status or message is a test failure.

diff --git a/test/cpp/end2end/xds/xds_gcp_authn_end2end_test.cc b/test/cpp/end2end/xds/xds_gcp_authn_end2end_test.cc
@@ -234,16 +234,10 @@ TEST_P(XdsGcpAuthnEnd2endTest, CacheRetainedAcrossXdsUpdates) {
       balancer_.get(), BuildListenerWithGcpAuthnFilter(), route_config);
   // Send RPCs with the header "foo" and wait for them to start failing.
   // When they do, we know that the client has seen the update.
-  SendRpcsUntil(
-      DEBUG_LOCATION,
-      [&](const RpcResult& result) {
-        if (result.status.ok()) return true;
-        EXPECT_EQ(StatusCode::UNAVAILABLE, result.status.error_code());
-        EXPECT_EQ("Matching route has inappropriate action",
-                  result.status.error_message());
-        return false;
-      },
-      /*timeout_ms=*/15000, RpcOptions().set_metadata({{"foo", "bar"}}));
+  SendRpcsUntilFailure(DEBUG_LOCATION, StatusCode::UNAVAILABLE,
+                       "Matching route has inappropriate action",
+                       /*timeout_ms=*/15000,
+                       RpcOptions().set_metadata({{"foo", "bar"}}));
   // Now send an RPC without the header, which will go through the new
   // instance of the GCP auth filter.
   CheckRpcSendOk(DEBUG_LOCATION);

diff --git a/test/cpp/end2end/xds/xds_routing_end2end_test.cc b/test/cpp/end2end/xds/xds_routing_end2end_test.cc
@@ -100,14 +100,10 @@ TEST_P(LdsDeletionTest, ListenerDeleted) {
   // Unset LDS resource.
   balancer_->ads_service()->UnsetResource(kLdsTypeUrl, kServerName);
   // Wait for RPCs to start failing.
-  SendRpcsUntil(DEBUG_LOCATION, [](const RpcResult& result) {
-    if (result.status.ok()) return true;  // Keep going.
-    EXPECT_EQ(result.status.error_code(), StatusCode::UNAVAILABLE);
-    EXPECT_EQ(result.status.error_message(),
-              absl::StrCat("empty address list (LDS resource ", kServerName,
-                           ": does not exist (node ID:xds_end2end_test))"));
-    return false;
-  });
+  SendRpcsUntilFailure(
+      DEBUG_LOCATION, StatusCode::UNAVAILABLE,
+      absl::StrCat("empty address list \\(LDS resource ", kServerName,
+                   ": does not exist \\(node ID:xds_end2end_test\\)\\)"));
   // Make sure we ACK'ed the update.
   auto response_state = balancer_->ads_service()->lds_response_state();
   ASSERT_TRUE(response_state.has_value());