[xDS] add per-authority knob to fallback based on reachability only (grpc#39889)

markdroth · paulosjca · commit 7b9539a9aa91 · 2025-08-22T17:46:32.000-07:00
First part of A95 (grpc/proposal#486). Closes grpc#39889 COPYBARA_INTEGRATE_REVIEW=grpc#39889 from markdroth:xds_fallback_criteria_knob 70702bf PiperOrigin-RevId: 777375509
diff --git a/src/core/xds/grpc/xds_bootstrap_grpc.cc b/src/core/xds/grpc/xds_bootstrap_grpc.cc
@@ -84,6 +84,8 @@ const JsonLoaderInterface* GrpcXdsBootstrap::GrpcAuthority::JsonLoader(
               "client_listener_resource_name_template",
               &GrpcAuthority::client_listener_resource_name_template_)
           .OptionalField("xds_servers", &GrpcAuthority::servers_)
+          .OptionalField("fallback_on_reachability_only",
+                         &GrpcAuthority::fallback_on_reachability_only_)
           .Finish();
   return loader;
 }
diff --git a/src/core/xds/grpc/xds_bootstrap_grpc.h b/src/core/xds/grpc/xds_bootstrap_grpc.h
@@ -85,6 +85,10 @@ class GrpcXdsBootstrap final : public XdsBootstrap {
       return servers;
     }
 
+    bool FallbackOnReachabilityOnly() const override {
+      return fallback_on_reachability_only_;
+    }
+
     const std::string& client_listener_resource_name_template() const {
       return client_listener_resource_name_template_;
     }
@@ -94,6 +98,7 @@ class GrpcXdsBootstrap final : public XdsBootstrap {
    private:
     std::vector<GrpcXdsServer> servers_;
     std::string client_listener_resource_name_template_;
+    bool fallback_on_reachability_only_;
   };
 
   // Creates bootstrap object from json_string.
diff --git a/src/core/xds/xds_client/xds_bootstrap.h b/src/core/xds/xds_client/xds_bootstrap.h
@@ -88,6 +88,8 @@ class XdsBootstrap {
     virtual ~Authority() = default;
 
     virtual std::vector<const XdsServer*> servers() const = 0;
+
+    virtual bool FallbackOnReachabilityOnly() const = 0;
   };
 
   virtual ~XdsBootstrap() = default;
diff --git a/src/core/xds/xds_client/xds_client.cc b/src/core/xds/xds_client/xds_client.cc
@@ -47,8 +47,10 @@
 #include "src/core/lib/iomgr/exec_ctx.h"
 #include "src/core/util/backoff.h"
 #include "src/core/util/debug_location.h"
+#include "src/core/util/env.h"
 #include "src/core/util/orphanable.h"
 #include "src/core/util/ref_counted_ptr.h"
+#include "src/core/util/string.h"
 #include "src/core/util/sync.h"
 #include "src/core/util/upb_utils.h"
 #include "src/core/util/uri.h"
@@ -68,6 +70,18 @@
 
 namespace grpc_core {
 
+namespace {
+
+bool XdsEndpointFallbackEnabled() {
+  auto value = GetEnv("GRPC_EXPERIMENTAL_XDS_ENDPOINT_FALLBACK");
+  if (!value.has_value()) return false;
+  bool parsed_value;
+  bool parse_succeeded = gpr_parse_bool_value(value->c_str(), &parsed_value);
+  return parse_succeeded && parsed_value;
+}
+
+}  // namespace
+
 using ::grpc_event_engine::experimental::EventEngine;
 
 //
@@ -489,13 +503,20 @@ void XdsClient::XdsChannel::UnsubscribeLocked(const XdsResourceType* type,
 
 bool XdsClient::XdsChannel::MaybeFallbackLocked(
     const std::string& authority, AuthorityState& authority_state) {
-  if (!xds_client_->HasUncachedResources(authority_state)) {
-    return false;
-  }
   std::vector<const XdsBootstrap::XdsServer*> xds_servers;
+  bool fallback_on_reachability_only = false;
   if (authority != kOldStyleAuthority) {
-    xds_servers =
-        xds_client_->bootstrap().LookupAuthority(authority)->servers();
+    auto* bootstrap_authority =
+        xds_client_->bootstrap().LookupAuthority(authority);
+    xds_servers = bootstrap_authority->servers();
+    if (XdsEndpointFallbackEnabled()) {
+      fallback_on_reachability_only =
+          bootstrap_authority->FallbackOnReachabilityOnly();
+    }
+  }
+  if (!fallback_on_reachability_only &&
+      !xds_client_->HasUncachedResources(authority_state)) {
+    return false;
   }
   if (xds_servers.empty()) xds_servers = xds_client_->bootstrap().servers();
   for (size_t i = authority_state.xds_channels.size(); i < xds_servers.size();
diff --git a/test/core/xds/xds_client_test.cc b/test/core/xds/xds_client_test.cc
@@ -201,8 +201,17 @@ class XdsClientTest : public ::testing::Test {
         server_ = std::move(server);
       }
 
+      bool FallbackOnReachabilityOnly() const override {
+        return fallback_on_reachability_only_;
+      }
+
+      void SetFallbackOnReachabilityOnly() {
+        fallback_on_reachability_only_ = true;
+      }
+
      private:
       std::optional<FakeXdsServer> server_;
+      bool fallback_on_reachability_only_ = false;
     };
 
     class Builder {
@@ -6263,6 +6272,254 @@ TEST_F(XdsClientTest, FallbackOnStartup) {
                /*resource_names=*/{"foo1"});
 }
 
+TEST_F(XdsClientTest, FallbackOnReachabilityOnly) {
+  testing::ScopedExperimentalEnvVar env_var(
+      "GRPC_EXPERIMENTAL_XDS_ENDPOINT_FALLBACK");
+  constexpr char kAuthority[] = "xds.example.com";
+  const std::string kXdstpResourceName = absl::StrCat(
+      "xdstp://", kAuthority, "/", XdsFooResource::TypeUrl(), "/foo1");
+  FakeXdsBootstrap::FakeAuthority authority;
+  authority.SetFallbackOnReachabilityOnly();
+  FakeXdsBootstrap::FakeXdsServer primary_server(kDefaultXdsServerUrl);
+  FakeXdsBootstrap::FakeXdsServer fallback_server("fallback_xds_server");
+  InitXdsClient(FakeXdsBootstrap::Builder()
+                    .AddAuthority(kAuthority, authority)
+                    .SetServers({primary_server, fallback_server}));
+  // Start a watch.
+  auto watcher = StartFooWatch(kXdstpResourceName);
+  EXPECT_THAT(GetServerConnections(), ::testing::ElementsAre(::testing::Pair(
+                                          kDefaultXdsServerUrl, true)));
+  EXPECT_THAT(
+      GetResourceCounts(),
+      ::testing::ElementsAre(::testing::Pair(
+          ResourceCountLabelsEq(
+              kAuthority, XdsFooResourceType::Get()->type_url(), "requested"),
+          1)));
+  EXPECT_TRUE(metrics_reporter_->WaitForMetricsReporterData(
+      ::testing::IsEmpty(), ::testing::_, ::testing::ElementsAre()));
+  // CSDS should show that the resource has been requested.
+  ClientConfig csds = DumpCsds();
+  EXPECT_THAT(csds.generic_xds_configs(),
+              ::testing::ElementsAre(CsdsResourceRequested(
+                  XdsFooResourceType::Get()->type_url(), kXdstpResourceName)));
+  // XdsClient should have created an ADS stream to the primary server.
+  auto stream = WaitForAdsStream();
+  ASSERT_TRUE(stream != nullptr);
+  // XdsClient should have sent a subscription request on the ADS stream.
+  auto request = WaitForRequest(stream.get());
+  ASSERT_TRUE(request.has_value());
+  CheckRequest(*request, XdsFooResourceType::Get()->type_url(),
+               /*version_info=*/"", /*response_nonce=*/"",
+               /*error_detail=*/absl::OkStatus(),
+               /*resource_names=*/{kXdstpResourceName});
+  // Primary server sends initial response.
+  stream->SendMessageToClient(
+      ResponseBuilder(XdsFooResourceType::Get()->type_url())
+          .set_version_info("20")
+          .set_nonce("O")
+          .AddFooResource(XdsFooResource(kXdstpResourceName, 6))
+          .Serialize());
+  // Resource should be delivered to watcher.
+  auto resource = watcher->WaitForNextResource();
+  ASSERT_NE(resource, nullptr);
+  EXPECT_EQ(resource->name, kXdstpResourceName);
+  EXPECT_EQ(resource->value, 6);
+  // Metrics should show 1 resource update and 1 cached resource.
+  EXPECT_TRUE(metrics_reporter_->WaitForMetricsReporterData(
+      ::testing::ElementsAre(::testing::Pair(
+          ::testing::Pair(kDefaultXdsServerUrl,
+                          XdsFooResourceType::Get()->type_url()),
+          1)),
+      ::testing::_, ::testing::_));
+  EXPECT_THAT(
+      GetResourceCounts(),
+      ::testing::ElementsAre(::testing::Pair(
+          ResourceCountLabelsEq(kAuthority,
+                                XdsFooResourceType::Get()->type_url(), "acked"),
+          1)));
+  // Check CSDS data.
+  csds = DumpCsds();
+  EXPECT_THAT(csds.generic_xds_configs(),
+              ::testing::UnorderedElementsAre(CsdsResourceAcked(
+                  XdsFooResourceType::Get()->type_url(), kXdstpResourceName,
+                  resource->AsJsonString(), "20", TimestampProtoEq(kTime0))));
+  // Client should send ACK to server.
+  request = WaitForRequest(stream.get());
+  ASSERT_TRUE(request.has_value());
+  CheckRequest(*request, XdsFooResourceType::Get()->type_url(),
+               /*version_info=*/"20", /*response_nonce=*/"O",
+               /*error_detail=*/absl::OkStatus(),
+               /*resource_names=*/{kXdstpResourceName});
+  // Trigger connection failure to primary.
+  TriggerConnectionFailure(primary_server,
+                           absl::UnavailableError("Server down"));
+  // This should trigger fallback.
+  auto fallback_stream = WaitForAdsStream(fallback_server);
+  ASSERT_NE(fallback_stream, nullptr);
+  request = WaitForRequest(fallback_stream.get());
+  ASSERT_TRUE(request.has_value());
+  CheckRequest(*request, XdsFooResourceType::Get()->type_url(),
+               /*version_info=*/"", /*response_nonce=*/"",
+               /*error_detail=*/absl::OkStatus(),
+               /*resource_names=*/{kXdstpResourceName});
+  // Metrics should show primary channel failing and fallback channel working.
+  EXPECT_THAT(
+      GetServerConnections(),
+      ::testing::ElementsAre(
+          ::testing::Pair(kDefaultXdsServerUrl, false),
+          ::testing::Pair(fallback_server.target()->server_uri(), true)));
+  EXPECT_TRUE(metrics_reporter_->WaitForMetricsReporterData(
+      ::testing::_, ::testing::_,
+      ::testing::ElementsAre(::testing::Pair(kDefaultXdsServerUrl, 1))));
+  // Fallback server sends a response.
+  fallback_stream->SendMessageToClient(
+      ResponseBuilder(XdsFooResourceType::Get()->type_url())
+          .set_version_info("5")
+          .set_nonce("A")
+          .AddFooResource(XdsFooResource(kXdstpResourceName, 30))
+          .Serialize());
+  // Resource is delivered to watcher.
+  resource = watcher->WaitForNextResource();
+  ASSERT_NE(resource, nullptr);
+  EXPECT_EQ(resource->name, kXdstpResourceName);
+  EXPECT_EQ(resource->value, 30);
+  // Metrics show update.
+  EXPECT_TRUE(metrics_reporter_->WaitForMetricsReporterData(
+      ::testing::ElementsAre(
+          ::testing::Pair(
+              ::testing::Pair(kDefaultXdsServerUrl,
+                              XdsFooResourceType::Get()->type_url()),
+              1),
+          ::testing::Pair(
+              ::testing::Pair(fallback_server.target()->server_uri(),
+                              XdsFooResourceType::Get()->type_url()),
+              1)),
+      ::testing::_, ::testing::_));
+  EXPECT_THAT(
+      GetResourceCounts(),
+      ::testing::ElementsAre(::testing::Pair(
+          ResourceCountLabelsEq(kAuthority,
+                                XdsFooResourceType::Get()->type_url(), "acked"),
+          1)));
+  // Check CSDS data.
+  csds = DumpCsds();
+  EXPECT_THAT(csds.generic_xds_configs(),
+              ::testing::UnorderedElementsAre(CsdsResourceAcked(
+                  XdsFooResourceType::Get()->type_url(), kXdstpResourceName,
+                  resource->AsJsonString(), "5", TimestampProtoEq(kTime0))));
+  // Client should send ACK to fallback server.
+  request = WaitForRequest(fallback_stream.get());
+  ASSERT_TRUE(request.has_value());
+  CheckRequest(*request, XdsFooResourceType::Get()->type_url(),
+               /*version_info=*/"5", /*response_nonce=*/"A",
+               /*error_detail=*/absl::OkStatus(),
+               /*resource_names=*/{kXdstpResourceName});
+  // Clean up.
+  CancelFooWatch(watcher.get(), kXdstpResourceName);
+  EXPECT_TRUE(stream->IsOrphaned());
+  EXPECT_TRUE(fallback_stream->IsOrphaned());
+}
+
+TEST_F(XdsClientTest, FallbackOnReachabilityOnlyNotEnabled) {
+  constexpr char kAuthority[] = "xds.example.com";
+  const std::string kXdstpResourceName = absl::StrCat(
+      "xdstp://", kAuthority, "/", XdsFooResource::TypeUrl(), "/foo1");
+  FakeXdsBootstrap::FakeAuthority authority;
+  authority.SetFallbackOnReachabilityOnly();
+  FakeXdsBootstrap::FakeXdsServer primary_server(kDefaultXdsServerUrl);
+  FakeXdsBootstrap::FakeXdsServer fallback_server("fallback_xds_server");
+  InitXdsClient(FakeXdsBootstrap::Builder()
+                    .AddAuthority(kAuthority, authority)
+                    .SetServers({primary_server, fallback_server}));
+  // Start a watch.
+  auto watcher = StartFooWatch(kXdstpResourceName);
+  EXPECT_THAT(GetServerConnections(), ::testing::ElementsAre(::testing::Pair(
+                                          kDefaultXdsServerUrl, true)));
+  EXPECT_THAT(
+      GetResourceCounts(),
+      ::testing::ElementsAre(::testing::Pair(
+          ResourceCountLabelsEq(
+              kAuthority, XdsFooResourceType::Get()->type_url(), "requested"),
+          1)));
+  EXPECT_TRUE(metrics_reporter_->WaitForMetricsReporterData(
+      ::testing::IsEmpty(), ::testing::_, ::testing::ElementsAre()));
+  // CSDS should show that the resource has been requested.
+  ClientConfig csds = DumpCsds();
+  EXPECT_THAT(csds.generic_xds_configs(),
+              ::testing::ElementsAre(CsdsResourceRequested(
+                  XdsFooResourceType::Get()->type_url(), kXdstpResourceName)));
+  // XdsClient should have created an ADS stream.
+  auto stream = WaitForAdsStream();
+  ASSERT_TRUE(stream != nullptr);
+  // XdsClient should have sent a subscription request on the ADS stream.
+  auto request = WaitForRequest(stream.get());
+  ASSERT_TRUE(request.has_value());
+  CheckRequest(*request, XdsFooResourceType::Get()->type_url(),
+               /*version_info=*/"", /*response_nonce=*/"",
+               /*error_detail=*/absl::OkStatus(),
+               /*resource_names=*/{kXdstpResourceName});
+  // Server sends resource.
+  stream->SendMessageToClient(
+      ResponseBuilder(XdsFooResourceType::Get()->type_url())
+          .set_version_info("20")
+          .set_nonce("O")
+          .AddFooResource(XdsFooResource(kXdstpResourceName, 6))
+          .Serialize());
+  // Resource should be delivered to watcher.
+  auto resource = watcher->WaitForNextResource();
+  ASSERT_NE(resource, nullptr);
+  EXPECT_EQ(resource->name, kXdstpResourceName);
+  EXPECT_EQ(resource->value, 6);
+  // Metrics should show 1 resource update and 1 cached resource.
+  EXPECT_TRUE(metrics_reporter_->WaitForMetricsReporterData(
+      ::testing::ElementsAre(::testing::Pair(
+          ::testing::Pair(kDefaultXdsServerUrl,
+                          XdsFooResourceType::Get()->type_url()),
+          1)),
+      ::testing::_, ::testing::_));
+  EXPECT_THAT(
+      GetResourceCounts(),
+      ::testing::ElementsAre(::testing::Pair(
+          ResourceCountLabelsEq(kAuthority,
+                                XdsFooResourceType::Get()->type_url(), "acked"),
+          1)));
+  // Check CSDS data.
+  csds = DumpCsds();
+  EXPECT_THAT(csds.generic_xds_configs(),
+              ::testing::UnorderedElementsAre(CsdsResourceAcked(
+                  XdsFooResourceType::Get()->type_url(), kXdstpResourceName,
+                  resource->AsJsonString(), "20", TimestampProtoEq(kTime0))));
+  // Client should send ACK to server.
+  request = WaitForRequest(stream.get());
+  ASSERT_TRUE(request.has_value());
+  CheckRequest(*request, XdsFooResourceType::Get()->type_url(),
+               /*version_info=*/"20", /*response_nonce=*/"O",
+               /*error_detail=*/absl::OkStatus(),
+               /*resource_names=*/{kXdstpResourceName});
+  // Trigger connection failure to primary.
+  TriggerConnectionFailure(primary_server,
+                           absl::UnavailableError("Server down"));
+  // Error should be reported to the watcher.
+  auto error = watcher->WaitForNextAmbientError();
+  ASSERT_TRUE(error.has_value());
+  EXPECT_EQ(error->code(), absl::StatusCode::kUnavailable);
+  EXPECT_EQ(error->message(),
+            "xDS channel for server default_xds_server: Server down (node "
+            "ID:xds_client_test)");
+  // This should NOT trigger fallback.
+  auto fallback_stream = WaitForAdsStream(fallback_server);
+  ASSERT_EQ(fallback_stream, nullptr);
+  // Metrics should show primary channel failing, but no fallback channel.
+  EXPECT_THAT(GetServerConnections(), ::testing::ElementsAre(::testing::Pair(
+                                          kDefaultXdsServerUrl, false)));
+  EXPECT_TRUE(metrics_reporter_->WaitForMetricsReporterData(
+      ::testing::_, ::testing::_,
+      ::testing::ElementsAre(::testing::Pair(kDefaultXdsServerUrl, 1))));
+  // Clean up.
+  CancelFooWatch(watcher.get(), kXdstpResourceName);
+  EXPECT_TRUE(stream->IsOrphaned());
+}
+
 }  // namespace
 }  // namespace testing
 }  // namespace grpc_core

Original file line number	Diff line number	Diff line change
`@@ -84,6 +84,8 @@ const JsonLoaderInterface* GrpcXdsBootstrap::GrpcAuthority::JsonLoader(`
`84`	`84`	`"client_listener_resource_name_template",`
`85`	`85`	`&GrpcAuthority::client_listener_resource_name_template_)`
`86`	`86`	`.OptionalField("xds_servers", &GrpcAuthority::servers_)`
	`87`	`+ .OptionalField("fallback_on_reachability_only",`
	`88`	`+ &GrpcAuthority::fallback_on_reachability_only_)`
`87`	`89`	`.Finish();`
`88`	`90`	`return loader;`
`89`	`91`	`}`