[XdsClient] replace e2e test with unit test (#34258)

This should address one of the failures we're seeing in #34224.

The test failure is caused by the changes in timing triggering a race
condition. In the code at head, we delay sending out the subscription
for the first CDS watch until we've already seen the other two CDS
watches, because the previous send_message op has not yet completed, and
by the time it does, we've seen all 3 watches, so we can send a
subscription for all 3 at the same time. With the WorkSerializer change,
the send_message op is complete by the time we see the first CDS watch,
so we subscribe to only that resource, and then later add the other two.
The result is that we'll NACK twice with two different messages, the
first one including only the error about the first resource, and the
second one including all three.

I suspect this same race condition would have been triggered eventually
by the EventEngine migration anyway; the current test basically depends
on the single-thread timing of the iomgr approach. So I'm addressing it
by replacing the e2e test with a unit test that covers the same cases
without the timing issue.

test/core/xds/xds_client_test.cc

@@ -549,6 +549,14 @@ class XdsClientTest : public ::testing::Test {
       return *this;
     }
 
+    ResponseBuilder& AddInvalidResourceWrapper() {
+      auto* res = response_.add_resources();
+      res->set_type_url(
+          "type.googleapis.com/envoy.service.discovery.v3.Resource");
+      res->set_value(std::string("\0", 1));
+      return *this;
+    }
+
     ResponseBuilder& AddEmptyResource() {
       response_.add_resources();
       return *this;
@@ -1235,6 +1243,9 @@ TEST_F(XdsClientTest, ResourceValidationFailureMultipleResources) {
           // Empty resource.  Will be included in NACK but will not
           // affect any watchers.
           .AddEmptyResource()
+          // Invalid resource wrapper.  Will be included in NACK but
+          // will not affect any watchers.
+          .AddInvalidResourceWrapper()
           // foo3: JSON parsing fails, but it is wrapped in a Resource
           // wrapper, so we do know the resource's name.
           .AddInvalidResource(XdsFooResourceType::Get()->type_url(),
@@ -1287,8 +1298,10 @@ TEST_F(XdsClientTest, ResourceValidationFailureMultipleResources) {
                    "(should be \"",
                    XdsFooResourceType::Get()->type_url(),
                    "\"); "
+                   // invalid resource wrapper
+                   "resource index 3: Can't decode Resource proto wrapper; "
                    // foo3
-                   "resource index 3: foo3: "
+                   "resource index 4: foo3: "
                    "INVALID_ARGUMENT: JSON parsing failed: "
                    "[JSON parse error at index 15]]")),
                /*resource_names=*/{"foo1", "foo2", "foo3", "foo4"});

test/cpp/end2end/xds/xds_core_end2end_test.cc

@@ -139,82 +139,6 @@ TEST_P(XdsClientTest, RestartsRequestsUponReconnection) {
   WaitForAllBackends(DEBUG_LOCATION, 1, 2);
 }
 
-// Tests that the NACK for multiple bad resources includes both errors.
-TEST_P(XdsClientTest, MultipleBadCdsResources) {
-  constexpr char kClusterName2[] = "cluster_name_2";
-  constexpr char kClusterName3[] = "cluster_name_3";
-  CreateAndStartBackends(1);
-  balancer_->ads_service()->set_inject_bad_resources_for_resource_type(
-      kCdsTypeUrl);
-  // Add cluster with unsupported type.
-  auto cluster = default_cluster_;
-  cluster.set_name(kClusterName2);
-  cluster.set_type(Cluster::STATIC);
-  balancer_->ads_service()->SetCdsResource(cluster);
-  // Add second cluster with the same error.
-  cluster.set_name(kClusterName3);
-  balancer_->ads_service()->SetCdsResource(cluster);
-  // Change RouteConfig to point to all clusters.
-  RouteConfiguration route_config = default_route_config_;
-  route_config.mutable_virtual_hosts(0)->clear_routes();
-  // First route: default cluster, selected based on header.
-  auto* route = route_config.mutable_virtual_hosts(0)->add_routes();
-  route->mutable_match()->set_prefix("");
-  auto* header_matcher = route->mutable_match()->add_headers();
-  header_matcher->set_name("cluster");
-  header_matcher->set_exact_match(kDefaultClusterName);
-  route->mutable_route()->set_cluster(kDefaultClusterName);
-  // Second route: cluster 2, selected based on header.
-  route = route_config.mutable_virtual_hosts(0)->add_routes();
-  route->mutable_match()->set_prefix("");
-  header_matcher = route->mutable_match()->add_headers();
-  header_matcher->set_name("cluster");
-  header_matcher->set_exact_match(kClusterName2);
-  route->mutable_route()->set_cluster(kClusterName2);
-  // Third route: cluster 3, used by default.
-  route = route_config.mutable_virtual_hosts(0)->add_routes();
-  route->mutable_match()->set_prefix("");
-  route->mutable_route()->set_cluster(kClusterName3);
-  SetRouteConfiguration(balancer_.get(), route_config);
-  // Add EDS resource.
-  EdsResourceArgs args({{"locality0", CreateEndpointsForBackends()}});
-  balancer_->ads_service()->SetEdsResource(BuildEdsResource(args));
-  // Send RPC.
-  const auto response_state = WaitForCdsNack(DEBUG_LOCATION);
-  ASSERT_TRUE(response_state.has_value()) << "timed out waiting for NACK";
-  EXPECT_EQ(
-      response_state->error_message,
-      absl::StrCat("xDS response validation errors: ["
-                   "resource index 0: Can't decode Resource proto wrapper; ",
-                   "resource index 1: foo: "
-                   "INVALID_ARGUMENT: Can't parse Cluster resource.; "
-                   "resource index 3: ",
-                   kClusterName2,
-                   ": INVALID_ARGUMENT: errors validating Cluster resource: ["
-                   "field:type error:unknown discovery type]; "
-                   "resource index 4: ",
-                   kClusterName3,
-                   ": INVALID_ARGUMENT: errors validating Cluster resource: ["
-                   "field:type error:unknown discovery type]]"));
-  // RPCs for default cluster should succeed.
-  std::vector<std::pair<std::string, std::string>> metadata_default_cluster = {
-      {"cluster", kDefaultClusterName},
-  };
-  CheckRpcSendOk(
-      DEBUG_LOCATION, 1,
-      RpcOptions().set_metadata(std::move(metadata_default_cluster)));
-  // RPCs for cluster 2 should fail.
-  std::vector<std::pair<std::string, std::string>> metadata_cluster_2 = {
-      {"cluster", kClusterName2},
-  };
-  CheckRpcSendFailure(
-      DEBUG_LOCATION, StatusCode::UNAVAILABLE,
-      "cluster_name_2: UNAVAILABLE: invalid resource: "
-      "INVALID_ARGUMENT: errors validating Cluster resource: \\["
-      "field:type error:unknown discovery type\\] .*",
-      RpcOptions().set_metadata(std::move(metadata_cluster_2)));
-}
-
 TEST_P(XdsClientTest, XdsStreamErrorPropagation) {
   const std::string kErrorMessage = "test forced ADS stream failure";
   balancer_->ads_service()->ForceADSFailure(

test/cpp/end2end/xds/xds_server.h

@@ -79,11 +79,6 @@ class AdsServiceImpl
     wrap_resources_ = wrap_resources;
   }
 
-  void set_inject_bad_resources_for_resource_type(const std::string& type_url) {
-    grpc_core::MutexLock lock(&ads_mu_);
-    inject_bad_resources_for_resource_type_ = type_url;
-  }
-
   // Sets a resource to a particular value, overwriting any previous value.
   void SetResource(google::protobuf::Any resource, const std::string& type_url,
                    const std::string& name);
@@ -392,22 +387,6 @@ class AdsServiceImpl
         resource_types_to_ignore_.end()) {
       return;
     }
-    // Inject bad resources if needed.
-    if (inject_bad_resources_for_resource_type_ == request.type_url()) {
-      response->emplace();
-      // Unparseable Resource wrapper.
-      auto* resource = (*response)->add_resources();
-      resource->set_type_url(
-          "type.googleapis.com/envoy.service.discovery.v3.Resource");
-      resource->set_value(std::string("\0", 1));
-      // Unparseable resource within Resource wrapper.
-      envoy::service::discovery::v3::Resource resource_wrapper;
-      resource_wrapper.set_name("foo");
-      resource = resource_wrapper.mutable_resource();
-      resource->set_type_url(request.type_url());
-      resource->set_value(std::string("\0", 1));
-      (*response)->add_resources()->PackFrom(resource_wrapper);
-    }
     // Look at all the resource names in the request.
     auto& subscription_name_map = (*subscription_map)[request.type_url()];
     auto& resource_type_state = resource_map_[request.type_url()];
@@ -595,7 +574,6 @@ class AdsServiceImpl
   ResourceMap resource_map_ ABSL_GUARDED_BY(ads_mu_);
   absl::optional<Status> forced_ads_failure_ ABSL_GUARDED_BY(ads_mu_);
   bool wrap_resources_ ABSL_GUARDED_BY(ads_mu_) = false;
-  std::string inject_bad_resources_for_resource_type_ ABSL_GUARDED_BY(ads_mu_);
 
   grpc_core::Mutex clients_mu_;
   std::set<std::string> clients_ ABSL_GUARDED_BY(clients_mu_);