Schedule option

api/client/options.proto

@@ -3,6 +3,7 @@ syntax = "proto3";
 package nighthawk.client;
 
 import "google/protobuf/duration.proto";
+import "google/protobuf/timestamp.proto";
 import "google/protobuf/wrappers.proto";
 import "envoy/config/core/v3/base.proto";
 import "envoy/config/metrics/v3/stats.proto";
@@ -219,4 +220,7 @@ message CommandLineOptions {
   // "emit_previous_request_delta_in_response_header" to record elapsed time between request
   // arrivals.
   google.protobuf.StringValue latency_response_header_name = 36;
+  // Provide an execution starting date and time. Optional, any value specified must be in the
+  // future.
+  google.protobuf.Timestamp schedule = 105;
 }

api/client/output.proto

@@ -49,6 +49,7 @@ message Result {
   repeated Statistic statistics = 2;
   repeated Counter counters = 3;
   google.protobuf.Duration execution_duration = 4;
+  google.protobuf.Timestamp execution_start = 5;
 }
 
 message Output {

include/nighthawk/client/options.h

@@ -6,6 +6,7 @@
 #include <string>
 
 #include "envoy/common/pure.h"
+#include "envoy/common/time.h"
 #include "envoy/config/cluster/v3/cluster.pb.h"
 #include "envoy/config/core/v3/base.pb.h"
 #include "envoy/config/metrics/v3/stats.pb.h"
@@ -74,7 +75,7 @@ class Options {
   virtual std::vector<envoy::config::metrics::v3::StatsSink> statsSinks() const PURE;
   virtual uint32_t statsFlushInterval() const PURE;
   virtual std::string responseHeaderWithLatencyInput() const PURE;
-
+  virtual absl::optional<Envoy::SystemTime> schedule() const PURE;
   /**
    * Converts an Options instance to an equivalent CommandLineOptions instance in terms of option
    * values.

include/nighthawk/client/output_collector.h

@@ -3,9 +3,12 @@
 #include <memory>
 
 #include "envoy/common/pure.h"
+#include "envoy/common/time.h"
 
 #include "nighthawk/common/statistic.h"
 
+#include "absl/types/optional.h"
+
 namespace Nighthawk {
 namespace Client {
 
@@ -26,7 +29,8 @@ class OutputCollector {
    */
   virtual void addResult(absl::string_view name, const std::vector<StatisticPtr>& statistics,
                          const std::map<std::string, uint64_t>& counters,
-                         const std::chrono::nanoseconds execution_duration) PURE;
+                         const std::chrono::nanoseconds execution_duration,
+                         const absl::optional<Envoy::SystemTime>& first_acquisition_time) PURE;
   /**
    * Directly sets the output value.
    *

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;
 };
 
 /**

include/nighthawk/common/rate_limiter.h

@@ -33,6 +33,12 @@ class RateLimiter {
    * @return Envoy::TimeSource& time_source used to track time.
    */
   virtual Envoy::TimeSource& timeSource() PURE;
+
+  /**
+   * @return absl::optional<Envoy::SystemTime> Time of the first acquisition, if any.
+   */
+  virtual absl::optional<Envoy::SystemTime> firstAcquisitionTime() const PURE;
+
   /**
    * @return std::chrono::nanoseconds elapsed since the first call to tryAcquireOne(). Used by some
    * rate limiter implementations to compute acquisition rate.

include/nighthawk/common/sequencer.h

@@ -7,6 +7,7 @@
 #include "envoy/common/pure.h"
 
 #include "nighthawk/common/operation_callback.h"
+#include "nighthawk/common/rate_limiter.h"
 #include "nighthawk/common/statistic.h"
 
 namespace Nighthawk {
@@ -35,6 +36,11 @@ class Sequencer {
    */
   virtual std::chrono::nanoseconds executionDuration() const PURE;
 
+  /**
+   * @return RateLimiter& reference to the rate limiter associated to this sequencer.
+   */
+  virtual const RateLimiter& rate_limiter() const PURE;
+
   /**
    * @return double an up-to-date completions per second rate.
    */

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/options_impl.cc

@@ -651,7 +651,12 @@ OptionsImpl::OptionsImpl(const nighthawk::client::CommandLineOptions& options) {
   std::copy(options.labels().begin(), options.labels().end(), std::back_inserter(labels_));
   latency_response_header_name_ = PROTOBUF_GET_WRAPPED_OR_DEFAULT(
       options, latency_response_header_name, latency_response_header_name_);
-
+  if (options.has_schedule()) {
+    const auto elapsed_since_epoch = std::chrono::nanoseconds(options.schedule().nanos()) +
+                                     std::chrono::seconds(options.schedule().seconds());
+    schedule_ =
+        Envoy::SystemTime(std::chrono::time_point<std::chrono::system_clock>(elapsed_since_epoch));
+  }
   validate();
 }
 
@@ -828,6 +833,11 @@ CommandLineOptionsPtr OptionsImpl::toCommandLineOptionsInternal() const {
   command_line_options->mutable_stats_flush_interval()->set_value(stats_flush_interval_);
   command_line_options->mutable_latency_response_header_name()->set_value(
       latency_response_header_name_);
+  if (schedule_.has_value()) {
+    *(command_line_options->mutable_schedule()) =
+        Envoy::ProtobufUtil::TimeUtil::NanosecondsToTimestamp(
+            schedule_.value().time_since_epoch().count());
+  }
   return command_line_options;
 }
 

source/client/options_impl.h

@@ -93,6 +93,7 @@ class OptionsImpl : public Options, public Envoy::Logger::Loggable<Envoy::Logger
   std::string responseHeaderWithLatencyInput() const override {
     return latency_response_header_name_;
   };
+  absl::optional<Envoy::SystemTime> schedule() const override { return schedule_; }
 
 private:
   void parsePredicates(const TCLAP::MultiArg<std::string>& arg,
@@ -149,6 +150,7 @@ class OptionsImpl : public Options, public Envoy::Logger::Loggable<Envoy::Logger
   std::vector<envoy::config::metrics::v3::StatsSink> stats_sinks_;
   uint32_t stats_flush_interval_{5};
   std::string latency_response_header_name_;
+  absl::optional<Envoy::SystemTime> schedule_;
 };
 
 } // namespace Client

source/client/output_collector_impl.cc

@@ -23,12 +23,19 @@ OutputCollectorImpl::OutputCollectorImpl(Envoy::TimeSource& time_source, const O
 
 nighthawk::client::Output OutputCollectorImpl::toProto() const { return output_; }
 
-void OutputCollectorImpl::addResult(absl::string_view name,
-                                    const std::vector<StatisticPtr>& statistics,
-                                    const std::map<std::string, uint64_t>& counters,
-                                    const std::chrono::nanoseconds execution_duration) {
+void OutputCollectorImpl::addResult(
+    absl::string_view name, const std::vector<StatisticPtr>& statistics,
+    const std::map<std::string, uint64_t>& counters,
+    const std::chrono::nanoseconds execution_duration,
+    const absl::optional<Envoy::SystemTime>& first_acquisition_time) {
   auto result = output_.add_results();
   result->set_name(name.data(), name.size());
+  if (first_acquisition_time.has_value()) {
+    *(result->mutable_execution_start()) = Envoy::Protobuf::util::TimeUtil::NanosecondsToTimestamp(
+        std::chrono::duration_cast<std::chrono::nanoseconds>(
+            first_acquisition_time.value().time_since_epoch())
+            .count());
+  }
   for (auto& statistic : statistics) {
     // TODO(#292): Looking at if the statistic id ends with "_size" to determine how it should be
     // serialized is kind of hacky. Maybe we should have a lookup table of sorts, to determine how

source/client/output_collector_impl.h

@@ -20,7 +20,8 @@ class OutputCollectorImpl : public OutputCollector {
 
   void addResult(absl::string_view name, const std::vector<StatisticPtr>& statistics,
                  const std::map<std::string, uint64_t>& counters,
-                 const std::chrono::nanoseconds execution_duration) override;
+                 const std::chrono::nanoseconds execution_duration,
+                 const absl::optional<Envoy::SystemTime>& first_acquisition_time) override;
   void setOutput(const nighthawk::client::Output& output) override { output_ = output; }
 
   nighthawk::client::Output toProto() const override;

source/client/process_impl.cc

@@ -29,6 +29,7 @@
 #include "external/envoy/source/server/server.h"
 
 #include "absl/strings/str_replace.h"
+#include "absl/types/optional.h"
 
 // TODO(oschaaf): See if we can leverage a static module registration like Envoy does to avoid the
 // ifdefs in this file.
@@ -164,7 +165,8 @@ bool ProcessImpl::requestExecutionCancellation() {
   return true;
 }
 
-void ProcessImpl::createWorkers(const uint32_t concurrency) {
+void ProcessImpl::createWorkers(const uint32_t concurrency,
+                                const absl::optional<Envoy::SystemTime>& schedule) {
   // TODO(oschaaf): Expose kMinimalDelay in configuration.
   const std::chrono::milliseconds kMinimalWorkerDelay = 500ms + (concurrency * 50ms);
   ASSERT(workers_.empty());
@@ -179,7 +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 = 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;
@@ -445,7 +450,13 @@ void ProcessImpl::setupStatsSinks(const envoy::config::bootstrap::v3::Bootstrap&
 }
 
 bool ProcessImpl::runInternal(OutputCollector& collector, const std::vector<UriPtr>& uris,
-                              const UriPtr& request_source_uri, const UriPtr& tracing_uri) {
+                              const UriPtr& request_source_uri, const UriPtr& tracing_uri,
+                              const absl::optional<Envoy::SystemTime>& schedule) {
+  const Envoy::SystemTime now = time_system_.systemTime();
+  if (schedule.value_or(now) < now) {
+    ENVOY_LOG(error, "Scheduled execution date already transpired.");
+    return false;
+  }
   {
     auto guard = std::make_unique<Envoy::Thread::LockGuard>(workers_lock_);
     if (cancelled_) {
@@ -461,7 +472,7 @@ bool ProcessImpl::runInternal(OutputCollector& collector, const std::vector<UriP
       store_root_.setTagProducer(Envoy::Config::Utility::createTagProducer(bootstrap));
     }
 
-    createWorkers(number_of_workers);
+    createWorkers(number_of_workers, schedule);
     tls_.registerThread(*dispatcher_, true);
     store_root_.initializeThreading(*dispatcher_, tls_);
     runtime_singleton_ = std::make_unique<Envoy::Runtime::ScopedLoaderSingleton>(
@@ -522,15 +533,26 @@ bool ProcessImpl::runInternal(OutputCollector& collector, const std::vector<UriP
 
   int i = 0;
   std::chrono::nanoseconds total_execution_duration = 0ns;
+  absl::optional<Envoy::SystemTime> first_acquisition_time = absl::nullopt;
+
   for (auto& worker : workers_) {
     auto sequencer_execution_duration = worker->phase().sequencer().executionDuration();
+    absl::optional<Envoy::SystemTime> worker_first_acquisition_time =
+        worker->phase().sequencer().rate_limiter().firstAcquisitionTime();
+    if (worker_first_acquisition_time.has_value()) {
+      first_acquisition_time =
+          first_acquisition_time.has_value()
+              ? std::min(first_acquisition_time.value(), worker_first_acquisition_time.value())
+              : worker_first_acquisition_time.value();
+    }
     // We don't write per-worker results if we only have a single worker, because the global
     // results will be precisely the same.
     if (workers_.size() > 1) {
       StatisticFactoryImpl statistic_factory(options_);
       collector.addResult(fmt::format("worker_{}", i),
                           vectorizeStatisticPtrMap(worker->statistics()),
-                          worker->threadLocalCounterValues(), sequencer_execution_duration);
+                          worker->threadLocalCounterValues(), sequencer_execution_duration,
+                          worker_first_acquisition_time);
     }
     total_execution_duration += sequencer_execution_duration;
     i++;
@@ -545,7 +567,7 @@ bool ProcessImpl::runInternal(OutputCollector& collector, const std::vector<UriP
       store_root_, [](absl::string_view, uint64_t value) { return value > 0; });
   StatisticFactoryImpl statistic_factory(options_);
   collector.addResult("global", mergeWorkerStatistics(workers_), counters,
-                      total_execution_duration / workers_.size());
+                      total_execution_duration / workers_.size(), first_acquisition_time);
   return counters.find("sequencer.failed_terminations") == counters.end();
 }
 
@@ -585,7 +607,7 @@ bool ProcessImpl::run(OutputCollector& collector) {
   }
 
   try {
-    return runInternal(collector, uris, request_source_uri, tracing_uri);
+    return runInternal(collector, uris, request_source_uri, tracing_uri, options_.schedule());
   } catch (Envoy::EnvoyException& ex) {
     ENVOY_LOG(error, "Fatal exception: {}", ex.what());
     throw;

source/client/process_impl.h

@@ -109,7 +109,7 @@ class ProcessImpl : public Process, public Envoy::Logger::Loggable<Envoy::Logger
    *
    * @param concurrency the amount of workers that should be created.
    */
-  void createWorkers(const uint32_t concurrency);
+  void createWorkers(const uint32_t concurrency, const absl::optional<Envoy::SystemTime>& schedule);
   std::vector<StatisticPtr> vectorizeStatisticPtrMap(const StatisticPtrMap& statistics) const;
   std::vector<StatisticPtr>
   mergeWorkerStatistics(const std::vector<ClientWorkerPtr>& workers) const;
@@ -124,7 +124,8 @@ class ProcessImpl : public Process, public Envoy::Logger::Loggable<Envoy::Logger
   void setupStatsSinks(const envoy::config::bootstrap::v3::Bootstrap& bootstrap,
                        std::list<std::unique_ptr<Envoy::Stats::Sink>>& stats_sinks);
   bool runInternal(OutputCollector& collector, const std::vector<UriPtr>& uris,
-                   const UriPtr& request_source_uri, const UriPtr& tracing_uri);
+                   const UriPtr& request_source_uri, const UriPtr& tracing_uri,
+                   const absl::optional<Envoy::SystemTime>& schedule);
 
   std::shared_ptr<Envoy::ProcessWide> process_wide_;
   Envoy::PlatformImpl platform_impl_;