dispatcher: expose evwatch registration (#46163)

#46101 added an API to
dynamically register evwatch hooks into a libevent scheduler. This PR
just exposes the API via the dispatcher.

Risk Level: none (not yet used)
Testing: unit test added, underlying LibeventScheduler impl has its own
tests

Signed-off-by: antoniovleonti <leonti@google.com>

Author: antoniovleonti

envoy/event/BUILD

@@ -24,6 +24,7 @@ envoy_cc_library(
     deps = [
         ":deferred_deletable",
         ":dispatcher_thread_deletable",
+        ":evwatch_interface",
         ":file_event_interface",
         ":scaled_timer",
         ":schedulable_cb_interface",

envoy/event/dispatcher.h

@@ -8,6 +8,7 @@
 #include "envoy/common/scope_tracker.h"
 #include "envoy/common/time.h"
 #include "envoy/event/dispatcher_thread_deletable.h"
+#include "envoy/event/evwatch.h"
 #include "envoy/event/file_event.h"
 #include "envoy/event/scaled_timer.h"
 #include "envoy/event/schedulable_cb.h"
@@ -157,6 +158,15 @@ class Dispatcher : public DispatcherBase, public ScopeTracker {
   virtual void registerWatchdog(const Server::WatchDogSharedPtr& watchdog,
                                 std::chrono::milliseconds min_touch_interval) PURE;
 
+  /**
+   * Registers an Evwatch observer with this dispatcher.
+   * This should only be called on the dispatcher's thread.
+   * @param observer supplies the observer to register.
+   * @return Evwatch::ObserverHandlePtr handle that automatically unregisters the observer when
+   * destroyed.
+   */
+  virtual Evwatch::ObserverHandlePtr registerEvwatchObserver(Evwatch::ObserverPtr observer) PURE;
+
   /**
    * Returns a time-source to use with this dispatcher.
    */

source/common/event/dispatcher_impl.cc

@@ -98,6 +98,11 @@ void DispatcherImpl::registerWatchdog(const Server::WatchDogSharedPtr& watchdog,
       std::make_unique<WatchdogRegistration>(watchdog, *scheduler_, min_touch_interval, *this);
 }
 
+Evwatch::ObserverHandlePtr DispatcherImpl::registerEvwatchObserver(Evwatch::ObserverPtr observer) {
+  ASSERT(isThreadSafe());
+  return base_scheduler_.registerEvwatchObserver(std::move(observer));
+}
+
 void DispatcherImpl::initializeStats(Stats::Scope& scope,
                                      const std::optional<std::string>& prefix) {
   const std::string effective_prefix = prefix.has_value() ? *prefix : absl::StrCat(name_, ".");

source/common/event/dispatcher_impl.h

@@ -59,6 +59,7 @@ class DispatcherImpl : Logger::Loggable<Logger::Id::main>,
   const std::string& name() override { return name_; }
   void registerWatchdog(const Server::WatchDogSharedPtr& watchdog,
                         std::chrono::milliseconds min_touch_interval) override;
+  Evwatch::ObserverHandlePtr registerEvwatchObserver(Evwatch::ObserverPtr observer) override;
   TimeSource& timeSource() override { return time_source_; }
   void initializeStats(Stats::Scope& scope, const std::optional<std::string>& prefix) override;
   void clearDeferredDeleteList() override;

test/common/event/dispatcher_impl_test.cc

@@ -1525,6 +1525,35 @@ TEST_F(DispatcherConnectionTest, CreateEnvoyInternalConnectionWhenFactoryNotExis
       "");
 }
 
+TEST(EvwatchObserverTest, RegisterEvwatchObserver) {
+  class MockEvwatchObserver : public Evwatch::Observer {
+  public:
+    MOCK_METHOD(void, onPrepare,
+                (MonotonicTime prepare_time, std::optional<MonotonicTime::duration> timeout));
+    MOCK_METHOD(void, onCheck, (MonotonicTime check_time));
+  };
+
+  Api::ApiPtr api = Api::createApiForTest();
+  DispatcherPtr dispatcher = api->allocateDispatcher("test_thread");
+
+  EXPECT_EQ(nullptr, dispatcher->registerEvwatchObserver(nullptr));
+
+  auto observer = std::make_unique<NiceMock<MockEvwatchObserver>>();
+  auto* observer_ptr = observer.get();
+
+  EXPECT_CALL(*observer_ptr, onPrepare(_, _)).Times(testing::AtLeast(1));
+  EXPECT_CALL(*observer_ptr, onCheck(_)).Times(testing::AtLeast(1));
+
+  auto handle = dispatcher->registerEvwatchObserver(std::move(observer));
+  EXPECT_NE(nullptr, handle);
+
+  auto cb = dispatcher->createSchedulableCallback([]() {});
+  cb->scheduleCallbackCurrentIteration();
+  dispatcher->run(Dispatcher::RunType::NonBlock);
+
+  handle.reset();
+}
+
 } // namespace
 } // namespace Event
 } // namespace Envoy

test/mocks/event/mocks.h

@@ -114,9 +114,14 @@ class MockDispatcher : public Dispatcher {
     return SignalEventPtr{listenForSignal_(signal_num, cb)};
   }
 
+  Evwatch::ObserverHandlePtr registerEvwatchObserver(Evwatch::ObserverPtr observer) override {
+    return Evwatch::ObserverHandlePtr{registerEvwatchObserver_(observer.get())};
+  }
+
   // Event::Dispatcher
   MOCK_METHOD(void, registerWatchdog,
               (const Server::WatchDogSharedPtr&, std::chrono::milliseconds));
+  MOCK_METHOD(Evwatch::ObserverHandle*, registerEvwatchObserver_, (Evwatch::Observer * observer));
   MOCK_METHOD(void, initializeStats, (Stats::Scope&, const std::optional<std::string>&));
   MOCK_METHOD(void, clearDeferredDeleteList, ());
   MOCK_METHOD(Network::ServerConnection*, createServerConnection_, (StreamInfo::StreamInfo & info));

test/mocks/event/wrapped_dispatcher.h

@@ -25,6 +25,10 @@ class WrappedDispatcher : public Dispatcher {
     impl_.registerWatchdog(watchdog, min_touch_interval);
   }
 
+  Evwatch::ObserverHandlePtr registerEvwatchObserver(Evwatch::ObserverPtr observer) override {
+    return impl_.registerEvwatchObserver(std::move(observer));
+  }
+
   TimeSource& timeSource() override { return impl_.timeSource(); }
 
   void initializeStats(Stats::Scope& scope, const std::optional<std::string>& prefix) override {