: stats_{ALL_HDS_STATS(POOL_COUNTER_PREFIX(scope, "hds_delegate."))},

      service_method_(*Protobuf::DescriptorPool::generated_pool()->FindMethodByName(

          "envoy.service.health.v3.HealthDiscoveryService.StreamHealthCheck")),

      async_client_(std::move(async_client)), dispatcher_(server_context.mainThreadDispatcher()),

      server_context_(server_context), store_stats_(stats),

      ssl_context_manager_(ssl_context_manager),

      info_factory_(std::make_unique<ProdClusterInfoFactory>()),

      tls_(server_context_.threadLocal()) {

  health_check_request_.mutable_health_check_request()->mutable_node()->MergeFrom(

      server_context.localInfo().node());

  backoff_strategy_ = std::make_unique<JitteredExponentialBackOffStrategy>(

      RetryInitialDelayMilliseconds, RetryMaxDelayMilliseconds,

      server_context_.api().randomGenerator());

  hds_retry_timer_ = dispatcher_.createTimer([this]() -> void { establishNewStream(); });

  hds_stream_response_timer_ = dispatcher_.createTimer([this]() -> void { sendResponse(); });

  health_check_request_.mutable_health_check_request()

      ->mutable_capability()

      ->add_health_check_protocols(envoy::service::health::v3::Capability::HTTP);

  health_check_request_.mutable_health_check_request()

      ->mutable_capability()

      ->add_health_check_protocols(envoy::service::health::v3::Capability::TCP);

  establishNewStream();

}

void HdsDelegate::setHdsRetryTimer() {

  const auto retry_ms = std::chrono::milliseconds(backoff_strategy_->nextBackOffMs());

  ENVOY_LOG(warn, "HdsDelegate stream/connection failure, will retry in {} ms.", retry_ms.count());

  hds_retry_timer_->enableTimer(retry_ms);

}

void HdsDelegate::setHdsStreamResponseTimer() {

  hds_stream_response_timer_->enableTimer(std::chrono::milliseconds(server_response_ms_));

}

void HdsDelegate::establishNewStream() {

  ENVOY_LOG(debug, "Establishing new gRPC bidi stream for {}", service_method_.DebugString());

  stream_ = async_client_->start(service_method_, *this, Http::AsyncClient::StreamOptions());

  if (stream_ == nullptr) {

    ENVOY_LOG(warn, "Unable to establish new stream");

    handleFailure();

    return;

  }

  ENVOY_LOG(debug, "Sending HealthCheckRequest {} ", health_check_request_.DebugString());

  stream_->sendMessage(health_check_request_, false);

  stats_.responses_.inc();

  backoff_strategy_->reset();

}

void HdsDelegate::handleFailure() {

  stats_.errors_.inc();

  setHdsRetryTimer();

}

envoy::service::health::v3::HealthCheckRequestOrEndpointHealthResponse HdsDelegate::sendResponse() {

  envoy::service::health::v3::HealthCheckRequestOrEndpointHealthResponse response;

  for (const auto& cluster : hds_clusters_) {

    auto* cluster_health =

        response.mutable_endpoint_health_response()->add_cluster_endpoints_health();

    cluster_health->set_cluster_name(cluster->info()->name());

    for (const auto& hosts : cluster->prioritySet().hostSetsPerPriority()) {

      for (const auto& locality_hosts : hosts->hostsPerLocality().get()) {

        envoy::service::health::v3::LocalityEndpointsHealth* locality_health =

            cluster_health->add_locality_endpoints_health();

        locality_health->mutable_locality()->MergeFrom(locality_hosts[0]->locality());

        for (const auto& host : locality_hosts) {

          auto* endpoint = locality_health->add_endpoints_health();

          Network::Utility::addressToProtobufAddress(

              *host->address(), *endpoint->mutable_endpoint()->mutable_address());

          if (host->coarseHealth() == Host::Health::Healthy) {

            endpoint->set_health_status(envoy::config::core::v3::HEALTHY);

          } else {

            if (host->healthFlagGet(Host::HealthFlag::ACTIVE_HC_TIMEOUT)) {

              endpoint->set_health_status(envoy::config::core::v3::TIMEOUT);

            } else {

              endpoint->set_health_status(envoy::config::core::v3::UNHEALTHY);

            }

          }

          response.mutable_endpoint_health_response()->add_endpoints_health()->MergeFrom(*endpoint);

        }

      }

    }

  }

  ENVOY_LOG(debug, "Sending EndpointHealthResponse to server {}", response.DebugString());

  stream_->sendMessage(response, false);

  stats_.responses_.inc();

  setHdsStreamResponseTimer();

  return response;

}

void HdsDelegate::onCreateInitialMetadata(Http::RequestHeaderMap& metadata) {

  UNREFERENCED_PARAMETER(metadata);

}

void HdsDelegate::onReceiveInitialMetadata(Http::ResponseHeaderMapPtr&& metadata) {

  UNREFERENCED_PARAMETER(metadata);

}

    const envoy::service::health::v3::ClusterHealthCheck& cluster_health_check) {

  envoy::config::cluster::v3::Cluster cluster_config;

  cluster_config.set_name(cluster_health_check.cluster_name());

  cluster_config.mutable_connect_timeout()->set_seconds(ClusterTimeoutSeconds);

  cluster_config.mutable_per_connection_buffer_limit_bytes()->set_value(

      ClusterConnectionBufferLimitBytes);

  for (const auto& locality_endpoints : cluster_health_check.locality_endpoints()) {

    auto* endpoints = cluster_config.mutable_load_assignment()->add_endpoints();

    if (locality_endpoints.has_locality()) {

      endpoints->mutable_locality()->MergeFrom(locality_endpoints.locality());

    }

    for (const auto& endpoint : locality_endpoints.endpoints()) {

      if (endpoint.has_health_check_config() &&

          endpoint.health_check_config().disable_active_health_check()) {

        ENVOY_LOG(debug, "Skip adding the endpoint {} with optional disabled health check for HDS.",

                  endpoint.DebugString());

        continue;

      }

      auto* new_endpoint = endpoints->add_lb_endpoints()->mutable_endpoint();

      new_endpoint->mutable_address()->MergeFrom(endpoint.address());

      new_endpoint->mutable_health_check_config()->MergeFrom(endpoint.health_check_config());

    }

  }

  for (auto& health_check : cluster_health_check.health_checks()) {

    cluster_config.add_health_checks()->MergeFrom(health_check);

  }

  cluster_config.mutable_transport_socket_matches()->MergeFrom(

      cluster_health_check.transport_socket_matches());

  ENVOY_LOG(debug, "New HdsCluster config {} ", cluster_config.DebugString());

  return cluster_config;

}

                              const envoy::config::core::v3::BindConfig& bind_config) {

  return cluster->update(cluster_config, bind_config, *info_factory_, tls_);

}

                              const envoy::config::core::v3::BindConfig& bind_config) {

  auto new_cluster =

      std::make_shared<HdsCluster>(server_context_, std::move(cluster_config), bind_config,

                                   store_stats_, ssl_context_manager_, false, *info_factory_, tls_);

  new_cluster->initialize([] { return absl::OkStatus(); });

  new_cluster->initHealthchecks();

  return new_cluster;

}

    std::unique_ptr<envoy::service::health::v3::HealthCheckSpecifier>&& message) {

  ENVOY_LOG(debug, "New health check response message {} ", message->DebugString());

  ASSERT(message);

  std::vector<HdsClusterPtr> hds_clusters;

  absl::flat_hash_map<std::string, HdsClusterPtr> new_hds_clusters_name_map;

  for (const auto& cluster_health_check : message->cluster_health_checks()) {

    if (!new_hds_clusters_name_map.contains(cluster_health_check.cluster_name())) {

      HdsClusterPtr cluster_ptr;

      auto cluster_config = createClusterConfig(cluster_health_check);

      auto cluster_map_pair = hds_clusters_name_map_.find(cluster_health_check.cluster_name());

      if (cluster_map_pair != hds_clusters_name_map_.end()) {

        cluster_ptr = cluster_map_pair->second;

        absl::Status status = updateHdsCluster(cluster_ptr, cluster_config,

                                               cluster_health_check.upstream_bind_config());

        if (!status.ok()) {

          return status;

        }

      } else {

        cluster_ptr = createHdsCluster(cluster_config, cluster_health_check.upstream_bind_config());

      }

      if (!cluster_health_check.cluster_name().empty()) {

        new_hds_clusters_name_map.insert({cluster_health_check.cluster_name(), cluster_ptr});

      } else {

        ENVOY_LOG(warn,

                  "HDS Cluster has no cluster_name, it will be recreated instead of updated on "

                  "every reconfiguration.");

      }

      hds_clusters.push_back(cluster_ptr);

    } else {

      ENVOY_LOG(warn, "An HDS Cluster with this cluster_name has already been added, not using.");

    }

  }

  hds_clusters_name_map_ = std::move(new_hds_clusters_name_map);

  hds_clusters_ = std::move(hds_clusters);

  return absl::OkStatus();

}

    std::unique_ptr<envoy::service::health::v3::HealthCheckSpecifier>&& message) {

  stats_.requests_.inc();

  ENVOY_LOG(debug, "New health check response message {} ", message->DebugString());

  const uint64_t hash = MessageUtil::hash(*message);

  if (hash == specifier_hash_) {

    ENVOY_LOG(debug, "New health check specifier is unchanged, no action taken.");

    return;

  }

  TRY_ASSERT_MAIN_THREAD {

    MessageUtil::validate(*message,

                          server_context_.messageValidationContext().dynamicValidationVisitor());

  }

  END_TRY

  CATCH(const ProtoValidationException& ex, {

    stats_.errors_.inc();

    ENVOY_LOG(warn, "Unable to validate health check specifier: {}", ex.what());

    return;

  });

  auto server_response_ms = PROTOBUF_GET_MS_OR_DEFAULT(*message, interval, 1000);

  absl::Status status = processMessage(std::move(message));

  if (!status.ok()) {

    stats_.errors_.inc();

    ENVOY_LOG(warn, "Unable to validate health check specifier: {}", status.message());

    return;

  }

  stats_.updates_.inc();

  specifier_hash_ = hash;

  if (server_response_ms_ != server_response_ms) {

    server_response_ms_ = server_response_ms;

    setHdsStreamResponseTimer();

  }

}

void HdsDelegate::onReceiveTrailingMetadata(Http::ResponseTrailerMapPtr&& metadata) {

  UNREFERENCED_PARAMETER(metadata);

}

void HdsDelegate::onRemoteClose(Grpc::Status::GrpcStatus status, const std::string& message) {

  ENVOY_LOG(warn, "{} gRPC config stream closed: {}, {}", service_method_.name(), status, message);

  hds_stream_response_timer_->disableTimer();

  stream_ = nullptr;

  server_response_ms_ = 0;

  specifier_hash_ = 0;

  handleFailure();

}

    : server_context_(server_context), cluster_(std::move(cluster)), stats_(stats),

      ssl_context_manager_(ssl_context_manager), added_via_api_(added_via_api),

      hosts_(new HostVector()) {

  ENVOY_LOG(debug, "Creating an HdsCluster");

  priority_set_.getOrCreateHostSet(0);

  config_hash_ = MessageUtil::hash(cluster_);

  socket_match_hash_ = RepeatedPtrUtil::hash(cluster_.transport_socket_matches());

  info_ = info_factory.createClusterInfo(

      {server_context, cluster_, bind_config, stats_, ssl_context_manager_, added_via_api_, tls});

  std::vector<HostVector> hosts_by_locality;

  hosts_by_locality.reserve(cluster_.load_assignment().endpoints_size());

  for (const auto& locality_endpoints : cluster_.load_assignment().endpoints()) {

    hosts_by_locality.emplace_back();

    hosts_by_locality.back().reserve(locality_endpoints.lb_endpoints_size());

    for (const auto& host : locality_endpoints.lb_endpoints()) {

      const LocalityEndpointTuple endpoint_key = {locality_endpoints.locality(), host};

      auto address_or_error = Network::Address::resolveProtoAddress(host.endpoint().address());

      THROW_IF_NOT_OK_REF(address_or_error.status());

      auto const_locality_shared_pool = LocalityPool::getConstLocalitySharedPool(

          server_context_.singletonManager(), server_context_.mainThreadDispatcher());

      HostSharedPtr endpoint = std::shared_ptr<HostImpl>(THROW_OR_RETURN_VALUE(

          HostImpl::create(info_, "", std::move(address_or_error.value()), nullptr, nullptr, 1,

                           const_locality_shared_pool->getObject(locality_endpoints.locality()),

                           host.endpoint().health_check_config(), 0,

                           envoy::config::core::v3::UNKNOWN),

          std::unique_ptr<HostImpl>));

      hosts_->push_back(endpoint);

      hosts_by_locality.back().push_back(endpoint);

      hosts_map_.insert({endpoint_key, endpoint});

    }

  }

  hosts_per_locality_ =

      std::make_shared<Envoy::Upstream::HostsPerLocalityImpl>(std::move(hosts_by_locality), false);

}

                                ClusterInfoFactory& info_factory, ThreadLocal::SlotAllocator& tls) {

  const uint64_t config_hash = MessageUtil::hash(cluster);

  if (config_hash_ != config_hash) {

    config_hash_ = config_hash;

    cluster_ = std::move(cluster);

    bool update_cluster_info = false;

    const uint64_t socket_match_hash = RepeatedPtrUtil::hash(cluster_.transport_socket_matches());

    if (socket_match_hash_ != socket_match_hash) {

      socket_match_hash_ = socket_match_hash;

      update_cluster_info = true;

      info_ = info_factory.createClusterInfo({server_context_, cluster_, bind_config, stats_,

                                              ssl_context_manager_, added_via_api_, tls});

    }

    updateHosts(cluster_.load_assignment().endpoints(), update_cluster_info);

    absl::Status status = updateHealthchecks(cluster_.health_checks());

    if (!status.ok()) {

      return status;

    }

  }

  return absl::OkStatus();

}

    const Protobuf::RepeatedPtrField<envoy::config::core::v3::HealthCheck>& health_checks) {

  std::vector<Upstream::HealthCheckerSharedPtr> health_checkers;

  HealthCheckerMap health_checkers_map;

  for (const auto& health_check : health_checks) {

    auto health_checker = health_checkers_map_.find(health_check);

    if (health_checker != health_checkers_map_.end()) {

      health_checkers_map.insert({health_check, health_checker->second});

      health_checkers.push_back(health_checker->second);

    } else {

      auto checker_or_error =

          Upstream::HealthCheckerFactory::create(health_check, *this, server_context_);

      RETURN_IF_NOT_OK_REF(checker_or_error.status());

      auto new_health_checker = checker_or_error.value();

      health_checkers_map.insert({health_check, new_health_checker});

      health_checkers.push_back(new_health_checker);

      new_health_checker->start();

    }

  }

  health_checkers_ = std::move(health_checkers);

  health_checkers_map_ = std::move(health_checkers_map);

  return absl::OkStatus();

}

    bool update_cluster_info) {

  HostVectorSharedPtr hosts = std::make_shared<std::vector<HostSharedPtr>>();

  std::vector<HostSharedPtr> hosts_added;

  std::vector<HostSharedPtr> hosts_removed;

  std::vector<HostVector> hosts_by_locality;

  HostsMap hosts_map;

  for (auto& endpoints : locality_endpoints) {

    hosts_by_locality.emplace_back();

    for (auto& endpoint : endpoints.lb_endpoints()) {

      LocalityEndpointTuple endpoint_key = {endpoints.locality(), endpoint};

      auto host_pair = hosts_map_.find(endpoint_key);

      HostSharedPtr host;

      if (!update_cluster_info && host_pair != hosts_map_.end()) {

        host = host_pair->second;

      } else {

        auto address_or_error =

            Network::Address::resolveProtoAddress(endpoint.endpoint().address());

        THROW_IF_NOT_OK_REF(address_or_error.status());

        auto const_locality_shared_pool = LocalityPool::getConstLocalitySharedPool(

            server_context_.singletonManager(), server_context_.mainThreadDispatcher());

        host = std::shared_ptr<HostImpl>(THROW_OR_RETURN_VALUE(

            HostImpl::create(info_, "", std::move(address_or_error.value()), nullptr, nullptr, 1,

                             const_locality_shared_pool->getObject(endpoints.locality()),

                             endpoint.endpoint().health_check_config(), 0,

                             envoy::config::core::v3::UNKNOWN),

            std::unique_ptr<HostImpl>));

        host->healthFlagSet(Host::HealthFlag::FAILED_ACTIVE_HC);

        hosts_added.push_back(host);

      }

      hosts_by_locality.back().push_back(host);

      hosts->push_back(host);

      hosts_map.insert({endpoint_key, host});

    }

  }

  for (auto& host_pair : hosts_map_) {

    if (!hosts_map.contains(host_pair.first)) {

      hosts_removed.push_back(host_pair.second);

    }

  }

  hosts_ = std::move(hosts);

  hosts_map_ = std::move(hosts_map);

  ENVOY_LOG(debug, "Hosts Added: {}, Removed: {}, Reused: {}", hosts_added.size(),

            hosts_removed.size(), hosts_->size() - hosts_added.size());

  hosts_per_locality_ =

      std::make_shared<Envoy::Upstream::HostsPerLocalityImpl>(std::move(hosts_by_locality), false);

  priority_set_.updateHosts(0, HostSetImpl::partitionHosts(hosts_, hosts_per_locality_), {},

                            hosts_added, hosts_removed, absl::nullopt, absl::nullopt);

}

void HdsCluster::initHealthchecks() {

  for (auto& health_check : cluster_.health_checks()) {

    auto health_checker_or_error =

        Upstream::HealthCheckerFactory::create(health_check, *this, server_context_);

    THROW_IF_NOT_OK_REF(health_checker_or_error.status());

    auto health_checker = health_checker_or_error.value();

    health_checkers_.push_back(health_checker);

    health_checkers_map_.insert({health_check, health_checker});

    health_checker->start();

  }

}

void HdsCluster::initialize(std::function<absl::Status()> callback) {

  initialization_complete_callback_ = callback;

  if (!initialized_) {

    for (const auto& host : *hosts_) {

      host->healthFlagSet(Host::HealthFlag::FAILED_ACTIVE_HC);

    }

    priority_set_.updateHosts(0, HostSetImpl::partitionHosts(hosts_, hosts_per_locality_), {},

                              *hosts_, {}, absl::nullopt, absl::nullopt);

    initialized_ = true;

  }

}