[xDS] split LRS client into its own API (grpc#37605)

Closes grpc#37605 COPYBARA_INTEGRATE_REVIEW=grpc#37605 from markdroth:xds_client_lrs_refactor f9b86b8 PiperOrigin-RevId: 678281548
erm-g · Sep 24, 2024 · 06d3e40 · 06d3e40
1 parent 94b7927
commit 06d3e40
Show file tree

Hide file tree

Showing 41 changed files with 2,114 additions and 1,573 deletions.
diff --git a/BUILD b/BUILD
@@ -4395,17 +4395,18 @@ grpc_cc_library(
 grpc_cc_library(
     name = "xds_client",
     srcs = [
+        "//src/core:xds/xds_client/lrs_client.cc",
         "//src/core:xds/xds_client/xds_api.cc",
         "//src/core:xds/xds_client/xds_bootstrap.cc",
         "//src/core:xds/xds_client/xds_client.cc",
-        "//src/core:xds/xds_client/xds_client_stats.cc",
     ],
     hdrs = [
+        "//src/core:xds/xds_client/lrs_client.h",
         "//src/core:xds/xds_client/xds_api.h",
         "//src/core:xds/xds_client/xds_bootstrap.h",
         "//src/core:xds/xds_client/xds_channel_args.h",
         "//src/core:xds/xds_client/xds_client.h",
-        "//src/core:xds/xds_client/xds_client_stats.h",
+        "//src/core:xds/xds_client/xds_locality.h",
         "//src/core:xds/xds_client/xds_metrics.h",
         "//src/core:xds/xds_client/xds_resource_type.h",
         "//src/core:xds/xds_client/xds_resource_type_impl.h",
@@ -4464,6 +4465,7 @@ grpc_cc_library(
         "//src/core:json",
         "//src/core:per_cpu",
         "//src/core:ref_counted",
+        "//src/core:ref_counted_string",
         "//src/core:time",
         "//src/core:upb_utils",
         "//src/core:useful",

diff --git a/CMakeLists.txt b/CMakeLists.txt
diff --git a/Makefile b/Makefile
diff --git a/Package.swift b/Package.swift
diff --git a/build_autogenerated.yaml b/build_autogenerated.yaml
diff --git a/config.m4 b/config.m4
diff --git a/config.w32 b/config.w32
diff --git a/gRPC-C++.podspec b/gRPC-C++.podspec
diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec
diff --git a/grpc.gemspec b/grpc.gemspec
diff --git a/package.xml b/package.xml
diff --git a/src/core/BUILD b/src/core/BUILD
@@ -5696,6 +5696,7 @@ grpc_cc_library(
     external_deps = [
         "absl/base:core_headers",
         "absl/cleanup",
+        "absl/container:flat_hash_map",
         "absl/functional:bind_front",
         "absl/log:check",
         "absl/log:log",

diff --git a/src/core/load_balancing/xds/xds_cluster_impl.cc b/src/core/load_balancing/xds/xds_cluster_impl.cc
@@ -73,7 +73,7 @@
 #include "src/core/xds/grpc/xds_endpoint.h"
 #include "src/core/xds/xds_client/xds_bootstrap.h"
 #include "src/core/xds/xds_client/xds_client.h"
-#include "src/core/xds/xds_client/xds_client_stats.h"
+#include "src/core/xds/xds_client/xds_locality.h"
 
 namespace grpc_core {
 
@@ -189,13 +189,13 @@ class XdsClusterImplLb final : public LoadBalancingPolicy {
  private:
   class StatsSubchannelWrapper final : public DelegatingSubchannel {
    public:
-    // If load reporting is enabled and we have an XdsClusterLocalityStats
+    // If load reporting is enabled and we have a ClusterLocalityStats
     // object, that object already contains the locality label.  We
     // need to store the locality label directly only in the case where
     // load reporting is disabled.
     using LocalityData = absl::variant<
         RefCountedStringValue /*locality*/,
-        RefCountedPtr<XdsClusterLocalityStats> /*locality_stats*/>;
+        RefCountedPtr<LrsClient::ClusterLocalityStats> /*locality_stats*/>;
 
     StatsSubchannelWrapper(
         RefCountedPtr<SubchannelInterface> wrapped_subchannel,
@@ -209,20 +209,20 @@ class XdsClusterImplLb final : public LoadBalancingPolicy {
       return Match(
           locality_data_,
           [](RefCountedStringValue locality) { return locality; },
-          [](const RefCountedPtr<XdsClusterLocalityStats>& locality_stats) {
+          [](const RefCountedPtr<LrsClient::ClusterLocalityStats>&
+                 locality_stats) {
             return locality_stats->locality_name()->human_readable_string();
           });
     }
 
-    XdsClusterLocalityStats* locality_stats() const {
+    LrsClient::ClusterLocalityStats* locality_stats() const {
       return Match(
           locality_data_,
           [](const RefCountedStringValue&) {
-            return static_cast<XdsClusterLocalityStats*>(nullptr);
+            return static_cast<LrsClient::ClusterLocalityStats*>(nullptr);
           },
-          [](const RefCountedPtr<XdsClusterLocalityStats>& locality_stats) {
-            return locality_stats.get();
-          });
+          [](const RefCountedPtr<LrsClient::ClusterLocalityStats>&
+                 locality_stats) { return locality_stats.get(); });
     }
 
     const grpc_event_engine::experimental::Slice& hostname() const {
@@ -250,7 +250,7 @@ class XdsClusterImplLb final : public LoadBalancingPolicy {
     RefCountedStringValue service_telemetry_label_;
     RefCountedStringValue namespace_telemetry_label_;
     RefCountedPtr<XdsEndpointResource::DropConfig> drop_config_;
-    RefCountedPtr<XdsClusterDropStats> drop_stats_;
+    RefCountedPtr<LrsClient::ClusterDropStats> drop_stats_;
     RefCountedPtr<SubchannelPicker> picker_;
   };
 
@@ -304,7 +304,7 @@ class XdsClusterImplLb final : public LoadBalancingPolicy {
   RefCountedPtr<GrpcXdsClient> xds_client_;
 
   // The stats for client-side load reporting.
-  RefCountedPtr<XdsClusterDropStats> drop_stats_;
+  RefCountedPtr<LrsClient::ClusterDropStats> drop_stats_;
 
   OrphanablePtr<LoadBalancingPolicy> child_policy_;
 
@@ -324,7 +324,7 @@ class XdsClusterImplLb::Picker::SubchannelCallTracker final
   SubchannelCallTracker(
       std::unique_ptr<LoadBalancingPolicy::SubchannelCallTrackerInterface>
           original_subchannel_call_tracker,
-      RefCountedPtr<XdsClusterLocalityStats> locality_stats,
+      RefCountedPtr<LrsClient::ClusterLocalityStats> locality_stats,
       RefCountedPtr<CircuitBreakerCallCounterMap::CallCounter> call_counter)
       : original_subchannel_call_tracker_(
             std::move(original_subchannel_call_tracker)),
@@ -380,7 +380,7 @@ class XdsClusterImplLb::Picker::SubchannelCallTracker final
  private:
   std::unique_ptr<LoadBalancingPolicy::SubchannelCallTrackerInterface>
       original_subchannel_call_tracker_;
-  RefCountedPtr<XdsClusterLocalityStats> locality_stats_;
+  RefCountedPtr<LrsClient::ClusterLocalityStats> locality_stats_;
   RefCountedPtr<CircuitBreakerCallCounterMap::CallCounter> call_counter_;
 #ifndef NDEBUG
   bool started_ = false;
@@ -454,7 +454,7 @@ LoadBalancingPolicy::PickResult XdsClusterImplLb::Picker::Pick(
           subchannel_wrapper->locality());
     }
     // Handle load reporting.
-    RefCountedPtr<XdsClusterLocalityStats> locality_stats;
+    RefCountedPtr<LrsClient::ClusterLocalityStats> locality_stats;
     if (subchannel_wrapper->locality_stats() != nullptr) {
       locality_stats = subchannel_wrapper->locality_stats()->Ref(
           DEBUG_LOCATION, "SubchannelCallTracker");
@@ -618,14 +618,15 @@ absl::Status XdsClusterImplLb::UpdateLocked(UpdateArgs args) {
   // Note: We need a drop stats object whenever load reporting is enabled,
   // even if we have no EDS drop config, because we also use it when
   // reporting circuit breaker drops.
-  if (!new_cluster_config.cluster->lrs_load_reporting_server.has_value()) {
+  if (new_cluster_config.cluster->lrs_load_reporting_server == nullptr) {
     drop_stats_.reset();
   } else if (cluster_resource_ == nullptr ||
              old_eds_service_name != new_eds_service_name ||
-             cluster_resource_->lrs_load_reporting_server !=
-                 new_cluster_config.cluster->lrs_load_reporting_server) {
-    drop_stats_ = xds_client_->AddClusterDropStats(
-        *new_cluster_config.cluster->lrs_load_reporting_server,
+             !LrsServersEqual(
+                 cluster_resource_->lrs_load_reporting_server,
+                 new_cluster_config.cluster->lrs_load_reporting_server)) {
+    drop_stats_ = xds_client_->lrs_client().AddClusterDropStats(
+        new_cluster_config.cluster->lrs_load_reporting_server,
         new_config->cluster_name(), new_eds_service_name);
     if (drop_stats_ == nullptr) {
       LOG(ERROR)
@@ -819,12 +820,13 @@ RefCountedPtr<SubchannelInterface> XdsClusterImplLb::Helper::CreateSubchannel(
   // (if load reporting is enabled) the locality stats object, which
   // will be used by the picker.
   auto locality_name = per_address_args.GetObjectRef<XdsLocalityName>();
-  RefCountedPtr<XdsClusterLocalityStats> locality_stats;
-  if (parent()->cluster_resource_->lrs_load_reporting_server.has_value()) {
-    locality_stats = parent()->xds_client_->AddClusterLocalityStats(
-        parent()->cluster_resource_->lrs_load_reporting_server.value(),
-        parent()->config_->cluster_name(),
-        GetEdsResourceName(*parent()->cluster_resource_), locality_name);
+  RefCountedPtr<LrsClient::ClusterLocalityStats> locality_stats;
+  if (parent()->cluster_resource_->lrs_load_reporting_server != nullptr) {
+    locality_stats =
+        parent()->xds_client_->lrs_client().AddClusterLocalityStats(
+            parent()->cluster_resource_->lrs_load_reporting_server,
+            parent()->config_->cluster_name(),
+            GetEdsResourceName(*parent()->cluster_resource_), locality_name);
     if (locality_stats == nullptr) {
       LOG(ERROR)
           << "[xds_cluster_impl_lb " << parent()

diff --git a/src/core/load_balancing/xds/xds_wrr_locality.cc b/src/core/load_balancing/xds/xds_wrr_locality.cc
@@ -50,7 +50,7 @@
 #include "src/core/util/ref_counted_ptr.h"
 #include "src/core/util/ref_counted_string.h"
 #include "src/core/util/validation_errors.h"
-#include "src/core/xds/xds_client/xds_client_stats.h"
+#include "src/core/xds/xds_client/xds_locality.h"
 
 namespace grpc_core {