Changes to address envoyproxy#569

Signed-off-by: Otto van der Schaaf <oschaaf@we-amp.com>

include/nighthawk/common/factories.h

@@ -24,7 +24,7 @@ class SequencerFactory {
                               const SequencerTarget& sequencer_target,
                               TerminationPredicatePtr&& termination_predicate,
                               Envoy::Stats::Scope& scope,
-                              const Envoy::SystemTime scheduled_starting_time) const PURE;
+                              const Envoy::MonotonicTime scheduled_starting_time) const PURE;
 };
 
 class StatisticFactory {
@@ -46,7 +46,7 @@ class TerminationPredicateFactory {
   virtual ~TerminationPredicateFactory() = default;
   virtual TerminationPredicatePtr
   create(Envoy::TimeSource& time_source, Envoy::Stats::Scope& scope,
-         const Envoy::SystemTime scheduled_starting_time) const PURE;
+         const Envoy::MonotonicTime scheduled_starting_time) const PURE;
 };
 
 /**

source/client/client_worker_impl.cc

@@ -19,7 +19,7 @@ ClientWorkerImpl::ClientWorkerImpl(Envoy::Api::Api& api, Envoy::ThreadLocal::Ins
                                    const SequencerFactory& sequencer_factory,
                                    const RequestSourceFactory& request_generator_factory,
                                    Envoy::Stats::Store& store, const int worker_number,
-                                   const Envoy::SystemTime starting_time,
+                                   const Envoy::MonotonicTime starting_time,
                                    Envoy::Tracing::HttpTracerSharedPtr& http_tracer,
                                    const HardCodedWarmupStyle hardcoded_warmup_style)
     : WorkerImpl(api, tls, store),

source/client/client_worker_impl.h

@@ -33,7 +33,7 @@ class ClientWorkerImpl : public WorkerImpl, virtual public ClientWorker {
                    const SequencerFactory& sequencer_factory,
                    const RequestSourceFactory& request_generator_factory,
                    Envoy::Stats::Store& store, const int worker_number,
-                   const Envoy::SystemTime starting_time,
+                   const Envoy::MonotonicTime starting_time,
                    Envoy::Tracing::HttpTracerSharedPtr& http_tracer,
                    const HardCodedWarmupStyle hardcoded_warmup_style);
   StatisticPtrMap statistics() const override;

source/client/factories_impl.cc

@@ -65,12 +65,10 @@ BenchmarkClientPtr BenchmarkClientFactoryImpl::create(
 SequencerFactoryImpl::SequencerFactoryImpl(const Options& options)
     : OptionBasedFactoryImpl(options) {}
 
-SequencerPtr SequencerFactoryImpl::create(Envoy::TimeSource& time_source,
-                                          Envoy::Event::Dispatcher& dispatcher,
-                                          const SequencerTarget& sequencer_target,
-                                          TerminationPredicatePtr&& termination_predicate,
-                                          Envoy::Stats::Scope& scope,
-                                          const Envoy::SystemTime scheduled_starting_time) const {
+SequencerPtr SequencerFactoryImpl::create(
+    Envoy::TimeSource& time_source, Envoy::Event::Dispatcher& dispatcher,
+    const SequencerTarget& sequencer_target, TerminationPredicatePtr&& termination_predicate,
+    Envoy::Stats::Scope& scope, const Envoy::MonotonicTime scheduled_starting_time) const {
   StatisticFactoryImpl statistic_factory(options_);
   Frequency frequency(options_.requestsPerSecond());
   RateLimiterPtr rate_limiter = std::make_unique<ScheduledStartingRateLimiter>(
@@ -211,7 +209,7 @@ TerminationPredicateFactoryImpl::TerminationPredicateFactoryImpl(const Options&
 
 TerminationPredicatePtr
 TerminationPredicateFactoryImpl::create(Envoy::TimeSource& time_source, Envoy::Stats::Scope& scope,
-                                        const Envoy::SystemTime scheduled_starting_time) const {
+                                        const Envoy::MonotonicTime scheduled_starting_time) const {
   // We'll always link a predicate which checks for requests to cancel.
   TerminationPredicatePtr root_predicate =
       std::make_unique<StatsCounterAbsoluteThresholdTerminationPredicateImpl>(

source/client/factories_impl.h

@@ -44,7 +44,7 @@ class SequencerFactoryImpl : public OptionBasedFactoryImpl, public SequencerFact
   SequencerPtr create(Envoy::TimeSource& time_source, Envoy::Event::Dispatcher& dispatcher,
                       const SequencerTarget& sequencer_target,
                       TerminationPredicatePtr&& termination_predicate, Envoy::Stats::Scope& scope,
-                      const Envoy::SystemTime scheduled_starting_time) const override;
+                      const Envoy::MonotonicTime scheduled_starting_time) const override;
 };
 
 class StatisticFactoryImpl : public OptionBasedFactoryImpl, public StatisticFactory {
@@ -93,7 +93,7 @@ class TerminationPredicateFactoryImpl : public OptionBasedFactoryImpl,
 public:
   TerminationPredicateFactoryImpl(const Options& options);
   TerminationPredicatePtr create(Envoy::TimeSource& time_source, Envoy::Stats::Scope& scope,
-                                 const Envoy::SystemTime scheduled_starting_time) const override;
+                                 const Envoy::MonotonicTime scheduled_starting_time) const override;
   TerminationPredicate* linkConfiguredPredicates(
       TerminationPredicate& last_predicate, const TerminationPredicateMap& predicates,
       const TerminationPredicate::Status termination_status, Envoy::Stats::Scope& scope) const;

source/client/process_impl.cc

@@ -181,8 +181,10 @@ void ProcessImpl::createWorkers(const uint32_t concurrency,
   // TODO(oschaaf): Arguably, this ought to be the job of a rate limiter with awareness of the
   // global status quo, which we do not have right now. This has been noted in the
   // track-for-future issue.
-  const auto first_worker_start =
-      schedule.value_or(time_system_.systemTime() + kMinimalWorkerDelay);
+  const Envoy::MonotonicTime monotonic_now = time_system_.monotonicTime();
+  const std::chrono::nanoseconds offset =
+      schedule.has_value() ? schedule.value() - time_system_.systemTime() : kMinimalWorkerDelay;
+  const Envoy::MonotonicTime first_worker_start = monotonic_now + offset;
   const double inter_worker_delay_usec =
       (1. / options_.requestsPerSecond()) * 1000000 / concurrency;
   int worker_number = 0;

source/common/rate_limiter_impl.cc

@@ -53,16 +53,16 @@ void BurstingRateLimiter::releaseOne() {
 }
 
 ScheduledStartingRateLimiter::ScheduledStartingRateLimiter(
-    RateLimiterPtr&& rate_limiter, const Envoy::SystemTime scheduled_starting_time)
+    RateLimiterPtr&& rate_limiter, const Envoy::MonotonicTime scheduled_starting_time)
     : ForwardingRateLimiterImpl(std::move(rate_limiter)),
       scheduled_starting_time_(scheduled_starting_time) {
-  if (timeSource().systemTime() >= scheduled_starting_time_) {
+  if (timeSource().monotonicTime() >= scheduled_starting_time_) {
     ENVOY_LOG(error, "Scheduled starting time exceeded. This may cause unintended bursty traffic.");
   }
 }
 
 bool ScheduledStartingRateLimiter::tryAcquireOne() {
-  if (timeSource().systemTime() < scheduled_starting_time_) {
+  if (timeSource().monotonicTime() < scheduled_starting_time_) {
     aquisition_attempted_ = true;
     return false;
   }
@@ -76,7 +76,7 @@ bool ScheduledStartingRateLimiter::tryAcquireOne() {
 }
 
 void ScheduledStartingRateLimiter::releaseOne() {
-  if (timeSource().systemTime() < scheduled_starting_time_) {
+  if (timeSource().monotonicTime() < scheduled_starting_time_) {
     throw NighthawkException("Unexpected call to releaseOne()");
   }
   return rate_limiter_->releaseOne();

source/common/rate_limiter_impl.h

@@ -134,12 +134,12 @@ class ScheduledStartingRateLimiter : public ForwardingRateLimiterImpl,
    * @param scheduled_starting_time The starting time
    */
   ScheduledStartingRateLimiter(RateLimiterPtr&& rate_limiter,
-                               const Envoy::SystemTime scheduled_starting_time);
+                               const Envoy::MonotonicTime scheduled_starting_time);
   bool tryAcquireOne() override;
   void releaseOne() override;
 
 private:
-  const Envoy::SystemTime scheduled_starting_time_;
+  const Envoy::MonotonicTime scheduled_starting_time_;
   bool aquisition_attempted_{false};
 };
 

source/common/termination_predicate_impl.cc

@@ -16,8 +16,8 @@ TerminationPredicate::Status TerminationPredicateBaseImpl::evaluateChain() {
 }
 
 TerminationPredicate::Status DurationTerminationPredicateImpl::evaluate() {
-  return time_source_.systemTime() - start_ > duration_ ? TerminationPredicate::Status::TERMINATE
-                                                        : TerminationPredicate::Status::PROCEED;
+  return time_source_.monotonicTime() - start_ > duration_ ? TerminationPredicate::Status::TERMINATE
+                                                           : TerminationPredicate::Status::PROCEED;
 }
 
 TerminationPredicate::Status StatsCounterAbsoluteThresholdTerminationPredicateImpl::evaluate() {

source/common/termination_predicate_impl.h

@@ -35,13 +35,13 @@ class DurationTerminationPredicateImpl : public TerminationPredicateBaseImpl {
 public:
   DurationTerminationPredicateImpl(Envoy::TimeSource& time_source,
                                    std::chrono::microseconds duration,
-                                   const Envoy::SystemTime start)
+                                   const Envoy::MonotonicTime start)
       : time_source_(time_source), start_(start), duration_(duration) {}
   TerminationPredicate::Status evaluate() override;
 
 private:
   Envoy::TimeSource& time_source_;
-  const Envoy::SystemTime start_;
+  const Envoy::MonotonicTime start_;
   std::chrono::microseconds duration_;
 };
 

test/client_worker_test.cc

@@ -118,7 +118,7 @@ TEST_F(ClientWorkerTest, BasicTest) {
   auto worker = std::make_unique<ClientWorkerImpl>(
       *api_, tls_, cluster_manager_ptr_, benchmark_client_factory_, termination_predicate_factory_,
       sequencer_factory_, request_generator_factory_, store_, worker_number,
-      time_system_.systemTime(), http_tracer_, ClientWorkerImpl::HardCodedWarmupStyle::ON);
+      time_system_.monotonicTime(), http_tracer_, ClientWorkerImpl::HardCodedWarmupStyle::ON);
 
   worker->start();
   worker->waitForCompletion();

test/factories_test.cc

@@ -198,7 +198,7 @@ class SequencerFactoryTest
     };
     auto sequencer = factory.create(api_->timeSource(), dispatcher_, dummy_sequencer_target,
                                     std::make_unique<MockTerminationPredicate>(), stats_store_,
-                                    time_system.systemTime() + 10ms);
+                                    time_system.monotonicTime() + 10ms);
     EXPECT_NE(nullptr, sequencer.get());
   }
 };

test/mocks/common/mock_sequencer_factory.h

@@ -14,7 +14,7 @@ class MockSequencerFactory : public SequencerFactory {
                                           const SequencerTarget& sequencer_target,
                                           TerminationPredicatePtr&& termination_predicate,
                                           Envoy::Stats::Scope& scope,
-                                          const Envoy::SystemTime scheduled_starting_time));
+                                          const Envoy::MonotonicTime scheduled_starting_time));
 };
 
 } // namespace Nighthawk

test/mocks/common/mock_termination_predicate_factory.h

@@ -12,7 +12,7 @@ class MockTerminationPredicateFactory : public TerminationPredicateFactory {
   MOCK_CONST_METHOD3(create,
                      TerminationPredicatePtr(Envoy::TimeSource& time_source,
                                              Envoy::Stats::Scope& scope,
-                                             const Envoy::SystemTime scheduled_starting_time));
+                                             const Envoy::MonotonicTime scheduled_starting_time));
 };
 
 } // namespace Nighthawk

test/rate_limiter_test.cc

@@ -76,7 +76,8 @@ TEST_F(RateLimiterTest, ScheduledStartingRateLimiterTest) {
   // scheduled delay. This should be business as usual from a functional perspective, but internally
   // this rate limiter specializes on this case to log a warning message, and we want to cover that.
   for (const bool starting_late : std::vector<bool>{false, true}) {
-    const Envoy::SystemTime scheduled_starting_time = time_system.systemTime() + schedule_delay;
+    const Envoy::MonotonicTime scheduled_starting_time =
+        time_system.monotonicTime() + schedule_delay;
     std::unique_ptr<MockRateLimiter> mock_rate_limiter = std::make_unique<MockRateLimiter>();
     MockRateLimiter& unsafe_mock_rate_limiter = *mock_rate_limiter;
     InSequence s;
@@ -95,7 +96,7 @@ TEST_F(RateLimiterTest, ScheduledStartingRateLimiterTest) {
     }
 
     // We should expect zero releases until it is time to start.
-    while (time_system.systemTime() < scheduled_starting_time) {
+    while (time_system.monotonicTime() < scheduled_starting_time) {
       EXPECT_FALSE(rate_limiter->tryAcquireOne());
       time_system.advanceTimeWait(1ms);
     }
@@ -108,8 +109,8 @@ TEST_F(RateLimiterTest, ScheduledStartingRateLimiterTest) {
 TEST_F(RateLimiterTest, ScheduledStartingRateLimiterTestBadArgs) {
   Envoy::Event::SimulatedTimeSystem time_system;
   // Verify we enforce future-only scheduling.
-  for (const auto& timing :
-       std::vector<Envoy::SystemTime>{time_system.systemTime(), time_system.systemTime() - 10ms}) {
+  for (const auto& timing : std::vector<Envoy::MonotonicTime>{time_system.monotonicTime(),
+                                                              time_system.monotonicTime() - 10ms}) {
     std::unique_ptr<MockRateLimiter> mock_rate_limiter = std::make_unique<MockRateLimiter>();
     MockRateLimiter& unsafe_mock_rate_limiter = *mock_rate_limiter;
     EXPECT_CALL(unsafe_mock_rate_limiter, timeSource)

test/termination_predicate_test.cc

@@ -25,7 +25,7 @@ class TerminationPredicateTest : public Test {
 
 TEST_F(TerminationPredicateTest, DurationTerminationPredicateImplTest) {
   const auto duration = 100us;
-  DurationTerminationPredicateImpl pred(time_system, duration, time_system.systemTime());
+  DurationTerminationPredicateImpl pred(time_system, duration, time_system.monotonicTime());
   EXPECT_EQ(pred.evaluate(), TerminationPredicate::Status::PROCEED);
   // move to the edge.
   time_system.advanceTimeWait(duration);