datadog: event scheduling portion of new tracing library

source/extensions/tracers/datadog/BUILD

@@ -16,13 +16,15 @@ envoy_cc_library(
     srcs = [
         "datadog_tracer_impl.cc",
         "dict_util.cc",
+        "event_scheduler.cc",
         "logger.cc",
         "span.cc",
         "time_util.cc",
     ],
     hdrs = [
         "datadog_tracer_impl.h",
         "dict_util.h",
+        "event_scheduler.h",
         "logger.h",
         "span.h",
         "time_util.h",

source/extensions/tracers/datadog/event_scheduler.cc

@@ -0,0 +1,76 @@
+#include "source/extensions/tracers/datadog/event_scheduler.h"
+
+#include <memory>
+#include <utility>
+
+#include "source/common/common/assert.h"
+
+#include "datadog/json.hpp"
+
+namespace Envoy {
+namespace Extensions {
+namespace Tracers {
+namespace Datadog {
+
+EventScheduler::EventScheduler(Event::Dispatcher& dispatcher) : dispatcher_(dispatcher) {}
+
+EventScheduler::Cancel
+EventScheduler::schedule_recurring_event(std::chrono::steady_clock::duration interval,
+                                         std::function<void()> callback) {
+  auto self = std::make_shared<Event::Timer*>();
+  // Yes, a shared pointer to a pointer.
+  //
+  // Both the timer callback (argument to `createTimer`, below) and the returned
+  // cancellation function need a handle to the `Event::Timer`. The timer
+  // callback needs it so that it can reschedule the next round when the timer
+  // fires (that's how this is a _recurring_ event). The cancellation function
+  // needs it so that it can call `disableTimer` and remove the timer from
+  // `timers_`.
+  //
+  // Since we don't have a handle to the `Event::Timer` until `createTimer`
+  // returns, we need a "box" out of which the timer callback can extract the
+  // created timer. We then put the `Event::Timer*` in the "box" after
+  // `createTimer` returns the timer. `self` is the "box."
+  //
+  // The cancellation function returned by this function refers to the
+  // `Event::Timer` via a raw pointer (as does the timer callback, indirectly).
+  // The actual lifetime of the pointed-to `Event::Timer` is determined by its
+  // presence in `timers_`. The `Event::TimerPtr` returned by `createTimer` is
+  // moved into `timers_`.
+
+  Event::TimerPtr timer = dispatcher_.createTimer([self, interval, callback = std::move(callback)] {
+    (**self).enableTimer(std::chrono::duration_cast<std::chrono::milliseconds>(interval));
+    callback();
+  });
+
+  Event::Timer* timer_raw = timer.get();
+  *self = timer_raw;
+
+  timers_.insert(std::move(timer));
+
+  timer_raw->enableTimer(std::chrono::duration_cast<std::chrono::milliseconds>(interval));
+
+  return [this, timer = timer_raw]() mutable {
+    if (!timer) {
+      return; // idempotent
+    }
+
+    timer->disableTimer();
+    auto found = timers_.find(timer);
+    RELEASE_ASSERT(found != timers_.end(),
+                   "timer not found in registry of timers in Datadog::EventScheduler");
+    timers_.erase(found);
+    timer = nullptr;
+  };
+}
+
+nlohmann::json EventScheduler::config_json() const {
+  return nlohmann::json::object({
+      {"type", "Envoy::Extensions::Tracers::Datadog::EventScheduler"},
+  });
+}
+
+} // namespace Datadog
+} // namespace Tracers
+} // namespace Extensions
+} // namespace Envoy

source/extensions/tracers/datadog/event_scheduler.h

@@ -0,0 +1,48 @@
+#pragma once
+
+#include "envoy/event/dispatcher.h"
+#include "envoy/event/timer.h"
+
+#include "absl/container/flat_hash_set.h"
+#include "datadog/event_scheduler.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Tracers {
+namespace Datadog {
+
+/**
+ * Registry of recurring events (timers). This class implements dd-trace-cpp's
+ * \c datadog::tracing::EventScheduler interface in terms of an
+ * \c Event::Dispatcher, which is what is used by Envoy. An instance of this class
+ * is passed into dd-trace-cpp when tracing is configured, allowing dd-trace-cpp
+ * to periodically send batches of traces to the Datadog Agent.
+ */
+class EventScheduler : public datadog::tracing::EventScheduler {
+public:
+  explicit EventScheduler(Event::Dispatcher& dispatcher);
+
+  /**
+   * Repeatedly execute the specified \p callback with approximately \p interval
+   * between each invocation, starting after an initial \p interval. Return a
+   * function that cancels future invocations. If the returned function is
+   * invoked after this \c EventScheduler is destroyed, the behavior is
+   * undefined.
+   * @param interval how often the event will occur
+   * @param callback the function invoked when the event occurs
+   * @return a zero-parameter function that cancels the recurring event
+   */
+  Cancel schedule_recurring_event(std::chrono::steady_clock::duration interval,
+                                  std::function<void()> callback) override;
+
+  nlohmann::json config_json() const override;
+
+private:
+  Event::Dispatcher& dispatcher_;
+  absl::flat_hash_set<Event::TimerPtr> timers_;
+};
+
+} // namespace Datadog
+} // namespace Tracers
+} // namespace Extensions
+} // namespace Envoy

test/extensions/tracers/datadog/BUILD

@@ -16,6 +16,7 @@ envoy_extension_cc_test(
     srcs = [
         "datadog_tracer_impl_test.cc",
         "dict_util_test.cc",
+        "event_scheduler_test.cc",
         "logger_test.cc",
         "span_test.cc",
         "time_util_test.cc",

test/extensions/tracers/datadog/event_scheduler_test.cc

@@ -0,0 +1,97 @@
+#include <chrono>
+
+#include "source/extensions/tracers/datadog/event_scheduler.h"
+
+#include "test/mocks/event/mocks.h"
+#include "test/mocks/thread_local/mocks.h"
+
+#include "datadog/json.hpp"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Tracers {
+namespace Datadog {
+namespace {
+
+TEST(DatadogEventSchedulerTest, ScheduleRecurringEventCallsCreatesATimer) {
+  testing::NiceMock<ThreadLocal::MockInstance> thread_local_storage_;
+
+  EventScheduler scheduler{thread_local_storage_.dispatcher_};
+  testing::MockFunction<void()> callback;
+  // The interval is arbitrary in these tests; we just have to be able to
+  // compare it to what was passed to the mocks.
+  // The only requirement is that it be divisible by milliseconds, because
+  // that's what `Timer::enableTimer` accepts.
+  const std::chrono::milliseconds interval(2000);
+
+  EXPECT_CALL(thread_local_storage_.dispatcher_, createTimer_(testing::_));
+
+  scheduler.schedule_recurring_event(interval, callback.AsStdFunction());
+}
+
+// This could be tested above, but introducing an `Event::MockTimer` disrupts
+// our ability to track calls to `MockDispatcher::createTimer_`. So, two
+// separate tests.
+TEST(DatadogEventSchedulerTest, ScheduleRecurringEventEnablesATimer) {
+  testing::NiceMock<ThreadLocal::MockInstance> thread_local_storage_;
+  auto* const timer = new testing::NiceMock<Event::MockTimer>(&thread_local_storage_.dispatcher_);
+
+  EventScheduler scheduler{thread_local_storage_.dispatcher_};
+  testing::MockFunction<void()> callback;
+  const std::chrono::milliseconds interval(2000);
+
+  EXPECT_CALL(*timer, enableTimer(interval, testing::_));
+
+  scheduler.schedule_recurring_event(interval, callback.AsStdFunction());
+}
+
+TEST(DatadogEventSchedulerTest, TriggeredTimerInvokesCallbackAndReschedulesItself) {
+  testing::NiceMock<ThreadLocal::MockInstance> thread_local_storage_;
+  auto* const timer = new testing::NiceMock<Event::MockTimer>(&thread_local_storage_.dispatcher_);
+
+  EventScheduler scheduler{thread_local_storage_.dispatcher_};
+  testing::MockFunction<void()> callback;
+  const std::chrono::milliseconds interval(2000);
+
+  // Once for the initial round, and then again when the callback is invoked.
+  EXPECT_CALL(*timer, enableTimer(interval, testing::_)).Times(2);
+  // The user-supplied callback is called once when the timer triggers.
+  EXPECT_CALL(callback, Call());
+
+  scheduler.schedule_recurring_event(interval, callback.AsStdFunction());
+  timer->invokeCallback();
+}
+
+TEST(DatadogEventSchedulerTest, CancellationFunctionCallsDisableTimerOnce) {
+  testing::NiceMock<ThreadLocal::MockInstance> thread_local_storage_;
+  auto* const timer = new testing::NiceMock<Event::MockTimer>(&thread_local_storage_.dispatcher_);
+
+  EventScheduler scheduler{thread_local_storage_.dispatcher_};
+  testing::MockFunction<void()> callback;
+  const std::chrono::milliseconds interval(2000);
+
+  EXPECT_CALL(*timer, disableTimer());
+
+  const auto cancel = scheduler.schedule_recurring_event(interval, callback.AsStdFunction());
+  cancel();
+  cancel(); // idempotent
+  cancel(); // idempotent
+  cancel(); // idempotent
+  cancel(); // idempotent
+  cancel(); // idempotent
+}
+
+TEST(DatadogEventSchedulerTest, ConfigJson) {
+  testing::NiceMock<ThreadLocal::MockInstance> thread_local_storage_;
+  EventScheduler scheduler{thread_local_storage_.dispatcher_};
+  nlohmann::json config = scheduler.config_json();
+  EXPECT_EQ("Envoy::Extensions::Tracers::Datadog::EventScheduler", config["type"]);
+}
+
+} // namespace
+} // namespace Datadog
+} // namespace Tracers
+} // namespace Extensions
+} // namespace Envoy

test/extensions/tracers/datadog/logger_test.cc

@@ -50,14 +50,19 @@ TEST(DatadogTracerLoggerTest, Logger) {
   spdlog::logger spdlogger{"test", sink};
   Logger logger{spdlogger};
 
+  const auto reset = [&]() {
+    sink->reset();
+    spdlogger.set_level(spdlog::level::info);
+  };
+
   // callback-style error
   logger.log_error([](std::ostream& log) { log << "Beware the ides of March."; });
   EXPECT_EQ("Beware the ides of March.", sink->payload_);
   EXPECT_EQ(absl::nullopt, sink->pattern_);
   EXPECT_EQ(nullptr, sink->formatter_);
   EXPECT_FALSE(sink->flush_);
 
-  sink->reset();
+  reset();
   // callback-style startup banner
   logger.log_startup([](std::ostream& log) {
     log << "It's my stapler, the Swingline. It's been mine for a very long time.";
@@ -67,7 +72,7 @@ TEST(DatadogTracerLoggerTest, Logger) {
   EXPECT_EQ(nullptr, sink->formatter_);
   EXPECT_FALSE(sink->flush_);
 
-  sink->reset();
+  reset();
   // Error-style error
   logger.log_error(datadog::tracing::Error{datadog::tracing::Error::OTHER,
                                            "I'm sorry, Dave, I'm afraid I can't do that."});
@@ -76,7 +81,7 @@ TEST(DatadogTracerLoggerTest, Logger) {
   EXPECT_EQ(nullptr, sink->formatter_);
   EXPECT_FALSE(sink->flush_);
 
-  sink->reset();
+  reset();
   // string-style error
   logger.log_error("I must make my witness. I must lead the people from the waters. I must stay "
                    "their stampede to the sea.");
@@ -93,28 +98,28 @@ TEST(DatadogTracerLoggerTest, Logger) {
   // so if we set the level to `critical`, our `error` messages will not be
   // logged.
 
-  sink->reset();
+  reset();
   // callback-style error
-  sink->set_level(spdlog::level::critical);
+  spdlogger.set_level(spdlog::level::critical);
   logger.log_error([](std::ostream& log) { log << "Beware the ides of March."; });
   EXPECT_EQ(absl::nullopt, sink->payload_);
   EXPECT_EQ(absl::nullopt, sink->pattern_);
   EXPECT_EQ(nullptr, sink->formatter_);
   EXPECT_FALSE(sink->flush_);
 
-  sink->reset();
+  reset();
   // Error-style error
-  sink->set_level(spdlog::level::critical);
+  spdlogger.set_level(spdlog::level::critical);
   logger.log_error(datadog::tracing::Error{datadog::tracing::Error::OTHER,
                                            "I'm sorry, Dave, I'm afraid I can't do that."});
   EXPECT_EQ(absl::nullopt, sink->payload_);
   EXPECT_EQ(absl::nullopt, sink->pattern_);
   EXPECT_EQ(nullptr, sink->formatter_);
   EXPECT_FALSE(sink->flush_);
 
-  sink->reset();
+  reset();
   // string-style error
-  sink->set_level(spdlog::level::critical);
+  spdlogger.set_level(spdlog::level::critical);
   logger.log_error("I must make my witness. I must lead the people from the waters. I must stay "
                    "their stampede to the sea.");
   EXPECT_EQ(absl::nullopt, sink->payload_);
@@ -125,18 +130,18 @@ TEST(DatadogTracerLoggerTest, Logger) {
   // The startup banner is printed at "info" level, and so if the logger's level
   // threshold is more severe than that, the startup banner will not be logged.
 
-  sink->reset();
+  reset();
   // The default level is `info`, so the startup banner will be logged.
   logger.log_startup([](std::ostream& log) { log << "ג  וַיֹּאמֶר אֱלֹהִים, יְהִי אוֹר; וַיְהִי-אוֹר."; });
   EXPECT_EQ("ג  וַיֹּאמֶר אֱלֹהִים, יְהִי אוֹר; וַיְהִי-אוֹר.", sink->payload_);
   EXPECT_EQ(absl::nullopt, sink->pattern_);
   EXPECT_EQ(nullptr, sink->formatter_);
   EXPECT_FALSE(sink->flush_);
 
-  sink->reset();
+  reset();
   // Any level more severe than `info` (`warn`, `error`, `critical`) will
   // suppress the startup banner. `warn` suffices.
-  sink->set_level(spdlog::level::warn);
+  spdlogger.set_level(spdlog::level::warn);
   logger.log_startup([](std::ostream& log) { log << "R - ½ R g + Λ g  =  κ T"; });
   EXPECT_EQ(absl::nullopt, sink->payload_);
   EXPECT_EQ(absl::nullopt, sink->pattern_);