#pragma once

#include <array>

#include <cstdint>

#include <functional>

#include <list>

#include <map>

#include <memory>

#include <string>

#include <vector>

#include "envoy/api/api.h"

#include "envoy/common/callback.h"

#include "envoy/common/random_generator.h"

#include "envoy/config/bootstrap/v3/bootstrap.pb.h"

#include "envoy/config/cluster/v3/cluster.pb.h"

#include "envoy/config/core/v3/address.pb.h"

#include "envoy/config/core/v3/config_source.pb.h"

#include "envoy/config/xds_resources_delegate.h"

#include "envoy/http/codes.h"

#include "envoy/local_info/local_info.h"

#include "envoy/router/context.h"

#include "envoy/runtime/runtime.h"

#include "envoy/secret/secret_manager.h"

#include "envoy/server/instance.h"

#include "envoy/ssl/context_manager.h"

#include "envoy/stats/scope.h"

#include "envoy/tcp/async_tcp_client.h"

#include "envoy/thread_local/thread_local.h"

#include "envoy/upstream/cluster_manager.h"

#include "envoy/upstream/load_stats_reporter.h"

#include "source/common/common/cleanup.h"

#include "source/common/common/thread.h"

#include "source/common/http/async_client_impl.h"

#include "source/common/http/http_server_properties_cache_impl.h"

#include "source/common/http/http_server_properties_cache_manager_impl.h"

#include "source/common/quic/envoy_quic_network_observer_registry_factory.h"

#include "source/common/quic/quic_stat_names.h"

#include "source/common/tcp/async_tcp_client_impl.h"

#include "source/common/upstream/cluster_discovery_manager.h"

#include "source/common/upstream/host_utility.h"

#include "source/common/upstream/priority_conn_pool_map.h"

#include "source/common/upstream/upstream_impl.h"

namespace Envoy {

namespace Upstream {

/**

 * Production implementation of ClusterManagerFactory.

 */

class ProdClusterManagerFactory : public ClusterManagerFactory {

public:

  using LazyCreateDnsResolver = std::function<Network::DnsResolverSharedPtr()>;

  ProdClusterManagerFactory(Server::Configuration::ServerFactoryContext& context,

                            LazyCreateDnsResolver dns_resolver_fn,

                            Quic::QuicStatNames& quic_stat_names)

  // Upstream::ClusterManagerFactory

  absl::StatusOr<ClusterManagerPtr>

  clusterManagerFromProto(const envoy::config::bootstrap::v3::Bootstrap& bootstrap) override;

  Http::ConnectionPool::InstancePtr allocateConnPool(

      Event::Dispatcher& dispatcher, HostConstSharedPtr host, ResourcePriority priority,

      std::vector<Http::Protocol>& protocol,

      const absl::optional<envoy::config::core::v3::AlternateProtocolsCacheOptions>&

          alternate_protocol_options,

      const Network::ConnectionSocket::OptionsSharedPtr& options,

      const Network::TransportSocketOptionsConstSharedPtr& transport_socket_options,

      TimeSource& time_source, ClusterConnectivityState& state,

      Http::PersistentQuicInfoPtr& quic_info,

      OptRef<Quic::EnvoyQuicNetworkObserverRegistry> network_observer_registry) override;

  Tcp::ConnectionPool::InstancePtr

  allocateTcpConnPool(Event::Dispatcher& dispatcher, HostConstSharedPtr host,

                      ResourcePriority priority,

                      const Network::ConnectionSocket::OptionsSharedPtr& options,

                      Network::TransportSocketOptionsConstSharedPtr transport_socket_options,

                      ClusterConnectivityState& state,

                      absl::optional<std::chrono::milliseconds> tcp_pool_idle_timeout) override;

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

  clusterFromProto(const envoy::config::cluster::v3::Cluster& cluster,

                   Outlier::EventLoggerSharedPtr outlier_event_logger, bool added_via_api) override;

  absl::StatusOr<CdsApiPtr> createCds(const envoy::config::core::v3::ConfigSource& cds_config,

                                      const xds::core::v3::ResourceLocator* cds_resources_locator,

                                      ClusterManager& cm, bool support_multi_ads_sources) override;

protected:

  Server::Configuration::ServerFactoryContext& context_;

  Stats::Store& stats_;

  LazyCreateDnsResolver dns_resolver_fn_;

  Quic::QuicStatNames& quic_stat_names_;

  Http::HttpServerPropertiesCacheManager& alternate_protocols_cache_manager_;

};

// For friend declaration in ClusterManagerInitHelper.

class ClusterManagerImpl;

/**

 * Wrapper for a cluster owned by the cluster manager. Used by both the cluster manager and the

 * cluster manager init helper which needs to pass clusters back to the cluster manager.

 */

class ClusterManagerCluster {

public:

  // Return the underlying cluster.

  virtual Cluster& cluster() PURE;

  // Return a new load balancer factory if the cluster has one.

  virtual LoadBalancerFactorySharedPtr loadBalancerFactory() PURE;

  // Return true if a cluster has already been added or updated.

  virtual bool addedOrUpdated() PURE;

  // Set when a cluster has been added or updated. This is only called a single time for a cluster.

  virtual void setAddedOrUpdated() PURE;

  // Return true if the cluster must be ready-for-use before ADS (Aggregated Discovery Service) can

  // be initialized; will only occur if ADS is configured to use the cluster via EnvoyGrpc.

  virtual bool requiredForAds() const PURE;

};

/**

 * This is a helper class used during cluster management initialization. Dealing with primary

 * clusters, secondary clusters, and CDS, is quite complicated, so this makes it easier to test.

 */

class ClusterManagerInitHelper : Logger::Loggable<Logger::Id::upstream> {

public:

  /**

   * @param per_cluster_init_callback supplies the callback to call when a cluster has itself

   *        initialized. The cluster manager can use this for post-init processing.

   */

  ClusterManagerInitHelper(

      Config::XdsManager& xds_manager,

      const std::function<absl::Status(ClusterManagerCluster&)>& per_cluster_init_callback)

  enum class State {

    // Initial state. During this state all static clusters are loaded. Any primary clusters

    // immediately begin initialization.

    Loading,

    // In this state cluster manager waits for all primary clusters to finish initialization.

    // This state may immediately transition to the next state iff all clusters are STATIC and

    // without health checks enabled or health checks have failed immediately, since their

    // initialization completes immediately.

    WaitingForPrimaryInitializationToComplete,

    // During this state cluster manager waits to start initializing secondary clusters. In this

    // state all primary clusters have completed initialization. Initialization of the

    // secondary clusters is started by the `initializeSecondaryClusters` method.

    WaitingToStartSecondaryInitialization,

    // In this state cluster manager waits for all secondary clusters (if configured) to finish

    // initialization. Then, if CDS is configured, this state tracks waiting for the first CDS

    // response to populate dynamically configured clusters.

    WaitingToStartCdsInitialization,

    // During this state, all CDS populated clusters are undergoing either phase 1 or phase 2

    // initialization.

    CdsInitialized,

    // All clusters are fully initialized.

    AllClustersInitialized

  };

  void addCluster(ClusterManagerCluster& cluster);

  void onStaticLoadComplete();

  void removeCluster(ClusterManagerCluster& cluster);

  void setCds(CdsApi* cds);

  void setPrimaryClustersInitializedCb(ClusterManager::PrimaryClustersReadyCallback callback);

  void setInitializedCb(ClusterManager::InitializationCompleteCallback callback);

  void startInitializingSecondaryClusters();

private:

  // To enable invariant assertions on the cluster lists.

  friend ClusterManagerImpl;

  void initializeSecondaryClusters();

  void maybeFinishInitialize();

  absl::Status onClusterInit(ClusterManagerCluster& cluster);

  Config::XdsManager& xds_manager_;

  std::function<absl::Status(ClusterManagerCluster& cluster)> per_cluster_init_callback_;

  CdsApi* cds_{};

  ClusterManager::PrimaryClustersReadyCallback primary_clusters_initialized_callback_;

  ClusterManager::InitializationCompleteCallback initialized_callback_;

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

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

  State state_{State::Loading};

  bool started_secondary_initialize_{};

};

/**

 * All cluster manager stats. @see stats_macros.h

 */

#define ALL_CLUSTER_MANAGER_STATS(COUNTER, GAUGE)                                                  \

/**

 * Struct definition for all cluster manager stats. @see stats_macros.h

 */

struct ClusterManagerStats {

  ALL_CLUSTER_MANAGER_STATS(GENERATE_COUNTER_STRUCT, GENERATE_GAUGE_STRUCT)

};

/**

 * All thread local cluster manager stats. @see stats_macros.h

 */

/**

 * Struct definition for all cluster manager stats. @see stats_macros.h

 */

struct ThreadLocalClusterManagerStats {

  ALL_THREAD_LOCAL_CLUSTER_MANAGER_STATS(GENERATE_GAUGE_STRUCT)

};

/**

 * Implementation of ClusterManager that reads from a proto configuration, maintains a central

 * cluster list, as well as thread local caches of each cluster and associated connection pools.

 */

class ClusterManagerImpl : public ClusterManager,

                           public MissingClusterNotifier,

                           Logger::Loggable<Logger::Id::upstream> {

public:

  absl::Status initialize(const envoy::config::bootstrap::v3::Bootstrap& bootstrap) override;

  // Upstream::ClusterManager

  absl::StatusOr<bool> addOrUpdateCluster(const envoy::config::cluster::v3::Cluster& cluster,

                                          const std::string& version_info,

                                          const bool avoid_cds_removal = false) override;

    // The number of clusters that were added via API must be at most the number

    // of active clusters + number of warming clusters.

  ThreadLocalCluster* getThreadLocalCluster(absl::string_view cluster) override;

  bool removeCluster(const std::string& cluster, const bool remove_ignored = false) override;

    // Make sure we destroy all potential outgoing connections before this returns.

  ClusterUpdateCallbacksHandlePtr

  addThreadLocalClusterUpdateCallbacks(ClusterUpdateCallbacks&) override;

  absl::StatusOr<OdCdsApiHandlePtr>

  allocateOdCdsApi(OdCdsCreationFunction creation_function,

                   const envoy::config::core::v3::ConfigSource& odcds_config,

                   OptRef<xds::core::v3::ResourceLocator> odcds_resources_locator,

                   ProtobufMessage::ValidationVisitor& validation_visitor) override;

  // TODO(adisuissa): remove this method, and switch all the callers to invoke

  // it directly via the XdsManger.

  absl::Status

  initializeSecondaryClusters(const envoy::config::bootstrap::v3::Bootstrap& bootstrap) override;

  void drainConnections(const std::string& cluster,

                        DrainConnectionsHostPredicate predicate) override;

  void drainConnections(DrainConnectionsHostPredicate predicate,

                        ConnectionPool::DrainBehavior drain_behavior) override;

  absl::Status checkActiveStaticCluster(const std::string& cluster) override;

  // Upstream::MissingClusterNotifier

  void notifyMissingCluster(absl::string_view name) override;

  /*

   * Return shared_ptr for common_lb_config which is stored in an ObjectSharedPool

   *

   * @param common_lb_config The config field to be stored in ObjectSharedPool

   * @return shared_ptr to the CommonLbConfig in ObjectSharedPool

   */

  std::shared_ptr<const envoy::config::cluster::v3::Cluster::CommonLbConfig> getCommonLbConfigPtr(

  Config::EdsResourcesCacheOptRef edsResourcesCache() override;

  void

  createNetworkObserverRegistries(Quic::EnvoyQuicNetworkObserverRegistryFactory& factory) override;

protected:

  // ClusterManagerImpl's constructor should not be invoked directly; create instances from the

  // clusterManagerFromProto() static method. The init() method must be called after construction.

  ClusterManagerImpl(const envoy::config::bootstrap::v3::Bootstrap& bootstrap,

                     ClusterManagerFactory& factory,

                     Server::Configuration::ServerFactoryContext& context,

                     absl::Status& creation_status);

  virtual void postThreadLocalRemoveHosts(const Cluster& cluster, const HostVector& hosts_removed);

  // Parameters for calling postThreadLocalClusterUpdate()

  struct ThreadLocalClusterUpdateParams {

    struct PerPriority {

      PerPriority(uint32_t priority, const HostVector& hosts_added, const HostVector& hosts_removed)

      // TODO(kbaichoo): make the hosts_added_ vector const and have the

      // cluster initialization object have a stripped down version of this

      // struct.

      HostVector hosts_added_;

      const HostVector hosts_removed_;

      PrioritySet::UpdateHostsParams update_hosts_params_;

      LocalityWeightsConstSharedPtr locality_weights_;

      // Keep small members (bools and enums) at the end of class, to reduce alignment overhead.

      const uint32_t priority_;

      bool weighted_priority_health_;

      uint32_t overprovisioning_factor_;

    };

    ThreadLocalClusterUpdateParams(uint32_t priority, const HostVector& hosts_added,

                                   const HostVector& hosts_removed)

    std::vector<PerPriority> per_priority_update_params_;

  };

  /**

   * A cluster initialization object (CIO) encapsulates the relevant information

   * to create a cluster inline when there is traffic to it. We can thus use the

   * CIO to deferred instantiating clusters on workers until they are used.

   */

  struct ClusterInitializationObject {

    ClusterInitializationObject(const ThreadLocalClusterUpdateParams& params,

                                ClusterInfoConstSharedPtr cluster_info,

                                LoadBalancerFactorySharedPtr load_balancer_factory,

                                HostMapConstSharedPtr map, UnitFloat drop_overload,

                                absl::string_view drop_category);

    ClusterInitializationObject(

        const absl::flat_hash_map<int, ThreadLocalClusterUpdateParams::PerPriority>&

            per_priority_state,

        const ThreadLocalClusterUpdateParams& update_params, ClusterInfoConstSharedPtr cluster_info,

        LoadBalancerFactorySharedPtr load_balancer_factory, HostMapConstSharedPtr map,

        UnitFloat drop_overload, absl::string_view drop_category);

    absl::flat_hash_map<int, ThreadLocalClusterUpdateParams::PerPriority> per_priority_state_;

    const ClusterInfoConstSharedPtr cluster_info_;

    const LoadBalancerFactorySharedPtr load_balancer_factory_;

    const HostMapConstSharedPtr cross_priority_host_map_;

    UnitFloat drop_overload_{0};

    const std::string drop_category_;

  };

  using ClusterInitializationObjectConstSharedPtr =

      std::shared_ptr<const ClusterInitializationObject>;

  using ClusterInitializationMap =

      absl::flat_hash_map<std::string, ClusterInitializationObjectConstSharedPtr>;

  /**

   * An implementation of an on-demand CDS handle. It forwards the discovery request to the cluster

   * manager that created the handle.

   *

   * It's a protected type, so unit tests can use it.

   */

  class OdCdsApiHandleImpl : public OdCdsApiHandle {

  public:

    OdCdsApiHandleImpl(ClusterManagerImpl& parent, uint64_t config_source_key)

    ClusterDiscoveryCallbackHandlePtr

    requestOnDemandClusterDiscovery(absl::string_view name, ClusterDiscoveryCallbackPtr callback,

  private:

    ClusterManagerImpl& parent_;

    uint64_t config_source_key_;

  };

  virtual void postThreadLocalClusterUpdate(ClusterManagerCluster& cm_cluster,

                                            ThreadLocalClusterUpdateParams&& params);

  /**

   * Notifies cluster discovery managers in each worker thread that the discovery process for the

   * cluster with a passed name has timed out.

   *

   * It's protected, so the tests can use it.

   */

  void notifyExpiredDiscovery(absl::string_view name);

  /**

   * Creates a new discovery manager in current thread and swaps it with the one in thread local

   * cluster manager. This could be used to simulate requesting a cluster from a different

   * thread. Used for tests only.

   *

   * Protected, so tests can use it.

   *

   * @return the previous cluster discovery manager.

   */

  ClusterDiscoveryManager createAndSwapClusterDiscoveryManager(std::string thread_name);

private:

  // To enable access to the protected constructor.

  friend ProdClusterManagerFactory;

  /**

   * Thread local cached cluster data. Each thread local cluster gets updates from the parent

   * central dynamic cluster (if applicable). It maintains load balancer state and any created

   * connection pools.

   */

  struct ThreadLocalClusterManagerImpl : public ThreadLocal::ThreadLocalObject,

                                         public ClusterLifecycleCallbackHandler {

    struct ConnPoolsContainer {

      ConnPoolsContainer(Event::Dispatcher& dispatcher, const HostConstSharedPtr& host)

      using ConnPools = PriorityConnPoolMap<std::vector<uint8_t>, Http::ConnectionPool::Instance>;

      // Destroyed after pools.

      const HostHandlePtr host_handle_;

      // This is a shared_ptr so we can keep it alive while cleaning up.

      std::shared_ptr<ConnPools> pools_;

      // Protect from deletion while iterating through pools_. See comments and usage

      // in `ClusterManagerImpl::ThreadLocalClusterManagerImpl::drainConnPools()`.

      bool do_not_delete_{false};

    };

    struct TcpConnPoolsContainer {

      using ConnPools = std::map<std::vector<uint8_t>, Tcp::ConnectionPool::InstancePtr>;

      // Destroyed after pools.

      const HostHandlePtr host_handle_;

      ConnPools pools_;

    };

    // Holds an unowned reference to a connection, and watches for Closed events. If the connection

    // is closed, this container removes itself from the container that owns it.

    struct TcpConnContainer : public Network::ConnectionCallbacks, public Event::DeferredDeletable {

    public:

      TcpConnContainer(ThreadLocalClusterManagerImpl& parent, const HostConstSharedPtr& host,

                       Network::ClientConnection& connection)

      // Network::ConnectionCallbacks

      ThreadLocalClusterManagerImpl& parent_;

      HostConstSharedPtr host_;

      Network::ClientConnection& connection_;

    };

    struct TcpConnectionsMap {

      // Destroyed after pools.

      const HostHandlePtr host_handle_;

      absl::node_hash_map<Network::ClientConnection*, std::unique_ptr<TcpConnContainer>>

          connections_;

    };

    class ClusterEntry : public ThreadLocalCluster {

    public:

      ClusterEntry(ThreadLocalClusterManagerImpl& parent, ClusterInfoConstSharedPtr cluster,

                   const LoadBalancerFactorySharedPtr& lb_factory);

      ~ClusterEntry() override;

      // Upstream::ThreadLocalCluster

      HostSelectionResponse chooseHost(LoadBalancerContext* context) override;

      absl::optional<HttpPoolData> httpConnPool(HostConstSharedPtr host, ResourcePriority priority,

                                                absl::optional<Http::Protocol> downstream_protocol,

                                                LoadBalancerContext* context) override;

      absl::optional<TcpPoolData> tcpConnPool(HostConstSharedPtr host, ResourcePriority priority,

                                              LoadBalancerContext* context) override;

      absl::optional<TcpPoolData> tcpConnPool(ResourcePriority priority,

                                              LoadBalancerContext* context) override;

      Host::CreateConnectionData tcpConn(LoadBalancerContext* context) override;

      Http::AsyncClient& httpAsyncClient() override;

      Tcp::AsyncTcpClientPtr

      tcpAsyncClient(LoadBalancerContext* context,

                     Tcp::AsyncTcpClientOptionsConstSharedPtr options) override;

      // Updates the hosts in the priority set.

      void updateHosts(const std::string& name, uint32_t priority,

                       PrioritySet::UpdateHostsParams&& update_hosts_params,

                       LocalityWeightsConstSharedPtr locality_weights,

                       const HostVector& hosts_added, const HostVector& hosts_removed,

                       absl::optional<bool> weighted_priority_health,

                       absl::optional<uint32_t> overprovisioning_factor,

                       HostMapConstSharedPtr cross_priority_host_map);

      // Drains any connection pools associated with the removed hosts. All connections will be

      // closed gracefully and no new connections will be created.

      void drainConnPools(const HostVector& hosts_removed);

      // Drains any connection pools associated with the all hosts. All connections will be

      // closed gracefully and no new connections will be created.

      void drainConnPools();

      // Drain any connection pools associated with the hosts filtered by the predicate.

      void drainConnPools(DrainConnectionsHostPredicate predicate,

                          ConnectionPool::DrainBehavior behavior);

    private:

      Http::ConnectionPool::Instance*

      httpConnPoolImpl(HostConstSharedPtr host, ResourcePriority priority,

                       absl::optional<Http::Protocol> downstream_protocol,

                       LoadBalancerContext* context);

      Tcp::ConnectionPool::Instance* tcpConnPoolImpl(HostConstSharedPtr host,

                                                     ResourcePriority priority,

                                                     LoadBalancerContext* context);

      HostConstSharedPtr peekAnotherHost(LoadBalancerContext* context);

      ThreadLocalClusterManagerImpl& parent_;

      PrioritySetImpl priority_set_;

      UnitFloat drop_overload_{0};

      std::string drop_category_;

      // Don't change the order of cluster_info_ and lb_factory_/lb_ as the the lb_factory_/lb_

      // may keep a reference to the cluster_info_.

      ClusterInfoConstSharedPtr cluster_info_;

      // Factory to create active LB.

      LoadBalancerFactorySharedPtr lb_factory_;

      // Current active LB.

      LoadBalancerPtr lb_;

      Http::AsyncClientPtr lazy_http_async_client_;

      // Stores QUICHE specific objects which live through out the life time of the cluster and can

      // be shared across its hosts.

      Http::PersistentQuicInfoPtr quic_info_;

      // Expected override host statues. Every bit in the HostStatusSet represent an enum value

      // of envoy::config::core::v3::HealthStatus. The specific correspondence is shown below:

      //

      // * 0b000001: envoy::config::core::v3::HealthStatus::UNKNOWN

      // * 0b000010: envoy::config::core::v3::HealthStatus::HEALTHY

      // * 0b000100: envoy::config::core::v3::HealthStatus::UNHEALTHY

      // * 0b001000: envoy::config::core::v3::HealthStatus::DRAINING

      // * 0b010000: envoy::config::core::v3::HealthStatus::TIMEOUT

      // * 0b100000: envoy::config::core::v3::HealthStatus::DEGRADED

      //

      // If multiple bit fields are set, it is acceptable as long as the status of override host is

      // in any of these statuses.

      const HostUtility::HostStatusSet override_host_statuses_;

    };

    using ClusterEntryPtr = std::unique_ptr<ClusterEntry>;

    struct LocalClusterParams {

      LoadBalancerFactorySharedPtr load_balancer_factory_;

      ClusterInfoConstSharedPtr info_;

    };

    ThreadLocalClusterManagerImpl(ClusterManagerImpl& parent, Event::Dispatcher& dispatcher,

                                  const absl::optional<LocalClusterParams>& local_cluster_params);

    ~ThreadLocalClusterManagerImpl() override;

    // Drain or close connections of host. If no drain behavior is provided then closing will

    // be immediate.

    void drainOrCloseConnPools(const HostSharedPtr& host,

                               absl::optional<ConnectionPool::DrainBehavior> drain_behavior);

    void httpConnPoolIsIdle(HostConstSharedPtr host, ResourcePriority priority,

                            const std::vector<uint8_t>& hash_key);

    void tcpConnPoolIsIdle(HostConstSharedPtr host, const std::vector<uint8_t>& hash_key);

    void removeTcpConn(const HostConstSharedPtr& host, Network::ClientConnection& connection);

    void removeHosts(const std::string& name, const HostVector& hosts_removed);

    void updateClusterMembership(const std::string& name, uint32_t priority,

                                 PrioritySet::UpdateHostsParams update_hosts_params,

                                 LocalityWeightsConstSharedPtr locality_weights,

                                 const HostVector& hosts_added, const HostVector& hosts_removed,

                                 bool weighted_priority_health, uint64_t overprovisioning_factor,

                                 HostMapConstSharedPtr cross_priority_host_map);

    void onHostHealthFailure(const HostSharedPtr& host);

    ConnPoolsContainer* getHttpConnPoolsContainer(const HostConstSharedPtr& host,

                                                  bool allocate = false);

    // Upstream::ClusterLifecycleCallbackHandler

    ClusterUpdateCallbacksHandlePtr addClusterUpdateCallbacks(ClusterUpdateCallbacks& cb) override;

    /**

     * Transparently initialize the given thread local cluster if possible using

     * the Cluster Initialization object.

     *

     * @return The ClusterEntry of the newly initialized cluster or nullptr if there

     * is no cluster deferred cluster with that name.

     */

    ClusterEntry* initializeClusterInlineIfExists(absl::string_view cluster);

#ifdef ENVOY_ENABLE_QUIC

    void createThreadLocalNetworkObserverRegistry(

#endif

    ClusterManagerImpl& parent_;

    Event::Dispatcher& thread_local_dispatcher_;

    // Known clusters will exclusively exist in either `thread_local_clusters_`

    // or `thread_local_deferred_clusters_`.

    absl::flat_hash_map<std::string, ClusterEntryPtr> thread_local_clusters_;

    // Maps from a given cluster name to the CIO for that cluster.

    ClusterInitializationMap thread_local_deferred_clusters_;

    ClusterConnectivityState cluster_manager_state_;

    // These maps are owned by the ThreadLocalClusterManagerImpl instead of the ClusterEntry

    // to prevent lifetime/ownership issues when a cluster is dynamically removed.

    absl::node_hash_map<HostConstSharedPtr, ConnPoolsContainer> host_http_conn_pool_map_;

    absl::node_hash_map<HostConstSharedPtr, TcpConnPoolsContainer> host_tcp_conn_pool_map_;

    absl::node_hash_map<HostConstSharedPtr, TcpConnectionsMap> host_tcp_conn_map_;

    std::list<Envoy::Upstream::ClusterUpdateCallbacks*> update_callbacks_;

    const PrioritySet* local_priority_set_{};

    bool destroying_{};

    ClusterDiscoveryManager cdm_;

    ThreadLocalClusterManagerStats local_stats_;

  private:

    static ThreadLocalClusterManagerStats generateStats(Stats::Scope& scope,

                                                        const std::string& thread_name);

    Quic::EnvoyQuicNetworkObserverRegistryPtr network_observer_registry_;

  };

  struct ClusterData : public ClusterManagerCluster {

    ClusterData(const envoy::config::cluster::v3::Cluster& cluster_config,

                const uint64_t cluster_config_hash, const std::string& version_info,

                bool added_via_api, bool required_for_ads, ClusterSharedPtr&& cluster,

                TimeSource& time_source, const bool avoid_cds_removal = false)

    // ClusterManagerCluster

    const envoy::config::cluster::v3::Cluster cluster_config_;

    const uint64_t config_hash_;

    const std::string version_info_;

    // Don't change the order of cluster_ and thread_aware_lb_ as the thread_aware_lb_ may

    // keep a reference to the cluster_.

    ClusterSharedPtr cluster_;

    // Optional thread aware LB depending on the LB type. Not all clusters have one.

    ThreadAwareLoadBalancerPtr thread_aware_lb_;

    SystemTime last_updated_;

    Common::CallbackHandlePtr member_update_cb_;

    Common::CallbackHandlePtr priority_update_cb_;

    // Keep smaller fields near the end to reduce padding

    const bool added_via_api_ : 1;

    const bool avoid_cds_removal_ : 1;

    bool added_or_updated_ : 1;

    const bool required_for_ads_ : 1;

  };

  struct ClusterUpdateCallbacksHandleImpl : public ClusterUpdateCallbacksHandle,

                                            RaiiListElement<ClusterUpdateCallbacks*> {

    ClusterUpdateCallbacksHandleImpl(ClusterUpdateCallbacks& cb,

                                     std::list<ClusterUpdateCallbacks*>& parent)

  };

  using ClusterDataPtr = std::unique_ptr<ClusterData>;

  // This map is ordered so that config dumping is consistent.

  using ClusterMap = std::map<std::string, ClusterDataPtr>;

  struct PendingUpdates {

    Event::TimerPtr timer_;

    // This is default constructed to the clock's epoch:

    // https://en.cppreference.com/w/cpp/chrono/time_point/time_point

    //

    // Depending on your execution environment this value can be different.

    // When running as host process: This will usually be the computer's boot time, which means that

    // given a not very large `Cluster.CommonLbConfig.update_merge_window`, the first update will

    // trigger immediately (the expected behavior). When running in some sandboxed environment this

    // value can be set to the start time of the sandbox, which means that the delta calculated

    // between now and the start time may fall within the

    // `Cluster.CommonLbConfig.update_merge_window`, with the side effect to delay the first update.

    MonotonicTime last_updated_;

  };

  using PendingUpdatesPtr = std::unique_ptr<PendingUpdates>;

  using PendingUpdatesByPriorityMap = absl::node_hash_map<uint32_t, PendingUpdatesPtr>;

  using PendingUpdatesByPriorityMapPtr = std::unique_ptr<PendingUpdatesByPriorityMap>;

  using ClusterUpdatesMap = absl::node_hash_map<std::string, PendingUpdatesByPriorityMapPtr>;

  /**

   * Holds a reference to an on-demand CDS to keep it alive for the duration of a cluster discovery,

   * and an expiration timer notifying worker threads about discovery timing out.

   */

  struct ClusterCreation {

    OdCdsApiSharedPtr odcds_;

    Event::TimerPtr expiration_timer_;

  };

  using ClusterCreationsMap = absl::flat_hash_map<std::string, ClusterCreation>;

  void applyUpdates(ClusterManagerCluster& cluster, uint32_t priority, PendingUpdates& updates);

  bool scheduleUpdate(ClusterManagerCluster& cluster, uint32_t priority, bool mergeable,

                      const uint64_t timeout);

  ProtobufTypes::MessagePtr dumpClusterConfigs(const Matchers::StringMatcher& name_matcher);

  static ClusterManagerStats generateStats(Stats::Scope& scope);

  /**

   * @return ClusterDataPtr contains the previous cluster in the cluster_map, or

   * nullptr if cluster_map did not contain the same cluster or an error if

   * cluster load fails.

   */

  absl::StatusOr<ClusterDataPtr> loadCluster(const envoy::config::cluster::v3::Cluster& cluster,

                                             const uint64_t cluster_hash,

                                             const std::string& version_info, bool added_via_api,

                                             bool required_for_ads, ClusterMap& cluster_map,

                                             bool avoid_cds_removal = false);

  absl::Status onClusterInit(ClusterManagerCluster& cluster);

  void postThreadLocalHealthFailure(const HostSharedPtr& host);

  void updateClusterCounts();

  void clusterWarmingToActive(const std::string& cluster_name);

  static void maybePreconnect(ThreadLocalClusterManagerImpl::ClusterEntry& cluster_entry,

                              const ClusterConnectivityState& cluster_manager_state,

                              std::function<ConnectionPool::Instance*()> preconnect_pool);

  ClusterDiscoveryCallbackHandlePtr

  requestOnDemandClusterDiscovery(uint64_t subscription_key, std::string name,

                                  ClusterDiscoveryCallbackPtr callback,

                                  std::chrono::milliseconds timeout);

  void notifyClusterDiscoveryStatus(absl::string_view name, ClusterDiscoveryStatus status);

protected:

  ClusterInitializationMap cluster_initialization_map_;

  // OdCDS subscriptions keyed by config source hash. Subscriptions persist for the lifetime

  // of ClusterManagerImpl to avoid complexity around cleanup. A future optimization could

  // add proper lifetime management to close unnecessary connections.

  absl::flat_hash_map<uint64_t, OdCdsApiSharedPtr> odcds_subscriptions_;

private:

  /**

   * Builds the cluster initialization object for this given cluster.

   * @return a ClusterInitializationObjectSharedPtr that can be used to create

   * this cluster or nullptr if deferred cluster creation is off or the cluster

   * type is not supported.

   */

  ClusterInitializationObjectConstSharedPtr addOrUpdateClusterInitializationObjectIfSupported(

      const ThreadLocalClusterUpdateParams& params, ClusterInfoConstSharedPtr cluster_info,

      LoadBalancerFactorySharedPtr load_balancer_factory, HostMapConstSharedPtr map,

      UnitFloat drop_overload, absl::string_view drop_category);

  bool deferralIsSupportedForCluster(const ClusterInfoConstSharedPtr& info) const;

  Server::Configuration::ServerFactoryContext& context_;

  ClusterManagerFactory& factory_;

  Runtime::Loader& runtime_;

  Stats::Store& stats_;

  ThreadLocal::TypedSlot<ThreadLocalClusterManagerImpl> tls_;

  // Contains information about ongoing on-demand cluster discoveries.

  ClusterCreationsMap pending_cluster_creations_;

  Config::XdsManager& xds_manager_;

  Random::RandomGenerator& random_;

  const bool deferred_cluster_creation_;

  absl::optional<envoy::config::core::v3::BindConfig> bind_config_;

  Outlier::EventLoggerSharedPtr outlier_event_logger_;

  const LocalInfo::LocalInfo& local_info_;

  CdsApiPtr cds_api_;

  ClusterManagerStats cm_stats_;

  ClusterManagerInitHelper init_helper_;

  // Temporarily saved resume cds callback from updateClusterCounts invocation.

  Config::ScopedResume resume_cds_;

  LoadStatsReporterPtr load_stats_reporter_;

  // The name of the local cluster of this Envoy instance if defined.

  absl::optional<std::string> local_cluster_name_;

  Grpc::AsyncClientManagerPtr async_client_manager_;

  Server::ConfigTracker::EntryOwnerPtr config_tracker_entry_;

  TimeSource& time_source_;

  ClusterUpdatesMap updates_map_;

  Event::Dispatcher& dispatcher_;

  Http::Context& http_context_;

  Router::Context& router_context_;

  ClusterTrafficStatNames cluster_stat_names_;

  ClusterConfigUpdateStatNames cluster_config_update_stat_names_;

  ClusterLbStatNames cluster_lb_stat_names_;

  ClusterEndpointStatNames cluster_endpoint_stat_names_;

  ClusterLoadReportStatNames cluster_load_report_stat_names_;

  ClusterCircuitBreakersStatNames cluster_circuit_breakers_stat_names_;

  ClusterRequestResponseSizeStatNames cluster_request_response_size_stat_names_;

  ClusterTimeoutBudgetStatNames cluster_timeout_budget_stat_names_;

  std::shared_ptr<SharedPool::ObjectSharedPool<

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

      common_lb_config_pool_;

  ClusterSet primary_clusters_;

  bool initialized_{};

  bool ads_mux_initialized_{};

  std::atomic<bool> shutdown_;

  // Keep all the ClusterMaps at the end, so that they get destroyed first.

  // Clusters may keep references to the cluster manager and in destructor can call

  // cluster manager methods.

  //

  // This might make MSAN unhappy because it thinks that we are accessing uninitialized

  // memory when those methods access fields of the cluster manager class.

  ClusterMap warming_clusters_;

protected:

  ClusterMap active_clusters_;

};

} // namespace Upstream

} // namespace Envoy