dynamic_modules: added an E2E integration test for clusters (#43890)

## Description

This PR adds an E2E integration test for clusters Dynamic Modules.

---

**Commit Message:** dynamic_modules: added an E2E integration test for
clusters
**Additional Description:** Added an E2E integration test for clusters
Dynamic Modules.
**Risk Level:** Low
**Testing:** CI
**Docs Changes:** N/A
**Release Notes:** N/A

Signed-off-by: Rohit Agrawal <rohit.agrawal@databricks.com>

Author: agrawroh

source/extensions/clusters/dynamic_modules/abi_impl.cc

@@ -1078,19 +1078,42 @@ void envoy_dynamic_module_callback_cluster_lb_async_host_selection_complete(
     envoy_dynamic_module_type_cluster_host_envoy_ptr host,
     envoy_dynamic_module_type_module_buffer details) {
   auto* lb = getLb(lb_envoy_ptr);
-  auto* context = getContext(context_envoy_ptr);
-
-  Envoy::Upstream::HostConstSharedPtr host_shared;
-  if (host != nullptr) {
-    host_shared = lb->handle()->cluster()->findHost(host);
-  }
 
+  // Copy the details string on the calling thread. The pointer is not valid after we return.
   std::string details_str;
   if (details.ptr != nullptr && details.length > 0) {
     details_str.assign(details.ptr, details.length);
   }
 
-  context->onAsyncHostSelection(std::move(host_shared), std::move(details_str));
+  auto cancelled = lb->activeAsyncCancelled();
+  auto* dispatcher = lb->activeAsyncDispatcher();
+
+  if (dispatcher != nullptr) {
+    // Post all work to the worker thread. The host lookup and context access must happen
+    // on the worker thread because the module may call this from a background thread.
+    // Keep the cluster alive via the handle's shared_ptr until the callback fires.
+    auto handle = lb->handle();
+    dispatcher->post([context_envoy_ptr, host, details_str = std::move(details_str),
+                      cancelled = std::move(cancelled), handle = std::move(handle)]() {
+      if (cancelled != nullptr && cancelled->load(std::memory_order_acquire)) {
+        return;
+      }
+      auto* context = getContext(context_envoy_ptr);
+      Envoy::Upstream::HostConstSharedPtr host_shared;
+      if (host != nullptr) {
+        host_shared = handle->cluster()->findHost(host);
+      }
+      context->onAsyncHostSelection(std::move(host_shared), std::string(details_str));
+    });
+  } else {
+    // No worker dispatcher. Complete inline on the calling thread.
+    auto* context = getContext(context_envoy_ptr);
+    Envoy::Upstream::HostConstSharedPtr host_shared;
+    if (host != nullptr) {
+      host_shared = lb->handle()->cluster()->findHost(host);
+    }
+    context->onAsyncHostSelection(std::move(host_shared), std::move(details_str));
+  }
 }
 
 // =============================================================================

source/extensions/clusters/dynamic_modules/cluster.cc

@@ -2,6 +2,7 @@
 
 #include "envoy/config/core/v3/base.pb.h"
 #include "envoy/config/endpoint/v3/endpoint_components.pb.h"
+#include "envoy/network/connection.h"
 #include "envoy/network/drain_decision.h"
 #include "envoy/upstream/locality.h"
 
@@ -10,12 +11,9 @@
 #include "source/common/network/utility.h"
 #include "source/common/protobuf/protobuf.h"
 #include "source/common/protobuf/utility.h"
-#include "source/common/runtime/runtime_features.h"
 #include "source/common/upstream/upstream_impl.h"
 #include "source/extensions/dynamic_modules/dynamic_modules.h"
 
-#include "absl/strings/str_cat.h"
-
 namespace Envoy {
 namespace Extensions {
 namespace Clusters {
@@ -57,10 +55,9 @@ struct DynamicModuleThreadAwareLoadBalancer : public Upstream::ThreadAwareLoadBa
 
 absl::StatusOr<std::shared_ptr<DynamicModuleClusterConfig>> DynamicModuleClusterConfig::create(
     const std::string& cluster_name, const std::string& cluster_config,
-    const std::string& metrics_namespace,
     Envoy::Extensions::DynamicModules::DynamicModulePtr module, Stats::Scope& stats_scope) {
-  auto config = std::shared_ptr<DynamicModuleClusterConfig>(new DynamicModuleClusterConfig(
-      cluster_name, cluster_config, metrics_namespace, std::move(module), stats_scope));
+  auto config = std::shared_ptr<DynamicModuleClusterConfig>(
+      new DynamicModuleClusterConfig(cluster_name, cluster_config, std::move(module), stats_scope));
 
   // Resolve all required function pointers from the dynamic module.
 #define RESOLVE_SYMBOL(name, type, member)                                                         \
@@ -142,9 +139,8 @@ absl::StatusOr<std::shared_ptr<DynamicModuleClusterConfig>> DynamicModuleCluster
 
 DynamicModuleClusterConfig::DynamicModuleClusterConfig(
     const std::string& cluster_name, const std::string& cluster_config,
-    const std::string& metrics_namespace,
     Envoy::Extensions::DynamicModules::DynamicModulePtr module, Stats::Scope& stats_scope)
-    : stats_scope_(stats_scope.createScope(absl::StrCat(metrics_namespace, "."))),
+    : stats_scope_(stats_scope.createScope("dynamicmodulescustom.")),
       stat_name_pool_(stats_scope_->symbolTable()), cluster_name_(cluster_name),
       cluster_config_(cluster_config), dynamic_module_(std::move(module)) {}
 
@@ -161,6 +157,12 @@ DynamicModuleClusterConfig::~DynamicModuleClusterConfig() {
 DynamicModuleClusterHandle::~DynamicModuleClusterHandle() {
   std::shared_ptr<DynamicModuleCluster> cluster = std::move(cluster_);
   cluster_.reset();
+  // Release lifecycle handles eagerly while the lifecycle notifier is still valid. When the
+  // dispatcher destructor clears pending callbacks, the cluster destructor would otherwise try to
+  // unregister from already-destroyed lifecycle notifier lists.
+  cluster->server_initialized_handle_.reset();
+  cluster->shutdown_handle_.reset();
+  cluster->drain_handle_.reset();
   Event::Dispatcher& dispatcher = cluster->dispatcher_;
   dispatcher.post([cluster = std::move(cluster)]() mutable { cluster.reset(); });
 }
@@ -620,6 +622,14 @@ DynamicModuleLoadBalancer::chooseHost(Upstream::LoadBalancerContext* context) {
     return {nullptr};
   }
 
+  // Pre-capture the worker dispatcher and prepare the cancellation flag before calling into the
+  // module. The module's choose_host may spawn a background thread that calls
+  // async_host_selection_complete, which reads these fields. Setting them beforehand establishes
+  // a happens-before relationship via the thread::spawn synchronization in the module.
+  const auto* connection = context != nullptr ? context->downstreamConnection() : nullptr;
+  active_async_dispatcher_ = connection != nullptr ? &connection->dispatcher() : nullptr;
+  active_async_cancelled_ = std::make_shared<std::atomic<bool>>(false);
+
   envoy_dynamic_module_type_cluster_host_envoy_ptr host_ptr = nullptr;
   envoy_dynamic_module_type_cluster_lb_async_handle_module_ptr async_handle = nullptr;
   handle_->cluster_->config()->on_cluster_lb_choose_host_(in_module_lb_, context, &host_ptr,
@@ -629,10 +639,14 @@ DynamicModuleLoadBalancer::chooseHost(Upstream::LoadBalancerContext* context) {
     // Async pending: the module will call the completion callback later.
     auto cancelable = std::make_unique<DynamicModuleAsyncHostSelectionHandle>(
         async_handle, in_module_lb_,
-        handle_->cluster_->config()->on_cluster_lb_cancel_host_selection_);
+        handle_->cluster_->config()->on_cluster_lb_cancel_host_selection_, active_async_cancelled_);
     return Upstream::HostSelectionResponse{nullptr, std::move(cancelable)};
   }
 
+  // Synchronous result or no host. Clear the async state.
+  active_async_dispatcher_ = nullptr;
+  active_async_cancelled_ = nullptr;
+
   if (host_ptr == nullptr) {
     return {nullptr};
   }
@@ -642,12 +656,19 @@ DynamicModuleLoadBalancer::chooseHost(Upstream::LoadBalancerContext* context) {
   return {host};
 }
 
-void DynamicModuleAsyncHostSelectionHandle::cancel() {
-  if (cancel_fn_ != nullptr) {
+DynamicModuleAsyncHostSelectionHandle::~DynamicModuleAsyncHostSelectionHandle() {
+  // Free the module-side async handle. The cancel function takes ownership of the handle and
+  // drops it, so this works for both cancellation and normal completion paths.
+  if (async_handle_ != nullptr && cancel_fn_ != nullptr) {
     cancel_fn_(in_module_lb_, async_handle_);
+    async_handle_ = nullptr;
   }
 }
 
+void DynamicModuleAsyncHostSelectionHandle::cancel() {
+  cancelled_->store(true, std::memory_order_release);
+}
+
 const Upstream::PrioritySet& DynamicModuleLoadBalancer::prioritySet() const {
   return handle_->cluster_->prioritySet();
 }
@@ -724,28 +745,14 @@ DynamicModuleClusterFactory::createClusterWithConfig(
                                                   module_or_error.status().message()));
   }
 
-  // Use configured metrics namespace or fall back to the default.
-  const std::string metrics_namespace = module_config.metrics_namespace().empty()
-                                            ? std::string(DefaultMetricsNamespace)
-                                            : module_config.metrics_namespace();
-
   // Create the cluster configuration.
   auto config_or_error = DynamicModuleClusterConfig::create(
-      proto_config.cluster_name(), cluster_config_bytes, metrics_namespace,
-      std::move(module_or_error.value()), context.serverFactoryContext().serverScope());
+      proto_config.cluster_name(), cluster_config_bytes, std::move(module_or_error.value()),
+      context.serverFactoryContext().serverScope());
   if (!config_or_error.ok()) {
     return config_or_error.status();
   }
 
-  // When the runtime guard is enabled, register the metrics namespace as a custom stat namespace.
-  // This causes the namespace prefix to be stripped from prometheus output and no envoy_ prefix
-  // is added. This is the legacy behavior for backward compatibility.
-  if (Runtime::runtimeFeatureEnabled(
-          "envoy.reloadable_features.dynamic_modules_strip_custom_stat_prefix")) {
-    context.serverFactoryContext().api().customStatNamespaces().registerStatNamespace(
-        metrics_namespace);
-  }
-
   // Create the cluster.
   absl::Status creation_status = absl::OkStatus();
   auto new_cluster = std::shared_ptr<DynamicModuleCluster>(new DynamicModuleCluster(

source/extensions/clusters/dynamic_modules/cluster.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <atomic>
 #include <memory>
 #include <string>
 #include <vector>
@@ -31,10 +32,6 @@ namespace Extensions {
 namespace Clusters {
 namespace DynamicModules {
 
-// The default custom stat namespace which prepends all user-defined metrics.
-// This can be overridden via the ``metrics_namespace`` field in ``DynamicModuleConfig``.
-constexpr absl::string_view DefaultMetricsNamespace = "dynamicmodulescustom";
-
 class DynamicModuleCluster;
 class DynamicModuleClusterScheduler;
 class DynamicModuleClusterTestPeer;
@@ -70,14 +67,12 @@ class DynamicModuleClusterConfig {
    *
    * @param cluster_name the name identifying the cluster implementation in the module.
    * @param cluster_config the configuration bytes to pass to the module.
-   * @param metrics_namespace the namespace prefix for metrics emitted by this module.
    * @param module the loaded dynamic module.
    * @param stats_scope the stats scope for creating custom metrics.
    * @return a shared pointer to the config, or an error status.
    */
   static absl::StatusOr<std::shared_ptr<DynamicModuleClusterConfig>>
   create(const std::string& cluster_name, const std::string& cluster_config,
-         const std::string& metrics_namespace,
          Envoy::Extensions::DynamicModules::DynamicModulePtr module, Stats::Scope& stats_scope);
 
   ~DynamicModuleClusterConfig();
@@ -268,7 +263,6 @@ class DynamicModuleClusterConfig {
 
 private:
   DynamicModuleClusterConfig(const std::string& cluster_name, const std::string& cluster_config,
-                             const std::string& metrics_namespace,
                              Envoy::Extensions::DynamicModules::DynamicModulePtr module,
                              Stats::Scope& stats_scope);
 
@@ -471,15 +465,19 @@ class DynamicModuleAsyncHostSelectionHandle : public Upstream::AsyncHostSelectio
   DynamicModuleAsyncHostSelectionHandle(
       envoy_dynamic_module_type_cluster_lb_async_handle_module_ptr async_handle,
       envoy_dynamic_module_type_cluster_lb_module_ptr in_module_lb,
-      OnClusterLbCancelHostSelectionType cancel_fn)
-      : async_handle_(async_handle), in_module_lb_(in_module_lb), cancel_fn_(cancel_fn) {}
+      OnClusterLbCancelHostSelectionType cancel_fn, std::shared_ptr<std::atomic<bool>> cancelled)
+      : async_handle_(async_handle), in_module_lb_(in_module_lb), cancel_fn_(cancel_fn),
+        cancelled_(std::move(cancelled)) {}
+
+  ~DynamicModuleAsyncHostSelectionHandle() override;
 
   void cancel() override;
 
 private:
   envoy_dynamic_module_type_cluster_lb_async_handle_module_ptr async_handle_;
   envoy_dynamic_module_type_cluster_lb_module_ptr in_module_lb_;
   OnClusterLbCancelHostSelectionType cancel_fn_;
+  std::shared_ptr<std::atomic<bool>> cancelled_;
 };
 
 /**
@@ -510,6 +508,22 @@ class DynamicModuleLoadBalancer : public Upstream::LoadBalancer {
   // Access the handle for async host selection completion.
   const DynamicModuleClusterHandleSharedPtr& handle() const { return handle_; }
 
+  /**
+   * Returns the shared cancellation flag for the current async host selection. When the router
+   * cancels the selection (e.g., stream timeout), the flag is set so the posted completion
+   * callback becomes a no-op. Returns nullptr when there is no active async selection.
+   */
+  std::shared_ptr<std::atomic<bool>> activeAsyncCancelled() const {
+    return active_async_cancelled_;
+  }
+
+  /**
+   * Returns the worker thread's dispatcher captured during chooseHost. Used by the async
+   * completion callback in abi_impl.cc to post to the correct worker thread without accessing
+   * the LoadBalancerContext from a background thread.
+   */
+  Event::Dispatcher* activeAsyncDispatcher() const { return active_async_dispatcher_; }
+
   // Per-host custom data storage.
   bool setHostData(uint32_t priority, size_t index, uintptr_t data);
   bool getHostData(uint32_t priority, size_t index, uintptr_t* data) const;
@@ -522,6 +536,13 @@ class DynamicModuleLoadBalancer : public Upstream::LoadBalancer {
   const DynamicModuleClusterHandleSharedPtr handle_;
   envoy_dynamic_module_type_cluster_lb_module_ptr in_module_lb_;
 
+  // Shared cancellation flag for the active async host selection. Set in chooseHost when the
+  // module returns AsyncPending, and read by the posted completion callback in abi_impl.cc.
+  std::shared_ptr<std::atomic<bool>> active_async_cancelled_;
+
+  // Worker thread dispatcher captured during chooseHost for async completion posting.
+  Event::Dispatcher* active_async_dispatcher_{nullptr};
+
   // Per-host data storage keyed by (priority, index). This is per-LB-instance (per-worker).
   absl::flat_hash_map<std::pair<uint32_t, size_t>, uintptr_t> per_host_data_;
 

test/extensions/clusters/dynamic_modules/BUILD

@@ -41,3 +41,22 @@ envoy_cc_test(
         "@envoy_api//envoy/extensions/clusters/dynamic_modules/v3:pkg_cc_proto",
     ],
 )
+
+envoy_cc_test(
+    name = "integration_test",
+    srcs = ["integration_test.cc"],
+    data = [
+        "//test/extensions/dynamic_modules/test_data/rust:cluster_integration_test",
+    ],
+    rbe_pool = "6gig",
+    deps = [
+        "//source/extensions/clusters/dynamic_modules:cluster",
+        "//source/extensions/dynamic_modules:abi_impl",
+        "//source/extensions/load_balancing_policies/cluster_provided:config",
+        "//test/extensions/dynamic_modules:util",
+        "//test/integration:http_integration_lib",
+        "@envoy_api//envoy/config/bootstrap/v3:pkg_cc_proto",
+        "@envoy_api//envoy/config/cluster/v3:pkg_cc_proto",
+        "@envoy_api//envoy/extensions/clusters/dynamic_modules/v3:pkg_cc_proto",
+    ],
+)

test/extensions/clusters/dynamic_modules/cluster_test.cc

@@ -2059,8 +2059,10 @@ TEST_F(DynamicModuleClusterTest, AsyncHostSelectionHandleCancel) {
       reinterpret_cast<envoy_dynamic_module_type_cluster_lb_async_handle_module_ptr>(0xCAFE);
   auto* dummy_lb = reinterpret_cast<envoy_dynamic_module_type_cluster_lb_module_ptr>(0xBEEF);
 
-  DynamicModuleAsyncHostSelectionHandle handle(dummy_async_handle, dummy_lb, nullptr);
+  auto cancelled = std::make_shared<std::atomic<bool>>(false);
+  DynamicModuleAsyncHostSelectionHandle handle(dummy_async_handle, dummy_lb, nullptr, cancelled);
   handle.cancel();
+  EXPECT_TRUE(cancelled->load());
 }
 
 // Test DynamicModuleAsyncHostSelectionHandle cancel with null cancel_fn.
@@ -2069,7 +2071,8 @@ TEST_F(DynamicModuleClusterTest, AsyncHostSelectionHandleCancelNullFn) {
       reinterpret_cast<envoy_dynamic_module_type_cluster_lb_async_handle_module_ptr>(0xCAFE);
   auto* dummy_lb = reinterpret_cast<envoy_dynamic_module_type_cluster_lb_module_ptr>(0xBEEF);
 
-  DynamicModuleAsyncHostSelectionHandle handle(dummy_async_handle, dummy_lb, nullptr);
+  auto cancelled = std::make_shared<std::atomic<bool>>(false);
+  DynamicModuleAsyncHostSelectionHandle handle(dummy_async_handle, dummy_lb, nullptr, cancelled);
   // Should not crash with nullptr cancel function.
   handle.cancel();
 }