[WIP] HdsDelegate can healthcheck an endpoint

include/envoy/upstream/BUILD

@@ -40,6 +40,14 @@ envoy_cc_library(
     hdrs = ["health_check_host_monitor.h"],
 )
 
+envoy_cc_library(
+    name = "health_discovery_service_lib",
+    hdrs = ["health_discovery_service.h"],
+    deps = [
+        "//include/envoy/upstream:upstream_interface",
+    ],
+)
+
 envoy_cc_library(
     name = "host_description_interface",
     hdrs = ["host_description.h"],

include/envoy/upstream/health_discovery_service.h

@@ -0,0 +1,25 @@
+#include "common/upstream/upstream_impl.h"
+
+namespace Envoy {
+namespace Upstream {
+
+/**
+ * Factory for unit testing Upstream::HdsDelegate
+ */
+
+class ClusterInfoFactory {
+public:
+  virtual ~ClusterInfoFactory() {}
+  /**
+   * @return ClusterInfoConstSharedPtr
+   */
+
+  virtual ClusterInfoConstSharedPtr
+  createHdsClusterInfo(Runtime::Loader& runtime, const envoy::api::v2::Cluster& cluster,
+                       const envoy::api::v2::core::BindConfig& bind_config, Stats::Store& stats,
+                       Ssl::ContextManager& ssl_context_manager,
+                       Secret::SecretManager& secret_manager, bool added_via_api) PURE;
+};
+
+} // namespace Upstream
+} // namespace Envoy

source/common/upstream/BUILD

@@ -164,10 +164,18 @@ envoy_cc_library(
     srcs = ["health_discovery_service.cc"],
     hdrs = ["health_discovery_service.h"],
     deps = [
+        ":health_checker_lib",
+        ":upstream_includes",
         "//include/envoy/event:dispatcher_interface",
+        "//include/envoy/runtime:runtime_interface",
+        "//include/envoy/ssl:context_manager_interface",
         "//include/envoy/stats:stats_macros",
+        "//include/envoy/upstream:cluster_manager_interface",
+        "//include/envoy/upstream:health_discovery_service_lib",
+        "//include/envoy/upstream:upstream_interface",
         "//source/common/common:minimal_logger_lib",
         "//source/common/grpc:async_client_lib",
+        "//source/common/network:resolver_lib",
         "@envoy_api//envoy/service/discovery/v2:hds_cc",
     ],
 )

source/common/upstream/health_discovery_service.cc

@@ -6,19 +6,30 @@ namespace Envoy {
 namespace Upstream {
 
 HdsDelegate::HdsDelegate(const envoy::api::v2::core::Node& node, Stats::Scope& scope,
-                         Grpc::AsyncClientPtr async_client, Event::Dispatcher& dispatcher)
+                         Grpc::AsyncClientPtr async_client, Event::Dispatcher& dispatcher,
+                         Runtime::Loader& runtime, Envoy::Stats::Store& stats,
+                         Ssl::ContextManager& ssl_context_manager,
+                         Secret::SecretManager& secret_manager, Runtime::RandomGenerator& random,
+                         ClusterInfoFactory& info_factory)
     : stats_{ALL_HDS_STATS(POOL_COUNTER_PREFIX(scope, "hds_delegate."))},
       async_client_(std::move(async_client)),
       service_method_(*Protobuf::DescriptorPool::generated_pool()->FindMethodByName(
-          "envoy.service.discovery.v2.HealthDiscoveryService.StreamHealthCheck")) {
+          "envoy.service.discovery.v2.HealthDiscoveryService.StreamHealthCheck")),
+      runtime_(runtime), store_stats(stats), ssl_context_manager_(ssl_context_manager),
+      secret_manager_(secret_manager), random_(random), dispatcher_(dispatcher),
+      info_factory_(info_factory) {
   health_check_request_.mutable_node()->MergeFrom(node);
   retry_timer_ = dispatcher.createTimer([this]() -> void { establishNewStream(); });
-  response_timer_ = dispatcher.createTimer([this]() -> void { sendHealthCheckRequest(); });
+  server_response_timer_ = dispatcher.createTimer([this]() -> void { sendResponse(); });
   establishNewStream();
 }
 
 void HdsDelegate::setRetryTimer() {
-  retry_timer_->enableTimer(std::chrono::milliseconds(RETRY_DELAY_MS));
+  retry_timer_->enableTimer(std::chrono::milliseconds(RetryDelayMilliseconds));
+}
+
+void HdsDelegate::setServerResponseTimer() {
+  server_response_timer_->enableTimer(std::chrono::milliseconds(server_response_ms_));
 }
 
 void HdsDelegate::establishNewStream() {
@@ -30,22 +41,48 @@ void HdsDelegate::establishNewStream() {
     return;
   }
 
-  sendHealthCheckRequest();
-}
-
-void HdsDelegate::sendHealthCheckRequest() {
-  ENVOY_LOG(debug, "Sending HealthCheckRequest");
+  // TODO(lilika): Add support for other types of healthchecks
+  health_check_request_.mutable_capability()->add_health_check_protocol(
+      envoy::service::discovery::v2::Capability::HTTP);
+  ENVOY_LOG(debug, "Sending HealthCheckRequest {} ", health_check_request_.DebugString());
   stream_->sendMessage(health_check_request_, false);
   stats_.responses_.inc();
 }
 
+// TODO(lilika) : Use jittered backoff as in https://github.com/envoyproxy/envoy/pull/3791
 void HdsDelegate::handleFailure() {
-  ENVOY_LOG(warn, "Load reporter stats stream/connection failure, will retry in {} ms.",
-            RETRY_DELAY_MS);
+  ENVOY_LOG(warn, "HdsDelegate stream/connection failure, will retry in {} ms.",
+            RetryDelayMilliseconds);
   stats_.errors_.inc();
   setRetryTimer();
 }
 
+// TODO(lilika): Add support for the same endpoint in different clusters/ports
+void HdsDelegate::sendResponse() {
+  envoy::service::discovery::v2::HealthCheckRequestOrEndpointHealthResponse response;
+  for (uint32_t i = 0; i < hds_clusters_.size(); i++) {
+    for (const auto& hosts : hds_clusters_[i]->prioritySet().hostSetsPerPriority()) {
+      for (const auto& host : hosts->hosts()) {
+        auto* endpoint = response.mutable_endpoint_health_response()->add_endpoints_health();
+        endpoint->mutable_endpoint()->mutable_address()->mutable_socket_address()->set_address(
+            host->address()->ip()->addressAsString());
+        endpoint->mutable_endpoint()->mutable_address()->mutable_socket_address()->set_port_value(
+            host->address()->ip()->port());
+        // TODO(lilika): Add support for more granular options of envoy::api::v2::core::HealthStatus
+        if (host->healthy()) {
+          endpoint->set_health_status(envoy::api::v2::core::HealthStatus::HEALTHY);
+        } else {
+          endpoint->set_health_status(envoy::api::v2::core::HealthStatus::UNHEALTHY);
+        }
+      }
+    }
+  }
+  ENVOY_LOG(debug, "Sending EndpointHealthResponse to server {}", response.DebugString());
+  stream_->sendMessage(response, false);
+  stats_.responses_.inc();
+  setServerResponseTimer();
+}
+
 void HdsDelegate::onCreateInitialMetadata(Http::HeaderMap& metadata) {
   UNREFERENCED_PARAMETER(metadata);
 }
@@ -54,12 +91,71 @@ void HdsDelegate::onReceiveInitialMetadata(Http::HeaderMapPtr&& metadata) {
   UNREFERENCED_PARAMETER(metadata);
 }
 
+void HdsDelegate::processMessage(
+    std::unique_ptr<envoy::service::discovery::v2::HealthCheckSpecifier>&& message) {
+  ENVOY_LOG(debug, "New health check response message {} ", message->DebugString());
+  ASSERT(message);
+  for (int i = 0; i < message->health_check_size(); i++) {
+    // Create HdsCluster config
+    ENVOY_LOG(debug, "Creating HdsCluster config");
+    envoy::api::v2::core::BindConfig bind_config;
+
+    clusters_config_.emplace_back();
+    envoy::api::v2::Cluster& cluster_config = clusters_config_.back();
+
+    cluster_config.set_name(message->health_check(i).cluster_name());
+    cluster_config.mutable_connect_timeout()->set_seconds(ClusterTimeoutSeconds);
+    cluster_config.mutable_per_connection_buffer_limit_bytes()->set_value(
+        ClusterConnectionBufferLimitBytes);
+
+    // Add endpoints to cluster
+    for (int j = 0; j < message->health_check(i).endpoints_size(); j++) {
+      const auto& endpoints = message->health_check(i).endpoints(j);
+      for (int k = 0; k < endpoints.endpoints_size(); k++) {
+        cluster_config.add_hosts()->MergeFrom(endpoints.endpoints(k).address());
+      }
+    }
+
+    // TODO(lilika): Add support for optional per-endpoint health checks
+
+    // Add healthcheckers to cluster
+    auto health_checkers = message->health_check(i);
+    for (int j = 0; j < health_checkers.health_checks_size(); j++) {
+      cluster_config.add_health_checks()->MergeFrom(health_checkers.health_checks(j));
+    }
+
+    ENVOY_LOG(debug, "New HdsCluster config {} ", cluster_config.DebugString());
+
+    // Create HdsCluster
+    hds_clusters_.emplace_back(new HdsCluster(runtime_, cluster_config, bind_config, store_stats,
+                                              ssl_context_manager_, secret_manager_, false,
+                                              info_factory_));
+
+    for (int j = 0; j < message->health_check(i).health_checks_size(); j++) {
+      // Creating HealthCheckers
+      health_checkers_.push_back(Upstream::HealthCheckerFactory::create(
+          cluster_config.health_checks(j), *hds_clusters_[i], runtime_, random_, dispatcher_));
+    }
+  }
+}
+
+// TODO(lilika): Add support for subsequent HealthCheckSpecifier messages that
+// might modify the HdsClusters
 void HdsDelegate::onReceiveMessage(
     std::unique_ptr<envoy::service::discovery::v2::HealthCheckSpecifier>&& message) {
-  ENVOY_LOG(debug, "New health check response ", message->DebugString());
   stats_.requests_.inc();
-  stream_->sendMessage(health_check_request_, false);
-  stats_.responses_.inc();
+  ENVOY_LOG(debug, "New health check response message {} ", message->DebugString());
+
+  // Process the HealthCheckSpecifier message
+  processMessage(std::move(message));
+
+  // Initializing HealthCheckers
+  for (auto& health_checker : health_checkers_) {
+    health_checker->start();
+  }
+
+  server_response_ms_ = PROTOBUF_GET_MS_REQUIRED(*message, interval);
+  setServerResponseTimer();
 }
 
 void HdsDelegate::onReceiveTrailingMetadata(Http::HeaderMapPtr&& metadata) {
@@ -68,10 +164,82 @@ void HdsDelegate::onReceiveTrailingMetadata(Http::HeaderMapPtr&& metadata) {
 
 void HdsDelegate::onRemoteClose(Grpc::Status::GrpcStatus status, const std::string& message) {
   ENVOY_LOG(warn, "gRPC config stream closed: {}, {}", status, message);
-  response_timer_->disableTimer();
+  server_response_timer_->disableTimer();
   stream_ = nullptr;
   handleFailure();
 }
 
+HdsCluster::HdsCluster(Runtime::Loader& runtime, const envoy::api::v2::Cluster& cluster,
+                       const envoy::api::v2::core::BindConfig& bind_config, Stats::Store& stats,
+                       Ssl::ContextManager& ssl_context_manager,
+                       Secret::SecretManager& secret_manager, bool added_via_api,
+                       ClusterInfoFactory& info_factory)
+    : runtime_(runtime), cluster_(cluster), bind_config_(bind_config), stats_(stats),
+      ssl_context_manager_(ssl_context_manager), secret_manager_(secret_manager),
+      added_via_api_(added_via_api), initial_hosts_(new HostVector()) {
+  ENVOY_LOG(debug, "Creating an HdsCluster");
+  priority_set_.getOrCreateHostSet(0);
+
+  info_ = info_factory.createHdsClusterInfo(runtime_, cluster_, bind_config_, stats_,
+                                            ssl_context_manager_, secret_manager_, added_via_api_);
+
+  for (const auto& host : cluster.hosts()) {
+    initial_hosts_->emplace_back(HostSharedPtr{new HostImpl(
+        info_, "", resolveProtoAddress2(host), envoy::api::v2::core::Metadata::default_instance(),
+        1, envoy::api::v2::core::Locality().default_instance(),
+        envoy::api::v2::endpoint::Endpoint::HealthCheckConfig().default_instance())});
+  }
+  initialize([] {});
+}
+
+ClusterSharedPtr HdsCluster::create() { NOT_IMPLEMENTED; }
+
+HostVectorConstSharedPtr HdsCluster::createHealthyHostList(const HostVector& hosts) {
+  HostVectorSharedPtr healthy_list(new HostVector());
+  for (const auto& host : hosts) {
+    if (host->healthy()) {
+      healthy_list->emplace_back(host);
+    }
+  }
+  return healthy_list;
+}
+
+ClusterInfoConstSharedPtr RealClusterInfoFactory::createHdsClusterInfo(
+    Runtime::Loader& runtime, const envoy::api::v2::Cluster& cluster,
+    const envoy::api::v2::core::BindConfig& bind_config, Stats::Store& stats,
+    Ssl::ContextManager& ssl_context_manager, Secret::SecretManager& secret_manager,
+    bool added_via_api) {
+
+  return std::make_unique<ClusterInfoImpl>(cluster, bind_config, runtime, stats,
+                                           ssl_context_manager, secret_manager, added_via_api);
+}
+
+HostsPerLocalityConstSharedPtr HdsCluster::createHealthyHostLists(const HostsPerLocality& hosts) {
+  return hosts.filter([](const Host& host) { return host.healthy(); });
+}
+
+void HdsCluster::initialize(std::function<void()> callback) {
+  ASSERT(initialization_complete_callback_ == nullptr);
+  initialization_complete_callback_ = callback;
+  for (const auto& host : *initial_hosts_) {
+    host->healthFlagSet(Host::HealthFlag::FAILED_ACTIVE_HC);
+  }
+
+  auto& first_host_set = priority_set_.getOrCreateHostSet(0);
+  auto healthy = createHealthyHostList(*initial_hosts_);
+  ENVOY_LOG(debug, "About to update the priority set.");
+  first_host_set.updateHosts(initial_hosts_, healthy, HostsPerLocalityImpl::empty(),
+                             HostsPerLocalityImpl::empty(), {}, *initial_hosts_, {});
+  initial_hosts_ = nullptr;
+  ENVOY_LOG(debug, "Updated the priority set.");
+}
+
+const Network::Address::InstanceConstSharedPtr
+HdsCluster::resolveProtoAddress2(const envoy::api::v2::core::Address& address) {
+  return Network::Address::resolveProtoAddress(address);
+}
+
+void HdsCluster::setOutlierDetector(const Outlier::DetectorSharedPtr&) { NOT_IMPLEMENTED; }
+
 } // namespace Upstream
 } // namespace Envoy