dym sdk: add the config scheduler support to go sdk and cpp sdk (#43748)

Commit Message: dym sdk: add the config scheduler support to go sdk and
cpp sdk
Additional Description:

Add the config scheduler to go sdk and cpp sdk.

Risk Level:
Testing:
Docs Changes:
Release Notes:
Platform Specific Features:
[Optional Runtime guard:]
[Optional Fixes #Issue]
[Optional Fixes commit #PR or SHA]
[Optional Deprecated:]
[Optional [API
Considerations](https://github.com/envoyproxy/envoy/blob/main/api/review_checklist.md):]

---------

Signed-off-by: wbpcode/wangbaiping <wbphub@gmail.com>

Author: wbpcode

source/extensions/dynamic_modules/sdk/cpp/sdk.h

@@ -757,6 +757,13 @@ class HttpFilterConfigHandle {
    * @param message The message to log.
    */
   virtual void log(LogLevel level, absl::string_view message) = 0;
+
+  /**
+   * Returns a scheduler for deferred task execution. This can only be called on config loading
+   * event and then the returned Scheduler can be used in other threads.
+   * @return Unique pointer to Scheduler instance.
+   */
+  virtual std::shared_ptr<Scheduler> getScheduler() = 0;
 };
 
 /**

source/extensions/dynamic_modules/sdk/cpp/sdk_internal.cc

@@ -1,5 +1,7 @@
 #include <cstddef>
 #include <cstdint>
+#include <memory>
+#include <mutex>
 
 #include "source/extensions/dynamic_modules/abi/abi.h"
 
@@ -142,50 +144,76 @@ using ResponseHeaders = HeaderMapImpl<envoy_dynamic_module_type_http_header_type
 using ResponseTrailers = HeaderMapImpl<envoy_dynamic_module_type_http_header_type_ResponseTrailer>;
 
 // Scheduler implementation
-class SchedulerImpl : public Scheduler {
+template <bool IsConfigScheduler> class SchedulerImplBase : public Scheduler {
 public:
-  SchedulerImpl(void* host_plugin_ptr)
-      : scheduler_ptr_(envoy_dynamic_module_callback_http_filter_scheduler_new(host_plugin_ptr)) {}
+  SchedulerImplBase(void* host_ptr) : scheduler_ptr_(newScheduler(host_ptr)) {}
 
   void schedule(std::function<void()> func) override {
+    uint64_t task_id = 0;
+
     // Lock to protect access to tasks_ and next_task_id_ manually
-    mutex_.lock();
-    const uint64_t task_id = next_task_id_++;
-    tasks_[task_id] = std::move(func);
-    mutex_.unlock();
+    {
+      std::lock_guard<std::mutex> lock(mutex_);
+      task_id = next_task_id_++;
+      tasks_[task_id] = std::move(func);
+    }
 
-    envoy_dynamic_module_callback_http_filter_scheduler_commit(scheduler_ptr_, task_id);
+    commitScheduler(scheduler_ptr_, task_id);
   }
 
   void onScheduled(uint64_t task_id) {
     std::function<void()> func;
 
-    // Lock to protect access to tasks_ manually
-    mutex_.lock();
-    auto it = tasks_.find(task_id);
-    if (it != tasks_.end()) {
-      func = std::move(it->second);
-      tasks_.erase(it);
+    {
+      // Lock to protect access to tasks_ manually
+      std::lock_guard<std::mutex> lock(mutex_);
+      auto it = tasks_.find(task_id);
+      if (it != tasks_.end()) {
+        func = std::move(it->second);
+        tasks_.erase(it);
+      }
     }
-    mutex_.unlock();
 
     if (func) {
       func();
     }
   }
 
-  ~SchedulerImpl() override {
-    envoy_dynamic_module_callback_http_filter_scheduler_delete(scheduler_ptr_);
-  }
+  ~SchedulerImplBase() override { deleteScheduler(scheduler_ptr_); }
 
 private:
+  static void* newScheduler(void* host_ptr) {
+    if constexpr (IsConfigScheduler) {
+      return envoy_dynamic_module_callback_http_filter_config_scheduler_new(host_ptr);
+    } else {
+      return envoy_dynamic_module_callback_http_filter_scheduler_new(host_ptr);
+    }
+  }
+  static void deleteScheduler(void* scheduler_ptr) {
+    if constexpr (IsConfigScheduler) {
+      envoy_dynamic_module_callback_http_filter_config_scheduler_delete(scheduler_ptr);
+    } else {
+      envoy_dynamic_module_callback_http_filter_scheduler_delete(scheduler_ptr);
+    }
+  }
+  static void commitScheduler(void* scheduler_ptr, uint64_t task_id) {
+    if constexpr (IsConfigScheduler) {
+      envoy_dynamic_module_callback_http_filter_config_scheduler_commit(scheduler_ptr, task_id);
+    } else {
+      envoy_dynamic_module_callback_http_filter_scheduler_commit(scheduler_ptr, task_id);
+    }
+  }
+
   void* scheduler_ptr_{};
 
-  absl::Mutex mutex_;
-  uint64_t next_task_id_ ABSL_GUARDED_BY(mutex_){1}; // 0 is reserved.
-  absl::flat_hash_map<uint64_t, std::function<void()>> tasks_ ABSL_GUARDED_BY(mutex_);
+  std::mutex mutex_;
+  uint64_t next_task_id_{1}; // 0 is reserved.
+  absl::flat_hash_map<uint64_t, std::function<void()>> tasks_;
 };
 
+using SchedulerImpl = SchedulerImplBase<false>;
+using ConfigSchedulerImpl = SchedulerImplBase<true>;
+
 // HttpFilterHandle implementation
 class HttpFilterHandleImpl : public HttpFilterHandle {
 public:
@@ -626,6 +654,15 @@ class HttpFilterConfigHandleImpl : public HttpFilterConfigHandle {
         envoy_dynamic_module_type_module_buffer{message.data(), message.size()});
   }
 
+  std::shared_ptr<Scheduler> getScheduler() override {
+    if (!scheduler_) {
+      scheduler_ = std::make_shared<ConfigSchedulerImpl>(host_config_ptr_);
+    }
+    return scheduler_;
+  }
+
+  std::shared_ptr<ConfigSchedulerImpl> scheduler_;
+
 private:
   void* host_config_ptr_;
 };
@@ -657,7 +694,7 @@ template <class T> void* wrapPointer(const T* ptr) {
 }
 
 struct HttpFilterFactoryWrapper {
-  std::unique_ptr<HttpFilterConfigHandle> config_handle_;
+  std::unique_ptr<HttpFilterConfigHandleImpl> config_handle_;
   std::unique_ptr<HttpFilterFactory> factory_;
 };
 
@@ -992,6 +1029,17 @@ void envoy_dynamic_module_on_http_filter_downstream_below_write_buffer_low_water
     plugin_handle->downstream_watermark_callbacks_->onBelowWriteBufferLowWatermark();
   }
 }
+
+void envoy_dynamic_module_on_http_filter_config_scheduled(
+    envoy_dynamic_module_type_http_filter_config_envoy_ptr,
+    envoy_dynamic_module_type_http_filter_config_module_ptr filter_config_ptr, uint64_t event_id) {
+  auto* factory_wrapper = unwrapPointer<HttpFilterFactoryWrapper>(filter_config_ptr);
+  if (factory_wrapper == nullptr || factory_wrapper->config_handle_ == nullptr ||
+      factory_wrapper->config_handle_->scheduler_ == nullptr) {
+    return;
+  }
+  factory_wrapper->config_handle_->scheduler_->onScheduled(event_id);
+}
 }
 
 } // namespace DynamicModules

source/extensions/dynamic_modules/sdk/go/abi/internal.go

@@ -26,6 +26,7 @@ import (
 
 type httpFilterConfigWrapper struct {
 	pluginFactory shared.HttpFilterFactory
+	configHandle  *dymConfigHandle
 }
 
 type httpFilterConfigWrapperPerRoute struct {
@@ -349,18 +350,21 @@ func (b *dymBodyBuffer) Drain(size uint64) {
 }
 
 type dymScheduler struct {
-	schedulerPtr  C.envoy_dynamic_module_type_http_filter_scheduler_module_ptr
+	schedulerPtr  unsafe.Pointer
 	schedulerLock sync.Mutex
 	nextTaskID    uint64
 	tasks         map[uint64]func()
+	commitFunc    func(unsafe.Pointer, C.uint64_t)
 }
 
 func newDymScheduler(
-	schedulerPtr C.envoy_dynamic_module_type_http_filter_scheduler_module_ptr,
+	schedulerPtr unsafe.Pointer,
+	commitFunc func(unsafe.Pointer, C.uint64_t),
 ) *dymScheduler {
 	return &dymScheduler{
 		schedulerPtr: schedulerPtr,
 		tasks:        make(map[uint64]func()),
+		commitFunc:   commitFunc,
 	}
 }
 
@@ -372,10 +376,8 @@ func (s *dymScheduler) Schedule(task func()) {
 	s.tasks[taskID] = task
 	s.schedulerLock.Unlock()
 
-	C.envoy_dynamic_module_callback_http_filter_scheduler_commit(
-		s.schedulerPtr,
-		(C.uint64_t)(taskID),
-	)
+	// Call the host to schedule the task, passing the task ID as context
+	s.commitFunc(s.schedulerPtr, C.uint64_t(taskID))
 }
 
 func (s *dymScheduler) onScheduled(taskID uint64) {
@@ -850,11 +852,19 @@ func (h *dymHttpFilterHandle) GetScheduler() shared.Scheduler {
 		// in practice. But it will be nil in mock tests.
 		schedulerPtr := C.envoy_dynamic_module_callback_http_filter_scheduler_new(
 			h.hostPluginPtr)
-		h.scheduler = newDymScheduler(schedulerPtr)
+		h.scheduler = newDymScheduler(
+			unsafe.Pointer(schedulerPtr),
+			func(schedulerPtr unsafe.Pointer, taskID C.uint64_t) {
+				C.envoy_dynamic_module_callback_http_filter_scheduler_commit(
+					(C.envoy_dynamic_module_type_http_filter_scheduler_module_ptr)(schedulerPtr),
+					taskID,
+				)
+			},
+		)
 
 		runtime.SetFinalizer(h.scheduler, func(s *dymScheduler) {
 			C.envoy_dynamic_module_callback_http_filter_scheduler_delete(
-				s.schedulerPtr,
+				(C.envoy_dynamic_module_type_http_filter_scheduler_module_ptr)(s.schedulerPtr),
 			)
 		})
 	}
@@ -1113,6 +1123,7 @@ func newDymStreamPluginHandle(
 
 type dymConfigHandle struct {
 	hostConfigPtr C.envoy_dynamic_module_type_http_filter_config_envoy_ptr
+	scheduler     *dymScheduler
 }
 
 func (h *dymConfigHandle) Log(level shared.LogLevel, format string, args ...any) {
@@ -1195,6 +1206,31 @@ func (h *dymConfigHandle) DefineCounter(name string,
 	return shared.MetricID(metricID), shared.MetricsResult(result)
 }
 
+func (h *dymConfigHandle) GetScheduler() shared.Scheduler {
+	if h.scheduler == nil {
+		// The scheduler is created lazily and should never be nil
+		// in practice. But it will be nil in mock tests.
+		schedulerPtr := C.envoy_dynamic_module_callback_http_filter_config_scheduler_new(
+			h.hostConfigPtr)
+		h.scheduler = newDymScheduler(
+			unsafe.Pointer(schedulerPtr),
+			func(schedulerPtr unsafe.Pointer, taskID C.uint64_t) {
+				C.envoy_dynamic_module_callback_http_filter_config_scheduler_commit(
+					(C.envoy_dynamic_module_type_http_filter_config_scheduler_module_ptr)(schedulerPtr),
+					taskID,
+				)
+			},
+		)
+
+		runtime.SetFinalizer(h.scheduler, func(s *dymScheduler) {
+			C.envoy_dynamic_module_callback_http_filter_config_scheduler_delete(
+				(C.envoy_dynamic_module_type_http_filter_config_scheduler_module_ptr)(s.schedulerPtr),
+			)
+		})
+	}
+	return h.scheduler
+}
+
 type dymRouteConfigHandle struct{}
 
 func (h *dymRouteConfigHandle) Log(level shared.LogLevel, format string, args ...any) {
@@ -1236,7 +1272,7 @@ func envoy_dynamic_module_on_http_filter_config_new(
 		configHandle.Log(shared.LogLevelWarn, "Failed to load configuration: %v", err)
 		return nil
 	}
-	configPtr := configManager.record(&httpFilterConfigWrapper{pluginFactory: factory})
+	configPtr := configManager.record(&httpFilterConfigWrapper{pluginFactory: factory, configHandle: configHandle})
 	return C.envoy_dynamic_module_type_http_filter_config_module_ptr(configPtr)
 }
 
@@ -1248,6 +1284,7 @@ func envoy_dynamic_module_on_http_filter_config_destroy(
 	if factoryWrapper == nil {
 		return
 	}
+	factoryWrapper.configHandle.scheduler = nil
 	factoryWrapper.pluginFactory.OnDestroy()
 	configManager.remove(unsafe.Pointer(configPtr))
 }
@@ -1594,3 +1631,20 @@ func envoy_dynamic_module_on_http_filter_downstream_below_write_buffer_low_water
 		pluginWrapper.downstreamWatermarkCallbacks.OnBelowWriteBufferLowWatermark()
 	}
 }
+
+//export envoy_dynamic_module_on_http_filter_config_scheduled
+func envoy_dynamic_module_on_http_filter_config_scheduled(
+	_ C.envoy_dynamic_module_type_http_filter_config_envoy_ptr,
+	configPtr C.envoy_dynamic_module_type_http_filter_config_module_ptr,
+	taskID C.uint64_t,
+) {
+	configWrapper := configManager.unwrap(unsafe.Pointer(configPtr))
+	if configWrapper == nil || configWrapper.configHandle == nil {
+		return
+	}
+	ch := configWrapper.configHandle
+
+	if ch.scheduler != nil {
+		ch.scheduler.onScheduled(uint64(taskID))
+	}
+}

source/extensions/dynamic_modules/sdk/go/shared/base.go

@@ -656,4 +656,9 @@ type HttpFilterConfigHandle interface {
 	// @Return the counter metric id. This metric can never be used after the plugin
 	// config is unloaded.
 	DefineCounter(name string, tagKeys ...string) (MetricID, MetricsResult)
+
+	// GetScheduler retrieves a scheduler for deferred task execution in the config context.
+	// This should be called only during the plugin configuration phase, and the returned
+	// Scheduler can be used later even outside of the callbacks and at other threads.
+	GetScheduler() Scheduler
 }

source/extensions/dynamic_modules/sdk/go/shared/mocks/mock_base.go

@@ -90,10 +90,10 @@ class ConfigSchedulerConfigFactory : public HttpFilterConfigFactory {
     auto shared_status = std::make_shared<std::atomic<bool>>(false);
 
     // Simulate async config update.
-    std::thread([shared_status]() {
+    handle.getScheduler()->schedule([shared_status]() {
       std::this_thread::sleep_for(std::chrono::milliseconds(100)); // NO_CHECK_FORMAT(real_time)
       shared_status->store(true);
-    }).detach();
+    });
 
     return std::make_unique<ConfigSchedulerFilterFactory>(shared_status);
   }

test/extensions/dynamic_modules/test_data/cpp/http_integration_test.cc

@@ -53,11 +53,10 @@ func (f *ConfigSchedulerConfigFactory) Create(handle shared.HttpFilterConfigHand
 	sharedStatus := new(atomic.Bool)
 	sharedStatus.Store(false)
 
-	// TODO(wbpcode): to support the actual config scheduler in golang SDK.
-	go func() {
+	handle.GetScheduler().Schedule(func() {
 		time.Sleep(100 * time.Millisecond)
 		sharedStatus.Store(true)
-	}()
+	})
 
 	return &ConfigSchedulerFilterFactory{sharedStatus: sharedStatus}, nil
 }
@@ -67,6 +66,10 @@ type ConfigSchedulerFilterFactory struct {
 	sharedStatus *atomic.Bool
 }
 
+func (f *ConfigSchedulerFilterFactory) OnDestroy() {
+	runtime.GC()
+}
+
 func (f *ConfigSchedulerFilterFactory) Create(handle shared.HttpFilterHandle) shared.HttpFilter {
 	return &ConfigSchedulerFilter{sharedStatus: f.sharedStatus}
 }

test/extensions/dynamic_modules/test_data/go/http_integration_test/http_integration_test.go

@@ -882,7 +882,10 @@ def fix_source_path(self, file_path):
 
     def check_source_path(self, file_path):
         error_messages = []
-        if self.run_code_validation:
+        # This dynamic module SDK will be built into a shared library which we prefer to use
+        # standard library rather then the absl equivalents. We simply skip the content check
+        # for this directory to avoid false positives.
+        if self.run_code_validation and "dynamic_modules/sdk/cpp" not in file_path:
             error_messages = self.check_file_contents(file_path, self.check_source_line)
         if file_path.endswith((".cc", ".h")):
             error_messages += self.check_namespace(file_path)