hds: add support for delta updates in specifier

docs/root/version_history/current.rst

@@ -27,6 +27,7 @@ Removed Config or Runtime
 New Features
 ------------
 * grpc: implemented header value syntax support when defining :ref:`initial metadata <envoy_v3_api_field_config.core.v3.GrpcService.initial_metadata>` for gRPC-based `ext_authz` :ref:`HTTP <envoy_v3_api_field_extensions.filters.http.ext_authz.v3.ExtAuthz.grpc_service>` and :ref:`network <envoy_v3_api_field_extensions.filters.network.ext_authz.v3.ExtAuthz.grpc_service>` filters, and :ref:`ratelimit <envoy_v3_api_field_config.ratelimit.v3.RateLimitServiceConfig.grpc_service>` filters.
+* hds: added support for delta updates in the :ref:`HealthCheckSpecifier <envoy_v3_api_msg_service.health.v3.HealthCheckSpecifier>`, making only the Endpoints and Health Checkers that changed be reconstructed on receiving a new message, rather than the entire HDS.
 * health_check: added option to use :ref:`no_traffic_healthy_interval <envoy_v3_api_field_config.core.v3.HealthCheck.no_traffic_healthy_interval>` which allows a different no traffic interval when the host is healthy.
 * mongo_proxy: the list of commands to produce metrics for is now :ref:`configurable <envoy_v3_api_field_extensions.filters.network.mongo_proxy.v3.MongoProxy.commands>`.
 * ratelimit: added :ref:`disable_x_envoy_ratelimited_header <envoy_v3_api_msg_extensions.filters.http.ratelimit.v3.RateLimit>` option to disable `X-Envoy-RateLimited` header.

source/common/upstream/BUILD

@@ -202,12 +202,23 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "locality_endpoint_lib",
+    hdrs = ["locality_endpoint.h"],
+    deps = [
+        "//source/common/protobuf:utility_lib",
+        "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
+        "@envoy_api//envoy/config/endpoint/v3:pkg_cc_proto",
+    ],
+)
+
 envoy_cc_library(
     name = "health_discovery_service_lib",
     srcs = ["health_discovery_service.cc"],
     hdrs = ["health_discovery_service.h"],
     deps = [
         ":health_checker_lib",
+        ":locality_endpoint_lib",
         ":upstream_includes",
         "//include/envoy/api:api_interface",
         "//include/envoy/event:dispatcher_interface",

source/common/upstream/health_checker_base_impl.h

@@ -37,6 +37,23 @@ struct HealthCheckerStats {
   ALL_HEALTH_CHECKER_STATS(GENERATE_COUNTER_STRUCT, GENERATE_GAUGE_STRUCT)
 };
 
+/**
+ * HealthCheckerHash and HealthCheckerEqualTo are used to allow the HealthCheck proto to be used as
+ * a flat_hash_map key.
+ */
+struct HealthCheckerHash {
+  size_t operator()(const envoy::config::core::v3::HealthCheck& health_check) const {
+    return MessageUtil::hash(health_check);
+  }
+};
+
+struct HealthCheckerEqualTo {
+  bool operator()(const envoy::config::core::v3::HealthCheck& lhs,
+                  const envoy::config::core::v3::HealthCheck& rhs) const {
+    return Protobuf::util::MessageDifferencer::Equals(lhs, rhs);
+  }
+};
+
 /**
  * Base implementation for all health checkers.
  */

source/common/upstream/health_discovery_service.h

@@ -18,15 +18,24 @@
 #include "common/grpc/async_client_impl.h"
 #include "common/network/resolver_impl.h"
 #include "common/upstream/health_checker_impl.h"
+#include "common/upstream/locality_endpoint.h"
 #include "common/upstream/upstream_impl.h"
 
 #include "server/transport_socket_config_impl.h"
 
 #include "extensions/transport_sockets/well_known_names.h"
 
+#include "absl/container/flat_hash_map.h"
+
 namespace Envoy {
 namespace Upstream {
 
+using HostsMap = absl::flat_hash_map<LocalityEndpointTuple, HostSharedPtr, LocalityEndpointHash,
+                                     LocalityEndpointEqualTo>;
+using HealthCheckerMap =
+    absl::flat_hash_map<envoy::config::core::v3::HealthCheck, Upstream::HealthCheckerSharedPtr,
+                        HealthCheckerHash, HealthCheckerEqualTo>;
+
 class ProdClusterInfoFactory : public ClusterInfoFactory, Logger::Loggable<Logger::Id::upstream> {
 public:
   ClusterInfoConstSharedPtr createClusterInfo(const CreateClusterInfoParams& params) override;
@@ -60,12 +69,19 @@ class HdsCluster : public Cluster, Logger::Loggable<Logger::Id::upstream> {
   Outlier::Detector* outlierDetector() override { return outlier_detector_.get(); }
   const Outlier::Detector* outlierDetector() const override { return outlier_detector_.get(); }
   void initialize(std::function<void()> callback) override;
-
-  // Creates and starts healthcheckers to its endpoints
-  void startHealthchecks(AccessLog::AccessLogManager& access_log_manager, Runtime::Loader& runtime,
-                         Event::Dispatcher& dispatcher, Api::Api& api);
+  // Compare changes in the cluster proto, and update parts of the cluster as needed.
+  void update(Server::Admin& admin, envoy::config::cluster::v3::Cluster cluster,
+              ClusterInfoFactory& info_factory, ClusterManager& cm,
+              const LocalInfo::LocalInfo& local_info, Event::Dispatcher& dispatcher,
+              Singleton::Manager& singleton_manager, ThreadLocal::SlotAllocator& tls,
+              ProtobufMessage::ValidationVisitor& validation_visitor, Api::Api& api,
+              AccessLog::AccessLogManager& access_log_manager, Runtime::Loader& runtime);
+  // Creates healthcheckers and adds them to the list, then does initial start.
+  void initHealthchecks(AccessLog::AccessLogManager& access_log_manager, Runtime::Loader& runtime,
+                        Event::Dispatcher& dispatcher, Api::Api& api);
 
   std::vector<Upstream::HealthCheckerSharedPtr> healthCheckers() { return health_checkers_; };
+  std::vector<HostSharedPtr> hosts() { return *hosts_; };
 
 protected:
   PrioritySetImpl priority_set_;
@@ -76,17 +92,31 @@ class HdsCluster : public Cluster, Logger::Loggable<Logger::Id::upstream> {
   std::function<void()> initialization_complete_callback_;
 
   Runtime::Loader& runtime_;
-  const envoy::config::cluster::v3::Cluster cluster_;
+  envoy::config::cluster::v3::Cluster cluster_;
   const envoy::config::core::v3::BindConfig& bind_config_;
   Stats::Store& stats_;
   Ssl::ContextManager& ssl_context_manager_;
   bool added_via_api_;
+  bool initialized_ = false;
+  uint64_t config_hash_;
+  uint64_t socket_match_hash_;
 
-  HostVectorSharedPtr initial_hosts_;
-  HostsPerLocalitySharedPtr initial_hosts_per_locality_;
+  HostVectorSharedPtr hosts_;
+  HostsPerLocalitySharedPtr hosts_per_locality_;
+  HostsMap hosts_map_;
   ClusterInfoConstSharedPtr info_;
   std::vector<Upstream::HealthCheckerSharedPtr> health_checkers_;
+  HealthCheckerMap health_checkers_map_;
   ProtobufMessage::ValidationVisitor& validation_visitor_;
+
+  void updateHealthchecks(
+      const Protobuf::RepeatedPtrField<envoy::config::core::v3::HealthCheck>& health_checks,
+      AccessLog::AccessLogManager& access_log_manager, Runtime::Loader& runtime,
+      Event::Dispatcher& dispatcher, Api::Api& api);
+  void
+  updateHosts(const Protobuf::RepeatedPtrField<envoy::config::endpoint::v3::LocalityLbEndpoints>&
+                  locality_endpoints,
+              bool update_socket_matches);
 };
 
 using HdsClusterPtr = std::shared_ptr<HdsCluster>;
@@ -97,7 +127,8 @@ using HdsClusterPtr = std::shared_ptr<HdsCluster>;
 #define ALL_HDS_STATS(COUNTER)                                                                     \
   COUNTER(requests)                                                                                \
   COUNTER(responses)                                                                               \
-  COUNTER(errors)
+  COUNTER(errors)                                                                                  \
+  COUNTER(updates)
 
 /**
  * Struct definition for all hds stats. @see stats_macros.h
@@ -145,7 +176,11 @@ class HdsDelegate : Grpc::AsyncStreamCallbacks<envoy::service::health::v3::Healt
   // Establishes a connection with the management server
   void establishNewStream();
   void processMessage(std::unique_ptr<envoy::service::health::v3::HealthCheckSpecifier>&& message);
-
+  envoy::config::cluster::v3::Cluster
+  createClusterConfig(const envoy::service::health::v3::ClusterHealthCheck& cluster_health_check);
+  void updateHdsCluster(HdsClusterPtr cluster,
+                        const envoy::config::cluster::v3::Cluster& cluster_health_check);
+  HdsClusterPtr createHdsCluster(const envoy::config::cluster::v3::Cluster& cluster_health_check);
   HdsDelegateStats stats_;
   const Protobuf::MethodDescriptor& service_method_;
 
@@ -168,10 +203,11 @@ class HdsDelegate : Grpc::AsyncStreamCallbacks<envoy::service::health::v3::Healt
   ThreadLocal::SlotAllocator& tls_;
 
   envoy::service::health::v3::HealthCheckRequestOrEndpointHealthResponse health_check_request_;
-  std::unique_ptr<envoy::service::health::v3::HealthCheckSpecifier> health_check_message_;
+  uint64_t specifier_hash_;
 
   std::vector<std::string> clusters_;
   std::vector<HdsClusterPtr> hds_clusters_;
+  absl::flat_hash_map<std::string, HdsClusterPtr> hds_clusters_name_map_;
 
   Event::TimerPtr hds_stream_response_timer_;
   Event::TimerPtr hds_retry_timer_;

source/common/upstream/locality_endpoint.h

@@ -0,0 +1,29 @@
+#pragma once
+
+#include "envoy/config/core/v3/base.pb.h"
+#include "envoy/config/endpoint/v3/endpoint_components.pb.h"
+
+#include "common/protobuf/utility.h"
+
+namespace Envoy {
+namespace Upstream {
+
+using LocalityEndpointTuple = std::tuple<const envoy::config::core::v3::Locality&,
+                                         const envoy::config::endpoint::v3::LbEndpoint&>;
+struct LocalityEndpointHash {
+  size_t operator()(const LocalityEndpointTuple& values) const {
+    const auto locality_hash = MessageUtil::hash(std::get<0>(values));
+    const auto endpoint_hash = MessageUtil::hash(std::get<1>(values));
+    return locality_hash ^ endpoint_hash;
+  }
+};
+
+struct LocalityEndpointEqualTo {
+  bool operator()(const LocalityEndpointTuple& lhs, const LocalityEndpointTuple& rhs) const {
+    return Protobuf::util::MessageDifferencer::Equals(std::get<0>(lhs), std::get<0>(rhs)) &&
+           Protobuf::util::MessageDifferencer::Equals(std::get<1>(lhs), std::get<1>(rhs));
+  }
+};
+
+} // namespace Upstream
+} // namespace Envoy