xds: add tests for unknown authority (#29681)

* xds: add tests for unknown authority

* improve trace log message

* fix memory leak

src/core/ext/filters/client_channel/lb_policy/xds/cds.cc

@@ -224,8 +224,7 @@ void CdsLb::Helper::UpdateState(grpc_connectivity_state state,
                                 std::unique_ptr<SubchannelPicker> picker) {
   if (parent_->shutting_down_ || parent_->child_policy_ == nullptr) return;
   if (GRPC_TRACE_FLAG_ENABLED(grpc_cds_lb_trace)) {
-    gpr_log(GPR_INFO,
-            "[cdslb %p] state updated by child: %s message_state: (%s)", this,
+    gpr_log(GPR_INFO, "[cdslb %p] state updated by child: %s (%s)", this,
             ConnectivityStateName(state), status.ToString().c_str());
   }
   parent_->channel_control_helper()->UpdateState(state, status,
@@ -541,8 +540,8 @@ void CdsLb::OnError(const std::string& name, absl::Status status) {
   if (child_policy_ == nullptr) {
     channel_control_helper()->UpdateState(
         GRPC_CHANNEL_TRANSIENT_FAILURE, status,
-        absl::make_unique<TransientFailurePicker>(
-            absl::UnavailableError(status.ToString())));
+        absl::make_unique<TransientFailurePicker>(absl::UnavailableError(
+            absl::StrCat(name, ": ", status.ToString()))));
   }
 }
 

src/core/ext/filters/client_channel/resolver/xds/xds_resolver.cc

@@ -171,7 +171,7 @@ class XdsResolver : public Resolver {
       Ref().release();  // ref held by lambda
       resolver_->work_serializer_->Run(
           [this, status]() {
-            resolver_->OnError(status);
+            resolver_->OnError(resolver_->lds_resource_name_, status);
             Unref();
           },
           DEBUG_LOCATION);
@@ -210,7 +210,7 @@ class XdsResolver : public Resolver {
       Ref().release();  // ref held by lambda
       resolver_->work_serializer_->Run(
           [this, status]() {
-            resolver_->OnError(status);
+            resolver_->OnError(resolver_->route_config_name_, status);
             Unref();
           },
           DEBUG_LOCATION);
@@ -349,7 +349,7 @@ class XdsResolver : public Resolver {
 
   void OnListenerUpdate(XdsListenerResource listener);
   void OnRouteConfigUpdate(XdsRouteConfigResource rds_update);
-  void OnError(absl::Status status);
+  void OnError(absl::string_view context, absl::Status status);
   void OnResourceDoesNotExist();
 
   absl::StatusOr<RefCountedPtr<ServiceConfig>> CreateServiceConfig();
@@ -804,7 +804,7 @@ void XdsResolver::StartLocked() {
     if (authority_config == nullptr) {
       Result result;
       result.service_config = absl::UnavailableError(
-          absl::StrCat("Invalid target URI -- authority not found for %s.",
+          absl::StrCat("Invalid target URI -- authority not found for ",
                        uri_.authority().c_str()));
       result.args = grpc_channel_args_copy(args_);
       result_handler_->ReportResult(std::move(result));
@@ -937,9 +937,11 @@ void XdsResolver::OnRouteConfigUpdate(XdsRouteConfigResource rds_update) {
       VirtualHostListIterator(&rds_update.virtual_hosts),
       data_plane_authority_);
   if (!vhost_index.has_value()) {
-    OnError(absl::UnavailableError(
-        absl::StrCat("could not find VirtualHost for ", data_plane_authority_,
-                     " in RouteConfiguration")));
+    OnError(
+        route_config_name_.empty() ? lds_resource_name_ : route_config_name_,
+        absl::UnavailableError(absl::StrCat("could not find VirtualHost for ",
+                                            data_plane_authority_,
+                                            " in RouteConfiguration")));
     return;
   }
   // Save the virtual host in the resolver.
@@ -950,15 +952,15 @@ void XdsResolver::OnRouteConfigUpdate(XdsRouteConfigResource rds_update) {
   GenerateResult();
 }
 
-void XdsResolver::OnError(absl::Status status) {
-  gpr_log(GPR_ERROR, "[xds_resolver %p] received error from XdsClient: %s",
-          this, status.ToString().c_str());
+void XdsResolver::OnError(absl::string_view context, absl::Status status) {
+  gpr_log(GPR_ERROR, "[xds_resolver %p] received error from XdsClient: %s: %s",
+          this, std::string(context).c_str(), status.ToString().c_str());
   if (xds_client_ == nullptr) return;
   Result result;
   grpc_arg new_arg = xds_client_->MakeChannelArg();
   result.args = grpc_channel_args_copy_and_add(args_, &new_arg, 1);
-  result.service_config = absl::UnavailableError(
-      absl::StrCat("error obtaining xDS resources: ", status.ToString()));
+  result.service_config =
+      absl::UnavailableError(absl::StrCat(context, ": ", status.ToString()));
   result_handler_->ReportResult(std::move(result));
 }
 
@@ -1034,7 +1036,8 @@ void XdsResolver::GenerateResult() {
   grpc_error_handle error = GRPC_ERROR_NONE;
   auto config_selector = MakeRefCounted<XdsConfigSelector>(Ref(), &error);
   if (error != GRPC_ERROR_NONE) {
-    OnError(absl::UnavailableError(grpc_error_std_string(error)));
+    OnError("could not create ConfigSelector",
+            absl::UnavailableError(grpc_error_std_string(error)));
     GRPC_ERROR_UNREF(error);
     return;
   }

src/core/ext/xds/xds_client.cc

@@ -1938,11 +1938,9 @@ void XdsClient::CancelResourceWatch(const XdsResourceType* type,
                                     bool delay_unsubscription) {
   auto resource_name = ParseXdsResourceName(name, type);
   MutexLock lock(&mu_);
-  if (!resource_name.ok()) {
-    invalid_watchers_.erase(watcher);
-    return;
-  }
-  if (shutting_down_) return;
+  // We cannot be sure whether the watcher is in invalid_watchers_ or in
+  // authority_state_map_, so we check both, just to be safe.
+  invalid_watchers_.erase(watcher);
   // Find authority.
   auto authority_it = authority_state_map_.find(resource_name->authority);
   if (authority_it == authority_state_map_.end()) return;

test/cpp/end2end/xds/xds_core_end2end_test.cc

@@ -596,11 +596,10 @@ class XdsFederationTest : public XdsEnd2endTest {
 };
 
 // Get bootstrap from env var, so that there's a global XdsClient.
-// Runs with and without RDS.
+// Runs with RDS so that we know all resource types work properly.
 INSTANTIATE_TEST_SUITE_P(
     XdsTest, XdsFederationTest,
     ::testing::Values(
-        XdsTestType().set_bootstrap_source(XdsTestType::kBootstrapFromEnvVar),
         XdsTestType()
             .set_bootstrap_source(XdsTestType::kBootstrapFromEnvVar)
             .set_enable_rds_testing()),
@@ -808,6 +807,185 @@ TEST_P(XdsFederationTest, FederationTargetAuthorityWithResourceTemplate) {
   EXPECT_EQ(1U, backends_[1]->backend_service()->request_count());
 }
 
+TEST_P(XdsFederationTest, TargetUriAuthorityUnknown) {
+  ScopedExperimentalEnvVar env_var("GRPC_EXPERIMENTAL_XDS_FEDERATION");
+  const char* kAuthority = "xds.example.com";
+  const char* kNewServerName = "whee%/server.example.com";
+  const char* kNewListenerTemplate =
+      "xdstp://xds.example.com/envoy.config.listener.v3.Listener/"
+      "client/%s?psm_project_id=1234";
+  BootstrapBuilder builder = BootstrapBuilder();
+  builder.AddAuthority(
+      kAuthority, absl::StrCat("localhost:", grpc_pick_unused_port_or_die()),
+      kNewListenerTemplate);
+  InitClient(builder);
+  auto channel2 = CreateChannel(
+      /*failover_timeout_ms=*/0, kNewServerName, "xds.unknown.com");
+  auto stub2 = grpc::testing::EchoTestService::NewStub(channel2);
+  ClientContext context;
+  EchoRequest request;
+  request.set_message(kRequestMessage);
+  EchoResponse response;
+  grpc::Status status = stub2->Echo(&context, request, &response);
+  EXPECT_EQ(status.error_code(), StatusCode::UNAVAILABLE);
+  EXPECT_EQ(status.error_message(),
+            "Invalid target URI -- authority not found for xds.unknown.com");
+  ASSERT_EQ(GRPC_CHANNEL_TRANSIENT_FAILURE, channel2->GetState(false));
+}
+
+TEST_P(XdsFederationTest, RdsResourceNameAuthorityUnknown) {
+  ScopedExperimentalEnvVar env_var("GRPC_EXPERIMENTAL_XDS_FEDERATION");
+  const char* kAuthority = "xds.example.com";
+  const char* kNewServerName = "whee%/server.example.com";
+  const char* kNewListenerTemplate =
+      "xdstp://xds.example.com/envoy.config.listener.v3.Listener/"
+      "client/%s?psm_project_id=1234";
+  const char* kNewListenerName =
+      "xdstp://xds.example.com/envoy.config.listener.v3.Listener/"
+      "client/whee%25/server.example.com?psm_project_id=1234";
+  const char* kNewRouteConfigName =
+      "xdstp://xds.unknown.com/envoy.config.route.v3.RouteConfiguration/"
+      "new_route_config_name";
+  BootstrapBuilder builder = BootstrapBuilder();
+  builder.AddAuthority(kAuthority,
+                       absl::StrCat("localhost:", authority_balancer_->port()),
+                       kNewListenerTemplate);
+  InitClient(builder);
+  // New RouteConfig
+  RouteConfiguration new_route_config = default_route_config_;
+  new_route_config.set_name(kNewRouteConfigName);
+  // New Listener
+  Listener listener = default_listener_;
+  listener.set_name(kNewListenerName);
+  SetListenerAndRouteConfiguration(authority_balancer_.get(), listener,
+                                   new_route_config);
+  // Channel should report TRANSIENT_FAILURE.
+  auto channel2 =
+      CreateChannel(/*failover_timeout_ms=*/0, kNewServerName, kAuthority);
+  auto stub2 = grpc::testing::EchoTestService::NewStub(channel2);
+  ClientContext context;
+  EchoRequest request;
+  request.set_message(kRequestMessage);
+  EchoResponse response;
+  grpc::Status status = stub2->Echo(&context, request, &response);
+  EXPECT_EQ(status.error_code(), StatusCode::UNAVAILABLE);
+  EXPECT_EQ(status.error_message(),
+            absl::StrCat(
+                kNewRouteConfigName,
+                ": UNAVAILABLE: authority \"xds.unknown.com\" not present in "
+                "bootstrap config"));
+  ASSERT_EQ(GRPC_CHANNEL_TRANSIENT_FAILURE, channel2->GetState(false));
+}
+
+TEST_P(XdsFederationTest, CdsResourceNameAuthorityUnknown) {
+  ScopedExperimentalEnvVar env_var("GRPC_EXPERIMENTAL_XDS_FEDERATION");
+  const char* kAuthority = "xds.example.com";
+  const char* kNewServerName = "whee%/server.example.com";
+  const char* kNewListenerTemplate =
+      "xdstp://xds.example.com/envoy.config.listener.v3.Listener/"
+      "client/%s?psm_project_id=1234";
+  const char* kNewListenerName =
+      "xdstp://xds.example.com/envoy.config.listener.v3.Listener/"
+      "client/whee%25/server.example.com?psm_project_id=1234";
+  const char* kNewRouteConfigName =
+      "xdstp://xds.example.com/envoy.config.route.v3.RouteConfiguration/"
+      "new_route_config_name";
+  const char* kNewClusterName =
+      "xdstp://xds.unknown.com/envoy.config.cluster.v3.Cluster/"
+      "cluster_name";
+  BootstrapBuilder builder = BootstrapBuilder();
+  builder.AddAuthority(kAuthority,
+                       absl::StrCat("localhost:", authority_balancer_->port()),
+                       kNewListenerTemplate);
+  InitClient(builder);
+  // New Route
+  RouteConfiguration new_route_config = default_route_config_;
+  new_route_config.set_name(kNewRouteConfigName);
+  new_route_config.mutable_virtual_hosts(0)
+      ->mutable_routes(0)
+      ->mutable_route()
+      ->set_cluster(kNewClusterName);
+  // New Listener
+  Listener listener = default_listener_;
+  listener.set_name(kNewListenerName);
+  SetListenerAndRouteConfiguration(authority_balancer_.get(), listener,
+                                   new_route_config);
+  // Channel should report TRANSIENT_FAILURE.
+  auto channel2 =
+      CreateChannel(/*failover_timeout_ms=*/0, kNewServerName, kAuthority);
+  auto stub2 = grpc::testing::EchoTestService::NewStub(channel2);
+  ClientContext context;
+  EchoRequest request;
+  request.set_message(kRequestMessage);
+  EchoResponse response;
+  grpc::Status status = stub2->Echo(&context, request, &response);
+  EXPECT_EQ(status.error_code(), StatusCode::UNAVAILABLE);
+  EXPECT_EQ(status.error_message(),
+            absl::StrCat(
+                kNewClusterName,
+                ": UNAVAILABLE: authority \"xds.unknown.com\" not present in "
+                "bootstrap config"));
+  ASSERT_EQ(GRPC_CHANNEL_TRANSIENT_FAILURE, channel2->GetState(false));
+}
+
+TEST_P(XdsFederationTest, EdsResourceNameAuthorityUnknown) {
+  ScopedExperimentalEnvVar env_var("GRPC_EXPERIMENTAL_XDS_FEDERATION");
+  const char* kAuthority = "xds.example.com";
+  const char* kNewServerName = "whee%/server.example.com";
+  const char* kNewListenerTemplate =
+      "xdstp://xds.example.com/envoy.config.listener.v3.Listener/"
+      "client/%s?psm_project_id=1234";
+  const char* kNewListenerName =
+      "xdstp://xds.example.com/envoy.config.listener.v3.Listener/"
+      "client/whee%25/server.example.com?psm_project_id=1234";
+  const char* kNewRouteConfigName =
+      "xdstp://xds.example.com/envoy.config.route.v3.RouteConfiguration/"
+      "new_route_config_name";
+  const char* kNewEdsServiceName =
+      "xdstp://xds.unknown.com/envoy.config.endpoint.v3.ClusterLoadAssignment/"
+      "edsservice_name";
+  const char* kNewClusterName =
+      "xdstp://xds.example.com/envoy.config.cluster.v3.Cluster/"
+      "cluster_name";
+  BootstrapBuilder builder = BootstrapBuilder();
+  builder.AddAuthority(kAuthority,
+                       absl::StrCat("localhost:", authority_balancer_->port()),
+                       kNewListenerTemplate);
+  InitClient(builder);
+  // New cluster
+  Cluster new_cluster = default_cluster_;
+  new_cluster.set_name(kNewClusterName);
+  new_cluster.mutable_eds_cluster_config()->set_service_name(
+      kNewEdsServiceName);
+  authority_balancer_->ads_service()->SetCdsResource(new_cluster);
+  // New Route
+  RouteConfiguration new_route_config = default_route_config_;
+  new_route_config.set_name(kNewRouteConfigName);
+  new_route_config.mutable_virtual_hosts(0)
+      ->mutable_routes(0)
+      ->mutable_route()
+      ->set_cluster(kNewClusterName);
+  // New Listener
+  Listener listener = default_listener_;
+  listener.set_name(kNewListenerName);
+  SetListenerAndRouteConfiguration(authority_balancer_.get(), listener,
+                                   new_route_config);
+  // Channel should report TRANSIENT_FAILURE.
+  auto channel2 =
+      CreateChannel(/*failover_timeout_ms=*/0, kNewServerName, kAuthority);
+  auto stub2 = grpc::testing::EchoTestService::NewStub(channel2);
+  ClientContext context;
+  EchoRequest request;
+  request.set_message(kRequestMessage);
+  EchoResponse response;
+  grpc::Status status = stub2->Echo(&context, request, &response);
+  EXPECT_EQ(status.error_code(), StatusCode::UNAVAILABLE);
+  // TODO(roth): Improve this error message as part of
+  // https://github.com/grpc/grpc/issues/22883.
+  EXPECT_EQ(status.error_message(), "no ready priority");
+  ASSERT_EQ(GRPC_CHANNEL_TRANSIENT_FAILURE, channel2->GetState(false));
+}
+
 // Setting server_listener_resource_name_template to start with "xdstp:" and
 // look up xds server under an authority map.
 TEST_P(XdsFederationTest, FederationServer) {