[xDS] require EDS resource name in CDS resources with xdstp names (#33425)

Implements the change described in
https://github.com/grpc/proposal/pull/377.

src/core/ext/xds/xds_cluster.cc

@@ -24,6 +24,7 @@
 
 #include "absl/status/status.h"
 #include "absl/status/statusor.h"
+#include "absl/strings/match.h"
 #include "absl/strings/str_cat.h"
 #include "absl/strings/str_join.h"
 #include "absl/strings/strip.h"
@@ -187,6 +188,8 @@ XdsClusterResource::Eds EdsConfigParse(
   if (eds_cluster_config == nullptr) {
     errors->AddError("field not present");
   } else {
+    // Validate ConfigSource.
+    {
       ValidationErrors::ScopedField field(errors, ".eds_config");
       const envoy_config_core_v3_ConfigSource* eds_config =
           envoy_config_cluster_v3_Cluster_EdsClusterConfig_eds_config(
@@ -198,12 +201,19 @@ XdsClusterResource::Eds EdsConfigParse(
             !envoy_config_core_v3_ConfigSource_has_self(eds_config)) {
           errors->AddError("ConfigSource is not ads or self");
         }
+      }
+    }
     // Record EDS service_name (if any).
-      upb_StringView service_name =
+    // This field is required if the CDS resource has an xdstp name.
+    eds.eds_service_name = UpbStringToStdString(
         envoy_config_cluster_v3_Cluster_EdsClusterConfig_service_name(
-              eds_cluster_config);
-      if (service_name.size != 0) {
-        eds.eds_service_name = UpbStringToStdString(service_name);
+            eds_cluster_config));
+    if (eds.eds_service_name.empty()) {
+      absl::string_view cluster_name =
+          UpbStringToAbsl(envoy_config_cluster_v3_Cluster_name(cluster));
+      if (absl::StartsWith(cluster_name, "xdstp:")) {
+        ValidationErrors::ScopedField field(errors, ".service_name");
+        errors->AddError("must be set if Cluster resource has an xdstp name");
       }
     }
   }

test/core/xds/xds_cluster_resource_type_test.cc

@@ -217,6 +217,28 @@ TEST_F(ClusterTypeTest, EdsServiceName) {
   EXPECT_EQ(eds->eds_service_name, "bar");
 }
 
+TEST_F(ClusterTypeTest, EdsServiceNameAbsentWithXdstpName) {
+  Cluster cluster;
+  cluster.set_name("xdstp:foo");
+  cluster.set_type(cluster.EDS);
+  auto* eds_cluster_config = cluster.mutable_eds_cluster_config();
+  eds_cluster_config->mutable_eds_config()->mutable_self();
+  std::string serialized_resource;
+  ASSERT_TRUE(cluster.SerializeToString(&serialized_resource));
+  auto* resource_type = XdsClusterResourceType::Get();
+  auto decode_result =
+      resource_type->Decode(decode_context_, serialized_resource);
+  ASSERT_TRUE(decode_result.name.has_value());
+  EXPECT_EQ(*decode_result.name, "xdstp:foo");
+  EXPECT_EQ(decode_result.resource.status().code(),
+            absl::StatusCode::kInvalidArgument);
+  EXPECT_EQ(decode_result.resource.status().message(),
+            "errors validating Cluster resource: ["
+            "field:eds_cluster_config.service_name "
+            "error:must be set if Cluster resource has an xdstp name]")
+      << decode_result.resource.status();
+}
+
 TEST_F(ClusterTypeTest, DiscoveryTypeNotPresent) {
   Cluster cluster;
   cluster.set_name("foo");