Repository: envoyproxy/envoy
File: source/common/upstream/cluster_manager_impl.cc (C++)

Code before:
      if (predicate != nullptr && !predicate(*host)) {
        continue;
      }

      parent_.drainOrCloseConnPools(host, behavior);
    }
  }
}

ClusterUpdateCallbacksHandlePtr
ClusterManagerImpl::addThreadLocalClusterUpdateCallbacks(ClusterUpdateCallbacks& cb) {
  ThreadLocalClusterManagerImpl& cluster_manager = *tls_;
  return cluster_manager.addClusterUpdateCallbacks(cb);
}

absl::StatusOr<OdCdsApiHandlePtr>
ClusterManagerImpl::allocateOdCdsApi(OdCdsCreationFunction creation_function,
                                     const envoy::config::core::v3::ConfigSource& odcds_config,
                                     OptRef<xds::core::v3::ResourceLocator> odcds_resources_locator,
                                     ProtobufMessage::ValidationVisitor& validation_visitor) {
  // Generate a unique key for this subscription based on config and locator.
  // This enables reuse of subscriptions with the same configuration.
  // Reference counting enables cleanup when the last handle is destroyed.
  uint64_t config_hash = MessageUtil::hash(odcds_config);
  if (odcds_resources_locator.has_value()) {
    config_hash = absl::HashOf(config_hash, MessageUtil::hash(*odcds_resources_locator));
  }

  auto it = odcds_subscriptions_.find(config_hash);
  if (it != odcds_subscriptions_.end()) {
    auto handle = OdCdsApiHandleImpl::create(*this, config_hash);
    it->second->ref_count.fetch_add(1, std::memory_order_relaxed);
    return handle;
  }

  auto odcds_or_error =
      creation_function(odcds_config, odcds_resources_locator, xds_manager_, *this, *this,
                        *stats_.rootScope(), validation_visitor, context_);
  RETURN_IF_NOT_OK_REF(odcds_or_error.status());
  odcds_subscriptions_.emplace(
      config_hash, std::make_unique<OdCdsSubscriptionEntry>(std::move(*odcds_or_error)));
  return OdCdsApiHandleImpl::create(*this, config_hash);
}

ClusterManagerImpl::OdCdsApiHandleImpl::~OdCdsApiHandleImpl() {
  if (parent_.shutdown_.load()) {
    return;
  }

  if (parent_.dispatcher_.isThreadSafe()) {
    parent_.releaseOdCdsSubscription(subscription_key_);

Review comment (suggestion):
Please add in the comment above a clarification that different [timeout](https://github.com/envoyproxy/envoy/blob/f11d3ef8d5573b4685ee3385a23d14f07f5a8405/api/envoy/extensions/filters/http/on_demand/v3/on_demand.proto#L37) values will use the same config-hash (clarifying this is intentional).

Code after:
      if (predicate != nullptr && !predicate(*host)) {
        continue;
      }

      parent_.drainOrCloseConnPools(host, behavior);
    }
  }
}

ClusterUpdateCallbacksHandlePtr
ClusterManagerImpl::addThreadLocalClusterUpdateCallbacks(ClusterUpdateCallbacks& cb) {
  ThreadLocalClusterManagerImpl& cluster_manager = *tls_;
  return cluster_manager.addClusterUpdateCallbacks(cb);
}

absl::StatusOr<OdCdsApiHandlePtr>
ClusterManagerImpl::allocateOdCdsApi(OdCdsCreationFunction creation_function,
                                     const envoy::config::core::v3::ConfigSource& odcds_config,
                                     OptRef<xds::core::v3::ResourceLocator> odcds_resources_locator,
                                     ProtobufMessage::ValidationVisitor& validation_visitor) {
  // Generate a unique key based on config and locator. This enables reuse of subscriptions
  // with the same configuration. Note that timeout is intentionally not part of the hash,
  // so different timeout values will share the same subscription.
  // Subscriptions persist for the lifetime of ClusterManagerImpl and are cleaned up when
  // it is destroyed.
  uint64_t config_hash = MessageUtil::hash(odcds_config);
  if (odcds_resources_locator.has_value()) {
    config_hash = absl::HashOf(config_hash, MessageUtil::hash(*odcds_resources_locator));
  }

  auto it = odcds_subscriptions_.find(config_hash);
  if (it != odcds_subscriptions_.end()) {
    return OdCdsApiHandleImpl::create(*this, config_hash);
  }

  auto odcds_or_error =
      creation_function(odcds_config, odcds_resources_locator, xds_manager_, *this, *this,
                        *stats_.rootScope(), validation_visitor, context_);
  RETURN_IF_NOT_OK_REF(odcds_or_error.status());
  odcds_subscriptions_.emplace(config_hash, std::move(*odcds_or_error));
  return OdCdsApiHandleImpl::create(*this, config_hash);
}

ClusterDiscoveryCallbackHandlePtr
ClusterManagerImpl::requestOnDemandClusterDiscovery(uint64_t config_source_key, std::string name,
                                                    ClusterDiscoveryCallbackPtr callback,
                                                    std::chrono::milliseconds timeout) {
  ThreadLocalClusterManagerImpl& cluster_manager = *tls_;

  auto [handle, discovery_in_progress, invoker] =
      cluster_manager.cdm_.addCallback(name, std::move(callback));