const Network::Socket::OptionsSharedPtr& to_add) {

  if (to_add != nullptr) {

    Network::Socket::appendOptions(options, to_add);

  }

}

getOrigin(const Network::TransportSocketOptionsConstSharedPtr& options, HostConstSharedPtr host) {

  std::string sni = std::string(host->transportSocketFactory().defaultServerNameIndication());

  if (options && options->serverNameOverride().has_value()) {

    sni = options->serverNameOverride().value();

  }

  if (sni.empty() || !host->address() || !host->address()->ip()) {

    return absl::nullopt;

  }

  return {{"https", sni, host->address()->ip()->port()}};

}

                          absl::string_view cluster_name) {

  bool blocking_ads_cluster = false;

  if (bootstrap.dynamic_resources().has_ads_config()) {

    const auto& ads_config_source = bootstrap.dynamic_resources().ads_config();

    blocking_ads_cluster =

        (ads_config_source.grpc_services_size() > 0 &&

         ads_config_source.grpc_services(0).has_envoy_grpc() &&

         ads_config_source.grpc_services(0).envoy_grpc().cluster_name() == cluster_name);

    if (Runtime::runtimeFeatureEnabled("envoy.restart_features.xds_failover_support")) {

      blocking_ads_cluster |=

          (ads_config_source.grpc_services_size() == 2 &&

           ads_config_source.grpc_services(1).has_envoy_grpc() &&

           ads_config_source.grpc_services(1).envoy_grpc().cluster_name() == cluster_name);

    }

  }

  return blocking_ads_cluster;

}

void ClusterManagerInitHelper::addCluster(ClusterManagerCluster& cm_cluster) {

  ASSERT(state_ != State::AllClustersInitialized);

  const auto initialize_cb = [&cm_cluster, this] {

    RETURN_IF_NOT_OK(onClusterInit(cm_cluster));

    cm_cluster.cluster().info()->configUpdateStats().warming_state_.set(0);

    return absl::OkStatus();

  };

  Cluster& cluster = cm_cluster.cluster();

  cluster.info()->configUpdateStats().warming_state_.set(1);

  if (cluster.initializePhase() == Cluster::InitializePhase::Primary) {

    primary_init_clusters_.insert_or_assign(cm_cluster.cluster().info()->name(), &cm_cluster);

    cluster.initialize(initialize_cb);

  } else {

    ASSERT(cluster.initializePhase() == Cluster::InitializePhase::Secondary);

    secondary_init_clusters_.insert_or_assign(cm_cluster.cluster().info()->name(), &cm_cluster);

    if (started_secondary_initialize_) {

      cluster.initialize(initialize_cb);

    }

  }

  ENVOY_LOG(debug, "cm init: adding: cluster={} primary={} secondary={}", cluster.info()->name(),

            primary_init_clusters_.size(), secondary_init_clusters_.size());

}

absl::Status ClusterManagerInitHelper::onClusterInit(ClusterManagerCluster& cluster) {

  ASSERT(state_ != State::AllClustersInitialized);

  RETURN_IF_NOT_OK(per_cluster_init_callback_(cluster));

  removeCluster(cluster);

  return absl::OkStatus();

}

void ClusterManagerInitHelper::removeCluster(ClusterManagerCluster& cluster) {

  if (state_ == State::AllClustersInitialized) {

    return;

  }

  absl::flat_hash_map<std::string, ClusterManagerCluster*>* cluster_map;

  if (cluster.cluster().initializePhase() == Cluster::InitializePhase::Primary) {

    cluster_map = &primary_init_clusters_;

  } else {

    ASSERT(cluster.cluster().initializePhase() == Cluster::InitializePhase::Secondary);

    cluster_map = &secondary_init_clusters_;

  }

  auto iter = cluster_map->find(cluster.cluster().info()->name());

  if (iter != cluster_map->end() && iter->second == &cluster) {

    cluster_map->erase(iter);

  }

  ENVOY_LOG(debug, "cm init: init complete: cluster={} primary={} secondary={}",

            cluster.cluster().info()->name(), primary_init_clusters_.size(),

            secondary_init_clusters_.size());

  maybeFinishInitialize();

}

void ClusterManagerInitHelper::initializeSecondaryClusters() {

  started_secondary_initialize_ = true;

  for (auto iter = secondary_init_clusters_.begin(); iter != secondary_init_clusters_.end();) {

    ClusterManagerCluster* cluster = iter->second;

    ENVOY_LOG(debug, "initializing secondary cluster {}", iter->first);

    ++iter;

    cluster->cluster().initialize([cluster, this] { return onClusterInit(*cluster); });

  }

}

void ClusterManagerInitHelper::maybeFinishInitialize() {

  ENVOY_LOG(debug, "maybe finish initialize state: {}", enumToInt(state_));

  if (state_ == State::Loading || state_ == State::WaitingToStartCdsInitialization) {

    return;

  }

  ASSERT(state_ == State::WaitingToStartSecondaryInitialization ||

         state_ == State::CdsInitialized ||

         state_ == State::WaitingForPrimaryInitializationToComplete);

  ENVOY_LOG(debug, "maybe finish initialize primary init clusters empty: {}",

            primary_init_clusters_.empty());

  if (!primary_init_clusters_.empty()) {

    return;

  } else if (state_ == State::WaitingForPrimaryInitializationToComplete) {

    state_ = State::WaitingToStartSecondaryInitialization;

    if (primary_clusters_initialized_callback_) {

      primary_clusters_initialized_callback_();

    }

    return;

  }

  ENVOY_LOG(debug, "maybe finish initialize secondary init clusters empty: {}",

            secondary_init_clusters_.empty());

  if (!secondary_init_clusters_.empty()) {

    if (!started_secondary_initialize_) {

      ENVOY_LOG(info, "cm init: initializing secondary clusters");

      const std::vector<std::string> paused_xds_types{

          Config::getTypeUrl<envoy::config::endpoint::v3::ClusterLoadAssignment>(),

          Config::getTypeUrl<envoy::config::endpoint::v3::LbEndpoint>(),

          Config::getTypeUrl<envoy::extensions::transport_sockets::tls::v3::Secret>()};

      Config::ScopedResume resume_eds_leds_sds = xds_manager_.pause(paused_xds_types);

      initializeSecondaryClusters();

    }

    return;

  }

  started_secondary_initialize_ = false;

  ENVOY_LOG(debug, "maybe finish initialize cds api ready: {}", cds_ != nullptr);

  if (state_ == State::WaitingToStartSecondaryInitialization && cds_) {

    ENVOY_LOG(info, "cm init: initializing cds");

    state_ = State::WaitingToStartCdsInitialization;

    cds_->initialize();

  } else {

    ENVOY_LOG(info, "cm init: all clusters initialized");

    state_ = State::AllClustersInitialized;

    if (initialized_callback_) {

      initialized_callback_();

    }

  }

}

void ClusterManagerInitHelper::onStaticLoadComplete() {

  ASSERT(state_ == State::Loading);

  state_ = State::WaitingForPrimaryInitializationToComplete;

  maybeFinishInitialize();

}

void ClusterManagerInitHelper::startInitializingSecondaryClusters() {

  ASSERT(state_ == State::WaitingToStartSecondaryInitialization);

  ENVOY_LOG(debug, "continue initializing secondary clusters");

  maybeFinishInitialize();

}

void ClusterManagerInitHelper::setCds(CdsApi* cds) {

  ASSERT(state_ == State::Loading);

  cds_ = cds;

  if (cds_) {

    cds_->setInitializedCb([this]() -> void {

      ASSERT(state_ == State::WaitingToStartCdsInitialization);

      state_ = State::CdsInitialized;

      maybeFinishInitialize();

    });

  }

}

    ClusterManager::InitializationCompleteCallback callback) {

  if (state_ == State::AllClustersInitialized) {

    callback();

  } else {

    initialized_callback_ = callback;

  }

}

    ClusterManager::PrimaryClustersReadyCallback callback) {

  ASSERT(state_ == State::WaitingToStartSecondaryInitialization ||

         state_ == State::WaitingForPrimaryInitializationToComplete || state_ == State::Loading);

  if (state_ == State::WaitingToStartSecondaryInitialization) {

    callback();

  } else {

    primary_clusters_initialized_callback_ = callback;

  }

}

    : context_(context), factory_(factory), runtime_(context.runtime()),

      stats_(context.serverScope().store()), tls_(context.threadLocal()),

      xds_manager_(context.xdsManager()), random_(context.api().randomGenerator()),

      deferred_cluster_creation_(bootstrap.cluster_manager().enable_deferred_cluster_creation()),

      bind_config_(bootstrap.cluster_manager().has_upstream_bind_config()

                       ? absl::make_optional(bootstrap.cluster_manager().upstream_bind_config())

                       : absl::nullopt),

      local_info_(context.localInfo()), cm_stats_(generateStats(*stats_.rootScope())),

      init_helper_(xds_manager_,

                   [this](ClusterManagerCluster& cluster) { return onClusterInit(cluster); }),

      time_source_(context.timeSource()), dispatcher_(context.mainThreadDispatcher()),

      http_context_(context.httpContext()), router_context_(context.routerContext()),

      cluster_stat_names_(stats_.symbolTable()),

      cluster_config_update_stat_names_(stats_.symbolTable()),

      cluster_lb_stat_names_(stats_.symbolTable()),

      cluster_endpoint_stat_names_(stats_.symbolTable()),

      cluster_load_report_stat_names_(stats_.symbolTable()),

      cluster_circuit_breakers_stat_names_(stats_.symbolTable()),

      cluster_request_response_size_stat_names_(stats_.symbolTable()),

      cluster_timeout_budget_stat_names_(stats_.symbolTable()),

          std::make_shared<SharedPool::ObjectSharedPool<

              const envoy::config::cluster::v3::Cluster::CommonLbConfig, MessageUtil, MessageUtil>>(

              dispatcher_)),

      shutdown_(false) {

  if (auto admin = context.admin(); admin.has_value()) {

    config_tracker_entry_ = admin->getConfigTracker().add(

        "clusters", [this](const Matchers::StringMatcher& name_matcher) {

          return dumpClusterConfigs(name_matcher);

        });

  }

  async_client_manager_ = std::make_unique<Grpc::AsyncClientManagerImpl>(

      bootstrap.grpc_async_client_manager_config(), context, context.grpcContext().statNames());

  const auto& cm_config = bootstrap.cluster_manager();

  if (cm_config.has_outlier_detection()) {

    const std::string event_log_file_path = cm_config.outlier_detection().event_log_path();

    if (!event_log_file_path.empty()) {

      auto outlier_or_error = Outlier::EventLoggerImpl::create(context.accessLogManager(),

                                                               event_log_file_path, time_source_);

      SET_AND_RETURN_IF_NOT_OK(outlier_or_error.status(), creation_status);

      outlier_event_logger_ = std::move(*outlier_or_error);

    }

  }

  if (!cm_config.local_cluster_name().empty()) {

    local_cluster_name_ = cm_config.local_cluster_name();

  }

  SET_AND_RETURN_IF_NOT_OK(xds_manager_.initialize(bootstrap, this), creation_status);

}

ClusterManagerImpl::initialize(const envoy::config::bootstrap::v3::Bootstrap& bootstrap) {

  ASSERT(!initialized_);

  initialized_ = true;

  auto is_primary_cluster = [](const envoy::config::cluster::v3::Cluster& cluster) -> bool {

    return cluster.type() != envoy::config::cluster::v3::Cluster::EDS ||

           (cluster.type() == envoy::config::cluster::v3::Cluster::EDS &&

            Config::SubscriptionFactory::isPathBasedConfigSource(

                cluster.eds_cluster_config().eds_config().config_source_specifier_case()));

  };

  for (const auto& cluster : bootstrap.static_resources().clusters()) {

    if (is_primary_cluster(cluster)) {

      primary_clusters_.insert(cluster.name());

    }

  }

  bool has_ads_cluster = false;

  for (const auto& cluster : bootstrap.static_resources().clusters()) {

    if (is_primary_cluster(cluster)) {

      const bool required_for_ads = isBlockingAdsCluster(bootstrap, cluster.name());

      has_ads_cluster |= required_for_ads;

      auto status_or_cluster =

          loadCluster(cluster, MessageUtil::hash(cluster), "", /*added_via_api=*/false,

                      required_for_ads, active_clusters_);

      RETURN_IF_NOT_OK_REF(status_or_cluster.status());

    }

  }

  RETURN_IF_NOT_OK(xds_manager_.initializeAdsConnections(bootstrap));

  for (const auto& cluster : bootstrap.static_resources().clusters()) {

    if (cluster.type() == envoy::config::cluster::v3::Cluster::EDS &&

        !Config::SubscriptionFactory::isPathBasedConfigSource(

            cluster.eds_cluster_config().eds_config().config_source_specifier_case())) {

      ASSERT(!isBlockingAdsCluster(bootstrap, cluster.name()));

      auto status_or_cluster =

          loadCluster(cluster, MessageUtil::hash(cluster), "", /*added_via_api=*/false,

                      /*required_for_ads=*/false, active_clusters_);

      if (!status_or_cluster.status().ok()) {

    }

  }

  cm_stats_.cluster_added_.add(bootstrap.static_resources().clusters().size());

  updateClusterCounts();

  absl::optional<ThreadLocalClusterManagerImpl::LocalClusterParams> local_cluster_params;

  if (local_cluster_name_) {

    auto local_cluster = active_clusters_.find(local_cluster_name_.value());

    if (local_cluster == active_clusters_.end()) {

      return absl::InvalidArgumentError(

          fmt::format("local cluster '{}' must be defined", local_cluster_name_.value()));

    }

    local_cluster_params.emplace();

    local_cluster_params->info_ = local_cluster->second->cluster().info();

    local_cluster_params->load_balancer_factory_ = local_cluster->second->loadBalancerFactory();

    local_cluster->second->setAddedOrUpdated();

  }

  tls_.set([this, local_cluster_params](Event::Dispatcher& dispatcher) {

    return std::make_shared<ThreadLocalClusterManagerImpl>(*this, dispatcher, local_cluster_params);

  });

  const auto& dyn_resources = bootstrap.dynamic_resources();

  if (dyn_resources.has_cds_config() || !dyn_resources.cds_resources_locator().empty()) {

    std::unique_ptr<xds::core::v3::ResourceLocator> cds_resources_locator;

    if (!dyn_resources.cds_resources_locator().empty()) {

      auto url_or_error =

          Config::XdsResourceIdentifier::decodeUrl(dyn_resources.cds_resources_locator());

      RETURN_IF_NOT_OK_REF(url_or_error.status());

      cds_resources_locator =

          std::make_unique<xds::core::v3::ResourceLocator>(std::move(url_or_error.value()));

    }

    const bool support_multi_ads_sources =

        bootstrap.has_default_config_source() || !bootstrap.config_sources().empty();

    auto cds_or_error = factory_.createCds(dyn_resources.cds_config(), cds_resources_locator.get(),

                                           *this, support_multi_ads_sources);

    RETURN_IF_NOT_OK_REF(cds_or_error.status())

    cds_api_ = std::move(*cds_or_error);

    init_helper_.setCds(cds_api_.get());

  } else {

    init_helper_.setCds(nullptr);

  }

  for (auto& cluster : active_clusters_) {

    init_helper_.addCluster(*cluster.second);

  }

  init_helper_.onStaticLoadComplete();

  if (!has_ads_cluster) {

    xds_manager_.adsMux()->start();

  }

  xds_manager_.startXdstpAdsMuxes();

  return absl::OkStatus();

}

    const envoy::config::bootstrap::v3::Bootstrap& bootstrap) {

  init_helper_.startInitializingSecondaryClusters();

  const auto& cm_config = bootstrap.cluster_manager();

  if (cm_config.has_load_stats_config()) {

    const auto& load_stats_config = cm_config.load_stats_config();

    absl::Status status = Config::Utility::checkTransportVersion(load_stats_config);

    RETURN_IF_NOT_OK(status);

    absl::StatusOr<Grpc::RawAsyncClientSharedPtr> client_or_error;

    if (Runtime::runtimeFeatureEnabled("envoy.restart_features.use_cached_grpc_client_for_xds")) {

      absl::StatusOr<Envoy::OptRef<const envoy::config::core::v3::GrpcService>> maybe_grpc_service =

          Envoy::Config::Utility::getGrpcConfigFromApiConfigSource(load_stats_config,

                                                                   /*grpc_service_idx*/ 0,

                                                                   /*xdstp_config_source*/ false);

      RETURN_IF_NOT_OK_REF(maybe_grpc_service.status());

      if (maybe_grpc_service.value().has_value()) {

        client_or_error = async_client_manager_->getOrCreateRawAsyncClientWithHashKey(

            Grpc::GrpcServiceConfigWithHashKey(*maybe_grpc_service.value()), *stats_.rootScope(),

            /*skip_cluster_check*/ false);

      } else {

    } else {

      auto factory_or_error = Config::Utility::factoryForGrpcApiConfigSource(

          *async_client_manager_, load_stats_config, *stats_.rootScope(), false, 0, false);

      RETURN_IF_NOT_OK_REF(factory_or_error.status());

      client_or_error = factory_or_error.value()->createUncachedRawAsyncClient();

    }

    RETURN_IF_NOT_OK_REF(client_or_error.status());

    load_stats_reporter_ = std::make_unique<LoadStatsReporterImpl>(

        local_info_, *this, *stats_.rootScope(), std::move(client_or_error.value()), dispatcher_);

  }

  return absl::OkStatus();

}

ClusterManagerStats ClusterManagerImpl::generateStats(Stats::Scope& scope) {

  const std::string final_prefix = "cluster_manager.";

  return {ALL_CLUSTER_MANAGER_STATS(POOL_COUNTER_PREFIX(scope, final_prefix),

                                    POOL_GAUGE_PREFIX(scope, final_prefix))};

}

                                                                 const std::string& thread_name) {

  const std::string final_prefix = absl::StrCat("thread_local_cluster_manager.", thread_name);

  return {ALL_THREAD_LOCAL_CLUSTER_MANAGER_STATS(POOL_GAUGE_PREFIX(scope, final_prefix))};

}

absl::Status ClusterManagerImpl::onClusterInit(ClusterManagerCluster& cm_cluster) {

  auto& cluster = cm_cluster.cluster();

  auto cluster_data = warming_clusters_.find(cluster.info()->name());

  if (cluster_data != warming_clusters_.end()) {

    clusterWarmingToActive(cluster.info()->name());

    updateClusterCounts();

  }

  cluster_data = active_clusters_.find(cluster.info()->name());

  if (cluster_data->second->thread_aware_lb_ != nullptr) {

    RETURN_IF_NOT_OK(cluster_data->second->thread_aware_lb_->initialize());

  }

  cluster_data->second->member_update_cb_ = cluster.prioritySet().addMemberUpdateCb(

      [&cluster, this](const HostVector&, const HostVector& hosts_removed) {

        if (cluster.info()->lbConfig().close_connections_on_host_set_change()) {

          for (const auto& host_set : cluster.prioritySet().hostSetsPerPriority()) {

            postThreadLocalRemoveHosts(cluster, host_set->hosts());

          }

        } else {

          if (!hosts_removed.empty()) {

            postThreadLocalRemoveHosts(cluster, hosts_removed);

          }

        }

      });

  cluster_data->second->priority_update_cb_ = cluster.prioritySet().addPriorityUpdateCb(

      [&cm_cluster, this](uint32_t priority, const HostVector& hosts_added,

                          const HostVector& hosts_removed) {

        bool scheduled = false;

        const auto merge_timeout = PROTOBUF_GET_MS_OR_DEFAULT(

            cm_cluster.cluster().info()->lbConfig(), update_merge_window, 1000);

        const bool is_mergeable = hosts_added.empty() && hosts_removed.empty();

        if (merge_timeout > 0) {

          scheduled = scheduleUpdate(cm_cluster, priority, is_mergeable, merge_timeout);

        }

        if (!scheduled) {

          cm_stats_.cluster_updated_.inc();

          postThreadLocalClusterUpdate(

              cm_cluster, ThreadLocalClusterUpdateParams(priority, hosts_added, hosts_removed));

        }

      });

  ThreadLocalClusterUpdateParams params;

  for (auto& host_set : cluster.prioritySet().hostSetsPerPriority()) {

    if (host_set->hosts().empty()) {

      continue;

    }

    params.per_priority_update_params_.emplace_back(host_set->priority(), host_set->hosts(),

                                                    HostVector{});

  }

  postThreadLocalClusterUpdate(cm_cluster, std::move(params));

  return absl::OkStatus();

}

                                        bool mergeable, const uint64_t timeout) {

  auto& updates_by_prio = updates_map_[cluster.cluster().info()->name()];

  if (!updates_by_prio) {

    updates_by_prio = std::make_unique<PendingUpdatesByPriorityMap>();

  }

  auto& updates = (*updates_by_prio)[priority];

  if (!updates) {

    updates = std::make_unique<PendingUpdates>();

  }

  const auto delta = time_source_.monotonicTime() - updates->last_updated_;

  const uint64_t delta_ms = std::chrono::duration_cast<std::chrono::milliseconds>(delta).count();

  const bool out_of_merge_window = delta_ms > timeout;

  if (out_of_merge_window || !mergeable) {

    if (mergeable && out_of_merge_window) {

      cm_stats_.update_out_of_merge_window_.inc();

    }

    if (updates->disableTimer()) {

      cm_stats_.update_merge_cancelled_.inc();

    }

    updates->last_updated_ = time_source_.monotonicTime();

    return false;

  }

  if (updates->timer_ == nullptr) {

    updates->timer_ = dispatcher_.createTimer([this, &cluster, priority, &updates]() -> void {

      applyUpdates(cluster, priority, *updates);

    });

  }

  if (!updates->timer_->enabled()) {

    updates->enableTimer(timeout);

  }

  return true;

}

                                      PendingUpdates& updates) {

  static const HostVector hosts_added;

  static const HostVector hosts_removed;

  postThreadLocalClusterUpdate(

      cluster, ThreadLocalClusterUpdateParams(priority, hosts_added, hosts_removed));

  cm_stats_.cluster_updated_via_merge_.inc();

  updates.last_updated_ = time_source_.monotonicTime();

}

                                       const bool avoid_cds_removal) {

  const std::string& cluster_name = cluster.name();

  const auto existing_active_cluster = active_clusters_.find(cluster_name);

  const auto existing_warming_cluster = warming_clusters_.find(cluster_name);

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

  if (existing_warming_cluster != warming_clusters_.end()) {

    if (existing_warming_cluster->second->blockUpdate(new_hash)) {

      return false;

    }

  } else if (existing_active_cluster != active_clusters_.end() &&

             existing_active_cluster->second->blockUpdate(new_hash)) {

    return false;

  }

  if (existing_active_cluster != active_clusters_.end() ||

      existing_warming_cluster != warming_clusters_.end()) {

    if (existing_active_cluster != active_clusters_.end()) {

      init_helper_.removeCluster(*existing_active_cluster->second);

    }

    cm_stats_.cluster_modified_.inc();

  } else {

    cm_stats_.cluster_added_.inc();

  }

  const bool all_clusters_initialized =

      init_helper_.state() == ClusterManagerInitHelper::State::AllClustersInitialized;

  auto status_or_cluster =

      loadCluster(cluster, new_hash, version_info, /*added_via_api=*/true,

                  /*required_for_ads=*/false, warming_clusters_, avoid_cds_removal);

  RETURN_IF_NOT_OK_REF(status_or_cluster.status());

  const ClusterDataPtr previous_cluster = std::move(status_or_cluster.value());

  auto& cluster_entry = warming_clusters_.at(cluster_name);

  cluster_entry->cluster_->info()->configUpdateStats().warming_state_.set(1);

  if (!all_clusters_initialized) {

    ENVOY_LOG(debug, "add/update cluster {} during init", cluster_name);

    init_helper_.addCluster(*cluster_entry);

  } else {

    ENVOY_LOG(debug, "add/update cluster {} starting warming", cluster_name);

    cluster_entry->cluster_->initialize([this, cluster_name] {

      ENVOY_LOG(debug, "warming cluster {} complete", cluster_name);

      auto state_changed_cluster_entry = warming_clusters_.find(cluster_name);

      state_changed_cluster_entry->second->cluster_->info()->configUpdateStats().warming_state_.set(

          0);

      return onClusterInit(*state_changed_cluster_entry->second);

    });

  }

  return true;

}

void ClusterManagerImpl::clusterWarmingToActive(const std::string& cluster_name) {

  auto warming_it = warming_clusters_.find(cluster_name);

  ASSERT(warming_it != warming_clusters_.end());

  updates_map_.erase(cluster_name);

  active_clusters_[cluster_name] = std::move(warming_it->second);

  warming_clusters_.erase(warming_it);

}

bool ClusterManagerImpl::removeCluster(const std::string& cluster_name, const bool remove_ignored) {

  bool removed = false;

  auto existing_active_cluster = active_clusters_.find(cluster_name);

  if (existing_active_cluster != active_clusters_.end() &&

      existing_active_cluster->second->added_via_api_ &&

      (!existing_active_cluster->second->avoid_cds_removal_ || remove_ignored)) {

    removed = true;

    init_helper_.removeCluster(*existing_active_cluster->second);

    active_clusters_.erase(existing_active_cluster);

    ENVOY_LOG(debug, "removing cluster {}", cluster_name);

    tls_.runOnAllThreads([cluster_name](OptRef<ThreadLocalClusterManagerImpl> cluster_manager) {

      ASSERT(cluster_manager->thread_local_clusters_.contains(cluster_name) ||

             cluster_manager->thread_local_deferred_clusters_.contains(cluster_name));

      ENVOY_LOG(debug, "removing TLS cluster {}", cluster_name);

      for (auto cb_it = cluster_manager->update_callbacks_.begin();

           cb_it != cluster_manager->update_callbacks_.end();) {

        auto curr_cb_it = cb_it;

        ++cb_it;

        (*curr_cb_it)->onClusterRemoval(cluster_name);

      }

      cluster_manager->thread_local_clusters_.erase(cluster_name);

      cluster_manager->thread_local_deferred_clusters_.erase(cluster_name);

      cluster_manager->local_stats_.clusters_inflated_.set(

          cluster_manager->thread_local_clusters_.size());

    });

    cluster_initialization_map_.erase(cluster_name);

  }

  auto existing_warming_cluster = warming_clusters_.find(cluster_name);

  if (existing_warming_cluster != warming_clusters_.end() &&

      existing_warming_cluster->second->added_via_api_ &&

      (!existing_warming_cluster->second->avoid_cds_removal_ || remove_ignored)) {

    removed = true;

    init_helper_.removeCluster(*existing_warming_cluster->second);

    warming_clusters_.erase(existing_warming_cluster);

    ENVOY_LOG(info, "removing warming cluster {}", cluster_name);

  }

  if (removed) {

    cm_stats_.cluster_removed_.inc();

    updateClusterCounts();

    updates_map_.erase(cluster_name);

  }

  return removed;

}

                                ClusterMap& cluster_map, const bool avoid_cds_removal) {

  absl::StatusOr<std::pair<ClusterSharedPtr, ThreadAwareLoadBalancerPtr>>

      new_cluster_pair_or_error =

          factory_.clusterFromProto(cluster, outlier_event_logger_, added_via_api);

  if (!new_cluster_pair_or_error.ok()) {

  auto& new_cluster = new_cluster_pair_or_error->first;

  auto& lb = new_cluster_pair_or_error->second;

  Cluster& cluster_reference = *new_cluster;

  const auto cluster_info = cluster_reference.info();

  if (!added_via_api) {

    if (cluster_map.find(cluster_info->name()) != cluster_map.end()) {

      return absl::InvalidArgumentError(

          fmt::format("cluster manager: duplicate cluster '{}'", cluster_info->name()));

    }

  }

  TypedLoadBalancerFactory& typed_lb_factory = cluster_info->loadBalancerFactory();

  const bool cluster_provided_lb =

      typed_lb_factory.name() == "envoy.load_balancing_policies.cluster_provided";

  if (cluster_provided_lb && lb == nullptr) {

    return absl::InvalidArgumentError(

        fmt::format("cluster manager: cluster provided LB specified but cluster "

                    "'{}' did not provide one. Check cluster documentation.",

                    cluster_info->name()));

  }

  if (!cluster_provided_lb && lb != nullptr) {

    return absl::InvalidArgumentError(

        fmt::format("cluster manager: cluster provided LB not specified but cluster "

                    "'{}' provided one. Check cluster documentation.",

                    cluster_info->name()));

  }

  if (new_cluster->healthChecker() != nullptr) {

    new_cluster->healthChecker()->addHostCheckCompleteCb(

        [this](HostSharedPtr host, HealthTransition changed_state, HealthState) {

          if (changed_state == HealthTransition::Changed &&

              host->healthFlagGet(Host::HealthFlag::FAILED_ACTIVE_HC)) {

            postThreadLocalHealthFailure(host);

          }

        });

  }

  if (new_cluster->outlierDetector() != nullptr) {

    new_cluster->outlierDetector()->addChangedStateCb([this](HostSharedPtr host) {

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

        ENVOY_LOG_EVENT(debug, "outlier_detection_ejection",

                        "host {} in cluster {} was ejected by the outlier detector",

                        host->address()->asStringView(), host->cluster().name());

        postThreadLocalHealthFailure(host);

      }

    });

  }

  ClusterDataPtr result;

  auto cluster_entry_it = cluster_map.find(cluster_info->name());

  if (cluster_entry_it != cluster_map.end()) {

    result = std::exchange(cluster_entry_it->second,

                           std::make_unique<ClusterData>(

                               cluster, cluster_hash, version_info, added_via_api, required_for_ads,

                               std::move(new_cluster), time_source_, avoid_cds_removal));

  } else {

    bool inserted = false;

    std::tie(cluster_entry_it, inserted) = cluster_map.emplace(

        cluster_info->name(),

        std::make_unique<ClusterData>(cluster, cluster_hash, version_info, added_via_api,

                                      required_for_ads, std::move(new_cluster), time_source_,

                                      avoid_cds_removal));

    ASSERT(inserted);

  }

  if (cluster_provided_lb) {

    cluster_entry_it->second->thread_aware_lb_ = std::move(lb);

  } else {

    cluster_entry_it->second->thread_aware_lb_ =

        typed_lb_factory.create(cluster_info->loadBalancerConfig(), *cluster_info,

                                cluster_reference.prioritySet(), runtime_, random_, time_source_);

  }

  updateClusterCounts();

  return result;

}

void ClusterManagerImpl::updateClusterCounts() {

  const bool all_clusters_initialized =

      init_helper_.state() == ClusterManagerInitHelper::State::AllClustersInitialized;

  if (all_clusters_initialized && xds_manager_.adsMux()) {

    const auto type_url = Config::getTypeUrl<envoy::config::cluster::v3::Cluster>();

    if (resume_cds_ == nullptr && !warming_clusters_.empty()) {

      resume_cds_ = xds_manager_.pause(type_url);

    } else if (warming_clusters_.empty()) {

      resume_cds_.reset();

    }

  }

  cm_stats_.active_clusters_.set(active_clusters_.size());

  cm_stats_.warming_clusters_.set(warming_clusters_.size());

}

ThreadLocalCluster* ClusterManagerImpl::getThreadLocalCluster(absl::string_view cluster) {

  ThreadLocalClusterManagerImpl& cluster_manager = *tls_;

  auto entry = cluster_manager.thread_local_clusters_.find(cluster);

  if (entry != cluster_manager.thread_local_clusters_.end()) {

    return entry->second.get();

  } else {

    return cluster_manager.initializeClusterInlineIfExists(cluster);

  }

}

    std::function<ConnectionPool::Instance*()> pick_preconnect_pool) {

  auto peekahead_ratio = cluster_entry.info()->peekaheadRatio();

  if (peekahead_ratio <= 1.0) {

    return;

  }

  for (int i = 0; i < 3; ++i) {

    if (!ConnectionPool::ConnPoolImplBase::shouldConnect(

            state.pending_streams_, state.active_streams_,

            state.connecting_and_connected_stream_capacity_, peekahead_ratio, true)) {

      return;

    }

    ConnectionPool::Instance* preconnect_pool = pick_preconnect_pool();

    if (!preconnect_pool || !preconnect_pool->maybePreconnect(peekahead_ratio)) {

      return;

    }

  }

}

    LoadBalancerContext* context) {

  auto pool = httpConnPoolImpl(host, priority, protocol, context);

  if (pool == nullptr) {

    return absl::nullopt;

  }

  HttpPoolData data(

      [this, priority, protocol, context]() -> void {

        maybePreconnect(

            *this, parent_.cluster_manager_state_, [this, &priority, &protocol, &context]() {

              HostConstSharedPtr peek_host = peekAnotherHost(context);

              return peek_host ? httpConnPoolImpl(peek_host, priority, protocol, context) : nullptr;

            });

      },

      pool);

  return data;

}

    HostConstSharedPtr host, ResourcePriority priority, LoadBalancerContext* context) {

  if (!host) {

    return absl::nullopt;

  }

  auto pool = tcpConnPoolImpl(host, priority, context);

  if (pool == nullptr) {

  TcpPoolData data(

      [this, priority, context]() -> void {

        maybePreconnect(*this, parent_.cluster_manager_state_, [this, &priority, &context]() {

          HostConstSharedPtr peek_host = peekAnotherHost(context);

          return peek_host ? tcpConnPoolImpl(peek_host, priority, context) : nullptr;

        });

      },

      pool);

  return data;

}

    ResourcePriority priority, LoadBalancerContext* context) {

  HostConstSharedPtr host = LoadBalancer::onlyAllowSynchronousHostSelection(chooseHost(context));

  if (!host) {

    return absl::nullopt;

  }

  return tcpConnPool(host, priority, context);

}

                                          DrainConnectionsHostPredicate predicate) {

  ENVOY_LOG_EVENT(debug, "drain_connections_call", "drainConnections called for cluster {}",

                  cluster);

  tls_.runOnAllThreads([cluster, predicate](OptRef<ThreadLocalClusterManagerImpl> cluster_manager) {

    auto cluster_entry = cluster_manager->thread_local_clusters_.find(cluster);

    if (cluster_entry != cluster_manager->thread_local_clusters_.end()) {

      cluster_entry->second->drainConnPools(

          predicate, ConnectionPool::DrainBehavior::DrainExistingConnections);

    }

  });

}

                                          ConnectionPool::DrainBehavior drain_behavior) {

  ENVOY_LOG_EVENT(debug, "drain_connections_call_for_all_clusters",

                  "drainConnections called for all clusters");

  tls_.runOnAllThreads(

      [predicate, drain_behavior](OptRef<ThreadLocalClusterManagerImpl> cluster_manager) {

        for (const auto& cluster_entry : cluster_manager->thread_local_clusters_) {

          cluster_entry.second->drainConnPools(predicate, drain_behavior);

        }

      });

}

absl::Status ClusterManagerImpl::checkActiveStaticCluster(const std::string& cluster) {

  const auto& it = active_clusters_.find(cluster);

  if (it == active_clusters_.end()) {

    return absl::InvalidArgumentError(fmt::format("Unknown gRPC client cluster '{}'", cluster));

  }

  if (it->second->added_via_api_) {

    return absl::InvalidArgumentError(

        fmt::format("gRPC client cluster '{}' is not static", cluster));

  }

  return absl::OkStatus();

}

                                                    const HostVector& hosts_removed) {

  tls_.runOnAllThreads([name = cluster.info()->name(),

                        hosts_removed](OptRef<ThreadLocalClusterManagerImpl> cluster_manager) {

    cluster_manager->removeHosts(name, hosts_removed);

  });

}

    const ClusterInfoConstSharedPtr& info) const {

  if (!deferred_cluster_creation_) {

    return false;

  }

  if (auto custom_cluster_type = info->clusterType(); custom_cluster_type.has_value()) {

    static const std::array<std::string, 4> supported_well_known_cluster_types = {

        "envoy.clusters.aggregate", "envoy.cluster.eds", "envoy.clusters.redis",

        "envoy.cluster.static"};

    if (std::find(supported_well_known_cluster_types.begin(),

                  supported_well_known_cluster_types.end(),

                  custom_cluster_type->name()) == supported_well_known_cluster_types.end()) {

  } else {

    static constexpr std::array<envoy::config::cluster::v3::Cluster::DiscoveryType, 2>

        supported_cluster_types = {envoy::config::cluster::v3::Cluster::EDS,

                                   envoy::config::cluster::v3::Cluster::STATIC};

    if (std::find(supported_cluster_types.begin(), supported_cluster_types.end(), info->type()) ==

        supported_cluster_types.end()) {

      return false;

    }

  }

  return true;

}

                                                      ThreadLocalClusterUpdateParams&& params) {

  bool add_or_update_cluster = false;

  if (!cm_cluster.addedOrUpdated()) {

    add_or_update_cluster = true;

    cm_cluster.setAddedOrUpdated();

  }

  LoadBalancerFactorySharedPtr load_balancer_factory;

  if (add_or_update_cluster) {

    load_balancer_factory = cm_cluster.loadBalancerFactory();

  }

  for (auto& per_priority : params.per_priority_update_params_) {

    const auto& host_set =

        cm_cluster.cluster().prioritySet().hostSetsPerPriority()[per_priority.priority_];

    per_priority.update_hosts_params_ = HostSetImpl::updateHostsParams(*host_set);

    per_priority.locality_weights_ = host_set->localityWeights();

    per_priority.weighted_priority_health_ = host_set->weightedPriorityHealth();

    per_priority.overprovisioning_factor_ = host_set->overprovisioningFactor();

  }

  HostMapConstSharedPtr host_map = cm_cluster.cluster().prioritySet().crossPriorityHostMap();

  pending_cluster_creations_.erase(cm_cluster.cluster().info()->name());

  const UnitFloat drop_overload = cm_cluster.cluster().dropOverload();

  const std::string drop_category = cm_cluster.cluster().dropCategory();

  ClusterInitializationObjectConstSharedPtr cluster_initialization_object =

      addOrUpdateClusterInitializationObjectIfSupported(params, cm_cluster.cluster().info(),

                                                        load_balancer_factory, host_map,

                                                        drop_overload, drop_category);

  tls_.runOnAllThreads([info = cm_cluster.cluster().info(), params = std::move(params),

                        add_or_update_cluster, load_balancer_factory, map = std::move(host_map),

                        cluster_initialization_object = std::move(cluster_initialization_object),

                        drop_overload, drop_category = std::move(drop_category)](

                           OptRef<ThreadLocalClusterManagerImpl> cluster_manager) {

    ASSERT(cluster_manager.has_value(),

           "Expected the ThreadLocalClusterManager to be set during ClusterManagerImpl creation.");

    if (const bool defer_unused_clusters =

            cluster_initialization_object != nullptr &&

            !cluster_manager->thread_local_clusters_.contains(info->name()) &&

            !Envoy::Thread::MainThread::isMainThread();

        defer_unused_clusters) {

      ENVOY_LOG(debug, "Deferring add or update for TLS cluster {}", info->name());

      cluster_manager->thread_local_deferred_clusters_[info->name()] =

          cluster_initialization_object;

      ThreadLocalClusterCommand command = [&cluster_manager,

                                           cluster_name = info->name()]() -> ThreadLocalCluster& {

        auto existing_cluster_entry = cluster_manager->thread_local_clusters_.find(cluster_name);

        if (existing_cluster_entry != cluster_manager->thread_local_clusters_.end()) {

        auto* cluster_entry = cluster_manager->initializeClusterInlineIfExists(cluster_name);

        ASSERT(cluster_entry != nullptr, "Deferred clusters initiailization should not fail.");

        return *cluster_entry;

      };

      for (auto cb_it = cluster_manager->update_callbacks_.begin();

           cb_it != cluster_manager->update_callbacks_.end();) {

        auto curr_cb_it = cb_it;

        ++cb_it;

        (*curr_cb_it)->onClusterAddOrUpdate(info->name(), command);

      }

    } else {

      ThreadLocalClusterManagerImpl::ClusterEntry* new_cluster = nullptr;

      if (add_or_update_cluster) {

        if (cluster_manager->thread_local_clusters_.contains(info->name())) {

          ENVOY_LOG(debug, "updating TLS cluster {}", info->name());

        } else {

          ENVOY_LOG(debug, "adding TLS cluster {}", info->name());

        }

        new_cluster = new ThreadLocalClusterManagerImpl::ClusterEntry(*cluster_manager, info,

                                                                      load_balancer_factory);

        cluster_manager->thread_local_clusters_[info->name()].reset(new_cluster);

        cluster_manager->local_stats_.clusters_inflated_.set(

            cluster_manager->thread_local_clusters_.size());

      }

      if (cluster_manager->thread_local_clusters_[info->name()]) {

        cluster_manager->thread_local_clusters_[info->name()]->setDropOverload(drop_overload);

        cluster_manager->thread_local_clusters_[info->name()]->setDropCategory(drop_category);

      }

      for (const auto& per_priority : params.per_priority_update_params_) {

        cluster_manager->updateClusterMembership(

            info->name(), per_priority.priority_, per_priority.update_hosts_params_,

            per_priority.locality_weights_, per_priority.hosts_added_, per_priority.hosts_removed_,

            per_priority.weighted_priority_health_, per_priority.overprovisioning_factor_, map);

      }

      if (new_cluster != nullptr) {

        ThreadLocalClusterCommand command = [&new_cluster]() -> ThreadLocalCluster& {

          return *new_cluster;

        };

        for (auto cb_it = cluster_manager->update_callbacks_.begin();

             cb_it != cluster_manager->update_callbacks_.end();) {

          auto curr_cb_it = cb_it;

          ++cb_it;

          (*curr_cb_it)->onClusterAddOrUpdate(info->name(), command);

        }

      }

    }

  });

  if (cm_cluster.requiredForAds() && !ads_mux_initialized_) {

    xds_manager_.adsMux()->start();

    ads_mux_initialized_ = true;

  }

}

    UnitFloat drop_overload, absl::string_view drop_category) {

  if (!deferralIsSupportedForCluster(cluster_info)) {

    return nullptr;

  }

  const std::string& cluster_name = cluster_info->name();

  auto entry = cluster_initialization_map_.find(cluster_name);

  const bool should_merge_with_prior_cluster =

      entry != cluster_initialization_map_.end() && entry->second->cluster_info_ == cluster_info;

  if (should_merge_with_prior_cluster) {

    auto new_initialization_object = std::make_shared<ClusterInitializationObject>(

        entry->second->per_priority_state_, params, std::move(cluster_info),

        load_balancer_factory == nullptr ? entry->second->load_balancer_factory_

                                         : load_balancer_factory,

        map, drop_overload, drop_category);

    cluster_initialization_map_[cluster_name] = new_initialization_object;

    return new_initialization_object;

  } else {