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

Code before:

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.
  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 subscription_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));
  // This check will catch requests for discoveries from this thread only. If other thread
  // requested the same discovery, we will detect it in the main thread later.
  if (discovery_in_progress) {
    ENVOY_LOG(debug,
              "cm odcds: on-demand discovery for cluster {} is already in progress, something else "
              "in thread {} has already requested it",
              name, cluster_manager.thread_local_dispatcher_.name());
    // This worker thread has already requested a discovery of a cluster with this name, so
    // nothing more left to do here.
    //
    // We can't "just" return handle here, because handle is a part of the structured binding done
    // above. So it's not really a ClusterDiscoveryCallbackHandlePtr, but more like
    // ClusterDiscoveryCallbackHandlePtr&, so named return value optimization does not apply here
    // - it needs to be moved.
    return std::move(handle);
  }
  ENVOY_LOG(

Review comment (suggestion):
Can you elaborate on the lifetime of the subscription?
Specifically I see that elements are inserted into `odcds_subscriptions_`, but I don't see where they are removed.

Code after:

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));
  // This check will catch requests for discoveries from this thread only. If other thread
  // requested the same discovery, we will detect it in the main thread later.
  if (discovery_in_progress) {
    ENVOY_LOG(debug,
              "cm odcds: on-demand discovery for cluster {} is already in progress, something else "
              "in thread {} has already requested it",
              name, cluster_manager.thread_local_dispatcher_.name());
    // This worker thread has already requested a discovery of a cluster with this name, so
    // nothing more left to do here.
    //
    // We can't "just" return handle here, because handle is a part of the structured binding done
    // above. So it's not really a ClusterDiscoveryCallbackHandlePtr, but more like
    // ClusterDiscoveryCallbackHandlePtr&, so named return value optimization does not apply here
    // - it needs to be moved.