xDS changes for server listener resource name template (#24965 An error occurred )

* xDS: Server listener resource name template changes

* Reviewer comments

src/core/ext/xds/xds_api.cc

@@ -524,8 +524,9 @@ std::string XdsApi::LdsUpdate::FilterChain::ToString() const {
 //
 
 std::string XdsApi::LdsUpdate::ToString() const {
-  absl::InlinedVector<std::string, 3> contents;
+  absl::InlinedVector<std::string, 4> contents;
   if (type == ListenerType::kTcpListener) {
+    contents.push_back(absl::StrCat("address=", address));
     std::vector<std::string> filter_chains_content;
     for (const auto& filter_chain : filter_chains) {
       filter_chains_content.push_back(filter_chain.ToString());
@@ -1912,12 +1913,39 @@ XdsApi::LdsUpdate::FilterChain FilterChainParse(
   return filter_chain;
 }
 
+grpc_error* AddressParse(const envoy_config_core_v3_Address* address_proto,
+                         std::string* address) {
+  const auto* socket_address =
+      envoy_config_core_v3_Address_socket_address(address_proto);
+  if (socket_address == nullptr) {
+    return GRPC_ERROR_CREATE_FROM_COPIED_STRING(
+        "Address does not have socket_address");
+  }
+  if (envoy_config_core_v3_SocketAddress_protocol(socket_address) !=
+      envoy_config_core_v3_SocketAddress_TCP) {
+    return GRPC_ERROR_CREATE_FROM_STATIC_STRING(
+        "SocketAddress protocol is not TCP");
+  }
+  uint32_t port = envoy_config_core_v3_SocketAddress_port_value(socket_address);
+  if (port > 65535) {
+    return GRPC_ERROR_CREATE_FROM_STATIC_STRING("Invalid port");
+  }
+  *address = JoinHostPort(
+      UpbStringToAbsl(
+          envoy_config_core_v3_SocketAddress_address(socket_address)),
+      port);
+  return GRPC_ERROR_NONE;
+}
+
 grpc_error* LdsResponseParseServer(
     const EncodingContext& context,
     const envoy_config_listener_v3_Listener* listener,
     XdsApi::LdsUpdate* lds_update) {
   lds_update->type = XdsApi::LdsUpdate::ListenerType::kTcpListener;
-  grpc_error* error = GRPC_ERROR_NONE;
+  grpc_error* error =
+      AddressParse(envoy_config_listener_v3_Listener_address(listener),
+                   &lds_update->address);
+  if (error != GRPC_ERROR_NONE) return error;
   // TODO(yashykt): As part of this, we'll need to refactor the code to process
   // the HttpConnectionManager config so that it is shared with the client-side
   // parsing.

src/core/ext/xds/xds_api.h

@@ -295,6 +295,8 @@ class XdsApi {
       std::string ToString() const;
     };
 
+    // host:port listening_address set when type is kTcpListener
+    std::string address;
     std::vector<FilterChain> filter_chains;
     absl::optional<FilterChain> default_filter_chain;
 
@@ -302,7 +304,7 @@ class XdsApi {
       return route_config_name == other.route_config_name &&
              rds_update == other.rds_update &&
              http_max_stream_duration == other.http_max_stream_duration &&
-             http_filters == other.http_filters &&
+             http_filters == other.http_filters && address == other.address &&
              filter_chains == other.filter_chains &&
              default_filter_chain == other.default_filter_chain;
     }

src/core/ext/xds/xds_bootstrap.cc

@@ -119,6 +119,11 @@ std::string BootstrapString(const XdsBootstrap& bootstrap) {
         absl::StrJoin(bootstrap.server().server_features, ", "), "],\n"));
   }
   parts.push_back("  }\n],\n");
+  if (!bootstrap.server_listener_resource_name_template().empty()) {
+    parts.push_back(
+        absl::StrFormat("server_listener_resource_name_template=\"%s\",\n",
+                        bootstrap.server_listener_resource_name_template()));
+  }
   parts.push_back("certificate_providers={\n");
   for (const auto& entry : bootstrap.certificate_providers()) {
     parts.push_back(
@@ -233,6 +238,16 @@ XdsBootstrap::XdsBootstrap(Json json, grpc_error** error) {
       if (parse_error != GRPC_ERROR_NONE) error_list.push_back(parse_error);
     }
   }
+  it = json.mutable_object()->find("server_listener_resource_name_template");
+  if (it != json.mutable_object()->end()) {
+    if (it->second.type() != Json::Type::STRING) {
+      error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
+          "\"server_listener_resource_name_template\" field is not a string"));
+    } else {
+      server_listener_resource_name_template_ =
+          std::move(*it->second.mutable_string_value());
+    }
+  }
   if (XdsSecurityEnabled()) {
     it = json.mutable_object()->find("certificate_providers");
     if (it != json.mutable_object()->end()) {

src/core/ext/xds/xds_bootstrap.h

@@ -88,6 +88,9 @@ class XdsBootstrap {
   // add support for fallback for the xds channel.
   const XdsServer& server() const { return servers_[0]; }
   const Node* node() const { return node_.get(); }
+  const std::string& server_listener_resource_name_template() const {
+    return server_listener_resource_name_template_;
+  }
 
   const CertificateProviderStore::PluginDefinitionMap& certificate_providers()
       const {
@@ -108,6 +111,7 @@ class XdsBootstrap {
 
   absl::InlinedVector<XdsServer, 1> servers_;
   std::unique_ptr<Node> node_;
+  std::string server_listener_resource_name_template_;
   CertificateProviderStore::PluginDefinitionMap certificate_providers_;
 };
 

src/core/ext/xds/xds_client.h

@@ -88,6 +88,8 @@ class XdsClient : public DualRefCounted<XdsClient> {
   explicit XdsClient(grpc_error** error);
   ~XdsClient() override;
 
+  const XdsBootstrap* bootstrap() const { return bootstrap_.get(); }
+
   CertificateProviderStore& certificate_provider_store() {
     return *certificate_provider_store_;
   }

src/core/ext/xds/xds_server_config_fetcher.cc

@@ -18,6 +18,8 @@
 
 #include <grpc/support/port_platform.h>
 
+#include "absl/strings/str_replace.h"
+
 #include "src/core/ext/xds/xds_certificate_provider.h"
 #include "src/core/ext/xds/xds_client.h"
 #include "src/core/lib/channel/channel_args.h"
@@ -48,9 +50,9 @@ class XdsServerConfigFetcher : public grpc_server_config_fetcher {
         std::move(watcher), args, xds_client_, serving_status_notifier_,
         listening_address);
     auto* listener_watcher_ptr = listener_watcher.get();
-    // TODO(yashykt): Get the resource name id from bootstrap
-    listening_address = absl::StrCat(
-        "grpc/server?xds.resource.listening_address=", listening_address);
+    listening_address = absl::StrReplaceAll(
+        xds_client_->bootstrap()->server_listener_resource_name_template(),
+        {{"%s", listening_address}});
     xds_client_->WatchListenerData(listening_address,
                                    std::move(listener_watcher));
     MutexLock lock(&mu_);
@@ -106,6 +108,11 @@ class XdsServerConfigFetcher : public grpc_server_config_fetcher {
             "[ListenerWatcher %p] Received LDS update from xds client %p: %s",
             this, xds_client_.get(), listener.ToString().c_str());
       }
+      if (listener.address != listening_address_) {
+        OnFatalError(absl::FailedPreconditionError(
+            "Address in LDS update does not match listening address"));
+        return;
+      }
       grpc_error* error = GRPC_ERROR_NONE;
       bool update_needed = UpdateXdsCertificateProvider(listener, &error);
       if (error != GRPC_ERROR_NONE) {
@@ -160,11 +167,11 @@ class XdsServerConfigFetcher : public grpc_server_config_fetcher {
       GRPC_ERROR_UNREF(error);
     }
 
-    void OnResourceDoesNotExist() override {
-      gpr_log(GPR_ERROR,
-              "ListenerWatcher:%p XdsClient reports requested listener does "
-              "not exist; not serving on %s",
-              this, listening_address_.c_str());
+    void OnFatalError(absl::Status status) {
+      gpr_log(
+          GPR_ERROR,
+          "ListenerWatcher:%p Encountered fatal error %s; not serving on %s",
+          this, status.ToString().c_str(), listening_address_.c_str());
       if (have_resource_) {
         // The server has started listening already, so we need to gracefully
         // stop serving.
@@ -174,10 +181,15 @@ class XdsServerConfigFetcher : public grpc_server_config_fetcher {
       if (serving_status_notifier_.on_serving_status_change != nullptr) {
         serving_status_notifier_.on_serving_status_change(
             serving_status_notifier_.user_data, listening_address_.c_str(),
-            GRPC_STATUS_NOT_FOUND, "Requested listener does not exist");
+            static_cast<grpc_status_code>(status.raw_code()),
+            std::string(status.message()).c_str());
       }
     }
 
+    void OnResourceDoesNotExist() override {
+      OnFatalError(absl::NotFoundError("Requested listener does not exist"));
+    }
+
    private:
     // Returns true if the xds certificate provider changed in a way that
     // required a new security connector to be created, false otherwise.
@@ -315,5 +327,13 @@ grpc_server_config_fetcher* grpc_server_config_fetcher_xds_create(
     GRPC_ERROR_UNREF(error);
     return nullptr;
   }
+  if (xds_client->bootstrap()
+          ->server_listener_resource_name_template()
+          .empty()) {
+    gpr_log(GPR_ERROR,
+            "server_listener_resource_name_template not provided in bootstrap "
+            "file.");
+    return nullptr;
+  }
   return new grpc_core::XdsServerConfigFetcher(std::move(xds_client), notifier);
 }

test/core/xds/xds_bootstrap_test.cc

@@ -110,6 +110,7 @@ TEST_P(XdsBootstrapTest, Basic) {
       "    },"
       "    \"ignore\": \"whee\""
       "  },"
+      "  \"server_listener_resource_name_template\": \"example/resource\","
       "  \"ignore\": {}"
       "}";
   grpc_error* error = GRPC_ERROR_NONE;
@@ -140,6 +141,8 @@ TEST_P(XdsBootstrapTest, Basic) {
                       ::testing::AllOf(
                           ::testing::Property(&Json::type, Json::Type::NUMBER),
                           ::testing::Property(&Json::string_value, "1")))));
+  EXPECT_EQ(bootstrap.server_listener_resource_name_template(),
+            "example/resource");
 }
 
 TEST_P(XdsBootstrapTest, ValidWithoutNode) {
@@ -271,6 +274,7 @@ TEST_P(XdsBootstrapTest, TopFieldsWrongTypes) {
       "{"
       "  \"xds_servers\":1,"
       "  \"node\":1,"
+      "  \"server_listener_resource_name_template\":1,"
       "  \"certificate_providers\":1"
       "}";
   grpc_error* error = GRPC_ERROR_NONE;
@@ -279,7 +283,9 @@ TEST_P(XdsBootstrapTest, TopFieldsWrongTypes) {
   XdsBootstrap bootstrap(std::move(json), &error);
   EXPECT_THAT(grpc_error_string(error),
               ::testing::ContainsRegex("\"xds_servers\" field is not an array.*"
-                                       "\"node\" field is not an object.*"));
+                                       "\"node\" field is not an object.*"
+                                       "\"server_listener_resource_name_"
+                                       "template\" field is not a string.*"));
   if (GetParam().parse_xds_certificate_providers()) {
     EXPECT_THAT(grpc_error_string(error),
                 ::testing::ContainsRegex(

test/cpp/end2end/xds_end2end_test.cc

@@ -200,6 +200,8 @@ constexpr char kBootstrapFileV3[] =
     "      \"sub_zone\": \"mp3\"\n"
     "    }\n"
     "  },\n"
+    "  \"server_listener_resource_name_template\": "
+    "\"grpc/server?xds.resource.listening_address=%s\",\n"
     "  \"certificate_providers\": {\n"
     "    \"fake_plugin1\": {\n"
     "      \"plugin_name\": \"fake1\"\n"
@@ -7221,10 +7223,6 @@ TEST_P(XdsEnabledServerTest, Basic) {
       backends_[0]->port());
   listener.add_filter_chains();
   balancers_[0]->ads_service()->SetLdsResource(listener);
-  listener.set_name(
-      absl::StrCat("grpc/server?xds.resource.listening_address=[::1]:",
-                   backends_[0]->port()));
-  balancers_[0]->ads_service()->SetLdsResource(listener);
   WaitForBackend(0);
   CheckRpcSendOk();
 }
@@ -7269,6 +7267,34 @@ TEST_P(XdsEnabledServerTest, BadLdsUpdateBothApiListenerAndAddress) {
       ::testing::HasSubstr("Listener has both address and ApiListener"));
 }
 
+// Verify that a mismatch of listening address results in "not serving" status.
+TEST_P(XdsEnabledServerTest, ListenerAddressMismatch) {
+  Listener listener;
+  listener.set_name(
+      absl::StrCat("grpc/server?xds.resource.listening_address=",
+                   ipv6_only_ ? "[::1]:" : "127.0.0.1:", backends_[0]->port()));
+  listener.mutable_address()->mutable_socket_address()->set_address(
+      ipv6_only_ ? "::1" : "127.0.0.1");
+  listener.mutable_address()->mutable_socket_address()->set_port_value(
+      backends_[0]->port());
+  listener.add_filter_chains();
+  balancers_[0]->ads_service()->SetLdsResource(listener);
+  WaitForBackend(0);
+  CheckRpcSendOk();
+  // Set a different listening address in the LDS update
+  listener.mutable_address()->mutable_socket_address()->set_address(
+      "192.168.1.1");
+  balancers_[0]->ads_service()->SetLdsResource(listener);
+  bool rpc_failed = false;
+  for (int i = 0; i < 100; ++i) {
+    if (!SendRpc().ok()) {
+      rpc_failed = true;
+      break;
+    }
+  }
+  EXPECT_TRUE(rpc_failed);
+}
+
 class XdsServerSecurityTest : public XdsEnd2endTest {
  protected:
   XdsServerSecurityTest()