test infrastructure: inject time everywhere

include/envoy/api/api.h

@@ -3,6 +3,7 @@
 #include <memory>
 #include <string>
 
+#include "envoy/common/time.h"
 #include "envoy/event/dispatcher.h"
 #include "envoy/filesystem/filesystem.h"
 #include "envoy/stats/store.h"
@@ -20,9 +21,10 @@ class Api {
 
   /**
    * Allocate a dispatcher.
+   * @param time_source the time source.
    * @return Event::DispatcherPtr which is owned by the caller.
    */
-  virtual Event::DispatcherPtr allocateDispatcher() PURE;
+  virtual Event::DispatcherPtr allocateDispatcher(TimeSource& time_source) PURE;
 
   /**
    * Create/open a local file that supports async appending.

include/envoy/common/time.h

@@ -16,6 +16,8 @@ typedef std::chrono::time_point<std::chrono::steady_clock> MonotonicTime;
 
 /**
  * Abstraction for getting the current system time. Useful for testing.
+ *
+ * TODO(#4160): eliminate this class and pass TimeSource everywhere.
  */
 class SystemTimeSource {
 public:
@@ -28,7 +30,10 @@ class SystemTimeSource {
 };
 
 /**
- * Abstraction for getting the current monotonically increasing time. Useful for testing.
+ * Abstraction for getting the current monotonically increasing time. Useful for
+ * testing.
+ *
+ * TODO(#4160): eliminate this class and pass TimeSource everywhere.
  */
 class MonotonicTimeSource {
 public:
@@ -39,4 +44,46 @@ class MonotonicTimeSource {
    */
   virtual MonotonicTime currentTime() PURE;
 };
+
+/**
+ * Captures a system-time source, capable of computing both monotonically increasing
+ * and real time.
+ *
+ * TODO(#4160): currently this is just a container for SystemTimeSource and
+ * MonotonicTimeSource but we should clean that up and just have this as the
+ * base class. Once that's done, TimeSource will be a pure interface.
+ */
+class TimeSource {
+public:
+  TimeSource(SystemTimeSource& system, MonotonicTimeSource& monotonic)
+      : system_(system), monotonic_(monotonic) {}
+
+  /**
+   * @return the current system time; not guaranteed to be monotonically increasing.
+   */
+  SystemTime systemTime() { return system_.currentTime(); }
+
+  /**
+   * @return the current monotonic time.
+   */
+  MonotonicTime monotonicTime() { return monotonic_.currentTime(); }
+
+  /**
+   * Compares two time-sources for equality; this is needed for mocks.
+   */
+  bool operator==(const TimeSource& ts) const {
+    return &system_ == &ts.system_ && &monotonic_ == &ts.monotonic_;
+  }
+
+  // TODO(jmarantz): Eliminate these methods and the SystemTimeSource and MonotonicTimeSource
+  // classes, and change method calls to work directly off of TimeSource.
+  SystemTimeSource& system() { return system_; }
+  MonotonicTimeSource& monotonic() { return monotonic_; }
+
+private:
+  // These are pointers rather than references in order to support assignment.
+  SystemTimeSource& system_;
+  MonotonicTimeSource& monotonic_;
+};
+
 } // namespace Envoy

include/envoy/event/dispatcher.h

@@ -32,6 +32,16 @@ class Dispatcher {
 public:
   virtual ~Dispatcher() {}
 
+  /**
+   * Returns a time-source to use with this dispatcher.
+   *
+   * TODO(#4160) the implementations currently manage timer events that
+   * ignore the time-source, and thus can't be mocked or faked. So it's
+   * difficult to mock time in an integration test without mocking out
+   * the dispatcher.
+   */
+  virtual TimeSource& timeSource() PURE;
+
   /**
    * Clear any items in the deferred deletion queue.
    */

include/envoy/server/filter_config.h

@@ -131,9 +131,16 @@ class FactoryContext {
   virtual const envoy::api::v2::core::Metadata& listenerMetadata() const PURE;
 
   /**
-   * @return SystemTimeSource& a reference to the top-level SystemTime source.
+   * @return SystemTimeSource& a reference to the system time source.
+   * TODO(#4160): This method should be eliminated, and call-sites and mocks should
+   * be converted to work with timeSource() below.
    */
   virtual SystemTimeSource& systemTimeSource() PURE;
+
+  /**
+   * @return TimeSource& a reference to the time source.
+   */
+  virtual TimeSource& timeSource() PURE;
 };
 
 class ListenerFactoryContext : public FactoryContext {

include/envoy/server/instance.h

@@ -192,6 +192,11 @@ class Instance {
    */
   virtual const LocalInfo::LocalInfo& localInfo() PURE;
 
+  /**
+   * @return the time source used for the server.
+   */
+  virtual TimeSource& timeSource() PURE;
+
   /**
    * @return the flush interval of stats sinks.
    */

include/envoy/upstream/cluster_manager.h

@@ -209,6 +209,11 @@ class ClusterManager {
   addThreadLocalClusterUpdateCallbacks(ClusterUpdateCallbacks& callbacks) PURE;
 
   virtual ClusterManagerFactory& clusterManagerFactory() PURE;
+
+  /**
+   * @return TimeSource& the time-source used with the cluster manager.
+   */
+  virtual TimeSource& timeSource() PURE;
 };
 
 typedef std::unique_ptr<ClusterManager> ClusterManagerPtr;

source/common/api/api_impl.cc

@@ -9,8 +9,8 @@
 namespace Envoy {
 namespace Api {
 
-Event::DispatcherPtr Impl::allocateDispatcher() {
-  return Event::DispatcherPtr{new Event::DispatcherImpl()};
+Event::DispatcherPtr Impl::allocateDispatcher(TimeSource& time_source) {
+  return Event::DispatcherPtr{new Event::DispatcherImpl(time_source)};
 }
 
 Impl::Impl(std::chrono::milliseconds file_flush_interval_msec)

source/common/api/api_impl.h

@@ -17,7 +17,7 @@ class Impl : public Api::Api {
   Impl(std::chrono::milliseconds file_flush_interval_msec);
 
   // Api::Api
-  Event::DispatcherPtr allocateDispatcher() override;
+  Event::DispatcherPtr allocateDispatcher(TimeSource& time_source) override;
   Filesystem::FileSharedPtr createFile(const std::string& path, Event::Dispatcher& dispatcher,
                                        Thread::BasicLockable& lock,
                                        Stats::Store& stats_store) override;

source/common/common/perf_annotation.cc

@@ -18,11 +18,10 @@
 namespace Envoy {
 
 PerfOperation::PerfOperation()
-    : start_time_(ProdMonotonicTimeSource::instance_.currentTime()),
-      context_(PerfAnnotationContext::getOrCreate()) {}
+    : context_(PerfAnnotationContext::getOrCreate()), start_time_(context_->currentTime()) {}
 
 void PerfOperation::record(absl::string_view category, absl::string_view description) {
-  const MonotonicTime end_time = ProdMonotonicTimeSource::instance_.currentTime();
+  const MonotonicTime end_time = context_->currentTime();
   const std::chrono::nanoseconds duration =
       std::chrono::duration_cast<std::chrono::nanoseconds>(end_time - start_time_);
   context_->record(duration, category, description);

source/common/common/perf_annotation.h

@@ -86,6 +86,9 @@ class PerfAnnotationContext {
   void record(std::chrono::nanoseconds duration, absl::string_view category,
               absl::string_view description);
 
+  /** @return MonotonicTime the current time */
+  MonotonicTime currentTime() { return time_source_.currentTime(); }
+
   /**
    * Renders the aggregated statistics as a string.
    * @return std::string the performance data as a formatted string.
@@ -138,6 +141,7 @@ class PerfAnnotationContext {
 #else
   DurationStatsMap duration_stats_map_;
 #endif
+  ProdMonotonicTimeSource time_source_;
 };
 
 /**
@@ -162,8 +166,8 @@ class PerfOperation {
   void record(absl::string_view category, absl::string_view description);
 
 private:
-  MonotonicTime start_time_;
   PerfAnnotationContext* context_;
+  MonotonicTime start_time_;
 };
 
 } // namespace Envoy

source/common/common/token_bucket_impl.cc

@@ -4,14 +4,15 @@
 
 namespace Envoy {
 
-TokenBucketImpl::TokenBucketImpl(uint64_t max_tokens, double fill_rate,
-                                 MonotonicTimeSource& time_source)
+TokenBucketImpl::TokenBucketImpl(uint64_t max_tokens, MonotonicTimeSource& monotonic_time_source,
+                                 double fill_rate)
     : max_tokens_(max_tokens), fill_rate_(std::abs(fill_rate)), tokens_(max_tokens),
-      last_fill_(time_source.currentTime()), time_source_(time_source) {}
+      last_fill_(monotonic_time_source.currentTime()),
+      monotonic_time_source_(monotonic_time_source) {}
 
 bool TokenBucketImpl::consume(uint64_t tokens) {
   if (tokens_ < max_tokens_) {
-    const auto time_now = time_source_.currentTime();
+    const auto time_now = monotonic_time_source_.currentTime();
     tokens_ = std::min((std::chrono::duration<double>(time_now - last_fill_).count() * fill_rate_) +
                            tokens_,
                        max_tokens_);
@@ -26,4 +27,4 @@ bool TokenBucketImpl::consume(uint64_t tokens) {
   return true;
 }
 
-} // namespace Envoy
+} // namespace Envoy

source/common/common/token_bucket_impl.h

@@ -16,10 +16,10 @@ class TokenBucketImpl : public TokenBucket {
    * @param max_tokens supplies the maximun number of tokens in the bucket.
    * @param fill_rate supplies the number of tokens that will return to the bucket on each second.
    * The default is 1.
-   * @param time_source supplies the time source. The default is ProdMonotonicTimeSource.
+   * @param time_source supplies the time source.
    */
-  explicit TokenBucketImpl(uint64_t max_tokens, double fill_rate = 1,
-                           MonotonicTimeSource& time_source = ProdMonotonicTimeSource::instance_);
+  explicit TokenBucketImpl(uint64_t max_tokens, MonotonicTimeSource& time_source,
+                           double fill_rate = 1);
 
   bool consume(uint64_t tokens = 1) override;
 
@@ -28,7 +28,7 @@ class TokenBucketImpl : public TokenBucket {
   const double fill_rate_;
   double tokens_;
   MonotonicTime last_fill_;
-  MonotonicTimeSource& time_source_;
+  MonotonicTimeSource& monotonic_time_source_;
 };
 
 } // namespace Envoy

source/common/common/utility.cc

@@ -201,9 +201,6 @@ std::string DateFormatter::now() {
   return fromTime(current_time_t);
 }
 
-ProdSystemTimeSource ProdSystemTimeSource::instance_;
-ProdMonotonicTimeSource ProdMonotonicTimeSource::instance_;
-
 ConstMemoryStreamBuffer::ConstMemoryStreamBuffer(const char* data, size_t size) {
   // std::streambuf won't modify `data`, but the interface still requires a char* for convenience,
   // so we need to const_cast.

source/common/common/utility.h

@@ -102,8 +102,6 @@ class ProdSystemTimeSource : public SystemTimeSource {
 public:
   // SystemTimeSource
   SystemTime currentTime() override { return std::chrono::system_clock::now(); }
-
-  static ProdSystemTimeSource instance_;
 };
 
 /**
@@ -113,8 +111,6 @@ class ProdMonotonicTimeSource : public MonotonicTimeSource {
 public:
   // MonotonicTimeSource
   MonotonicTime currentTime() override { return std::chrono::steady_clock::now(); }
-
-  static ProdMonotonicTimeSource instance_;
 };
 
 /**

source/common/config/grpc_mux_impl.cc

@@ -12,9 +12,9 @@ namespace Config {
 GrpcMuxImpl::GrpcMuxImpl(const LocalInfo::LocalInfo& local_info, Grpc::AsyncClientPtr async_client,
                          Event::Dispatcher& dispatcher,
                          const Protobuf::MethodDescriptor& service_method,
-                         Runtime::RandomGenerator& random, MonotonicTimeSource& time_source)
+                         Runtime::RandomGenerator& random)
     : local_info_(local_info), async_client_(std::move(async_client)),
-      service_method_(service_method), random_(random), time_source_(time_source) {
+      service_method_(service_method), random_(random), time_source_(dispatcher.timeSource()) {
   Config::Utility::checkLocalInfo("ads", local_info);
   retry_timer_ = dispatcher.createTimer([this]() -> void { establishNewStream(); });
   backoff_strategy_ = std::make_unique<JitteredBackOffStrategy>(RETRY_INITIAL_DELAY_MS,
@@ -111,9 +111,11 @@ GrpcMuxWatchPtr GrpcMuxImpl::subscribe(const std::string& type_url,
   // TODO(gsagula): move TokenBucketImpl params to a config.
   if (!api_state_[type_url].subscribed_) {
     // Bucket contains 100 tokens maximum and refills at 5 tokens/sec.
-    api_state_[type_url].limit_request_ = std::make_unique<TokenBucketImpl>(100, 5, time_source_);
+    api_state_[type_url].limit_request_ =
+        std::make_unique<TokenBucketImpl>(100, time_source_.monotonic(), 5);
     // Bucket contains 1 token maximum and refills 1 token on every ~5 seconds.
-    api_state_[type_url].limit_log_ = std::make_unique<TokenBucketImpl>(1, 0.2, time_source_);
+    api_state_[type_url].limit_log_ =
+        std::make_unique<TokenBucketImpl>(1, time_source_.monotonic(), 0.2);
     api_state_[type_url].request_.set_type_url(type_url);
     api_state_[type_url].request_.mutable_node()->MergeFrom(local_info_.node());
     api_state_[type_url].subscribed_ = true;

source/common/config/grpc_mux_impl.h

@@ -26,8 +26,7 @@ class GrpcMuxImpl : public GrpcMux,
 public:
   GrpcMuxImpl(const LocalInfo::LocalInfo& local_info, Grpc::AsyncClientPtr async_client,
               Event::Dispatcher& dispatcher, const Protobuf::MethodDescriptor& service_method,
-              Runtime::RandomGenerator& random,
-              MonotonicTimeSource& time_source = ProdMonotonicTimeSource::instance_);
+              Runtime::RandomGenerator& random);
   ~GrpcMuxImpl();
 
   void start() override;
@@ -104,7 +103,7 @@ class GrpcMuxImpl : public GrpcMux,
   std::list<std::string> subscriptions_;
   Event::TimerPtr retry_timer_;
   Runtime::RandomGenerator& random_;
-  MonotonicTimeSource& time_source_;
+  TimeSource time_source_;
   BackOffStrategyPtr backoff_strategy_;
 };
 

source/common/event/dispatched_thread.h

@@ -38,7 +38,7 @@ namespace Event {
  */
 class DispatchedThreadImpl : Logger::Loggable<Envoy::Logger::Id::main> {
 public:
-  DispatchedThreadImpl() : dispatcher_(new DispatcherImpl()) {}
+  DispatchedThreadImpl(TimeSource& time_source) : dispatcher_(new DispatcherImpl(time_source)) {}
 
   /**
    * Start the thread.

source/common/event/dispatcher_impl.cc

@@ -24,14 +24,14 @@
 namespace Envoy {
 namespace Event {
 
-DispatcherImpl::DispatcherImpl()
-    : DispatcherImpl(Buffer::WatermarkFactoryPtr{new Buffer::WatermarkBufferFactory}) {
+DispatcherImpl::DispatcherImpl(TimeSource& time_source)
+    : DispatcherImpl(time_source, Buffer::WatermarkFactoryPtr{new Buffer::WatermarkBufferFactory}) {
   // The dispatcher won't work as expected if libevent hasn't been configured to use threads.
   RELEASE_ASSERT(Libevent::Global::initialized(), "");
 }
 
-DispatcherImpl::DispatcherImpl(Buffer::WatermarkFactoryPtr&& factory)
-    : buffer_factory_(std::move(factory)), base_(event_base_new()),
+DispatcherImpl::DispatcherImpl(TimeSource& time_source, Buffer::WatermarkFactoryPtr&& factory)
+    : time_source_(time_source), buffer_factory_(std::move(factory)), base_(event_base_new()),
       deferred_delete_timer_(createTimer([this]() -> void { clearDeferredDeleteList(); })),
       post_timer_(createTimer([this]() -> void { runPostCallbacks(); })),
       current_to_delete_(&to_delete_1_) {

source/common/event/dispatcher_impl.h

@@ -5,6 +5,7 @@
 #include <list>
 #include <vector>
 
+#include "envoy/common/time.h"
 #include "envoy/event/deferred_deletable.h"
 #include "envoy/event/dispatcher.h"
 #include "envoy/network/connection_handler.h"
@@ -21,8 +22,8 @@ namespace Event {
  */
 class DispatcherImpl : Logger::Loggable<Logger::Id::main>, public Dispatcher {
 public:
-  DispatcherImpl();
-  DispatcherImpl(Buffer::WatermarkFactoryPtr&& factory);
+  explicit DispatcherImpl(TimeSource& time_source);
+  DispatcherImpl(TimeSource& time_source, Buffer::WatermarkFactoryPtr&& factory);
   ~DispatcherImpl();
 
   /**
@@ -31,6 +32,7 @@ class DispatcherImpl : Logger::Loggable<Logger::Id::main>, public Dispatcher {
   event_base& base() { return *base_; }
 
   // Event::Dispatcher
+  TimeSource& timeSource() override { return time_source_; }
   void clearDeferredDeleteList() override;
   Network::ConnectionPtr
   createServerConnection(Network::ConnectionSocketPtr&& socket,
@@ -66,6 +68,7 @@ class DispatcherImpl : Logger::Loggable<Logger::Id::main>, public Dispatcher {
     return run_tid_ == 0 || run_tid_ == Thread::Thread::currentThreadId();
   }
 
+  TimeSource time_source_;
   Thread::ThreadId run_tid_{};
   Buffer::WatermarkFactoryPtr buffer_factory_;
   Libevent::BasePtr base_;