Introduce Phases

include/nighthawk/client/benchmark_client.h

@@ -30,7 +30,7 @@ class BenchmarkClient {
    *
    * @param measure_latencies true iff latencies should be measured.
    */
-  virtual void setMeasureLatencies(bool measure_latencies) PURE;
+  virtual void setShouldMeasureLatencies(bool measure_latencies) PURE;
 
   /**
    * Gets the statistics, keyed by id.
@@ -59,7 +59,7 @@ class BenchmarkClient {
    *
    * @return bool indicating if latency measurement is enabled.
    */
-  virtual bool measureLatencies() const PURE;
+  virtual bool shouldMeasureLatencies() const PURE;
 };
 
 using BenchmarkClientPtr = std::unique_ptr<BenchmarkClient>;

include/nighthawk/client/client_worker.h

@@ -6,7 +6,7 @@
 #include "envoy/stats/store.h"
 
 #include "nighthawk/client/benchmark_client.h"
-#include "nighthawk/common/sequencer.h"
+#include "nighthawk/common/phase.h"
 #include "nighthawk/common/statistic.h"
 #include "nighthawk/common/worker.h"
 
@@ -31,9 +31,9 @@ class ClientWorker : virtual public Worker {
   virtual const std::map<std::string, uint64_t>& thread_local_counter_values() PURE;
 
   /**
-   * @return const Sequencer&
+   * @return const Phase& associated to this worker.
    */
-  virtual const Sequencer& sequencer() const PURE;
+  virtual const Phase& phase() const PURE;
 };
 
 using ClientWorkerPtr = std::unique_ptr<ClientWorker>;

include/nighthawk/client/factories.h

@@ -37,9 +37,10 @@ class SequencerFactory {
 public:
   virtual ~SequencerFactory() = default;
   virtual SequencerPtr create(Envoy::TimeSource& time_source, Envoy::Event::Dispatcher& dispatcher,
-                              Envoy::MonotonicTime start_time, BenchmarkClient& benchmark_client,
-                              TerminationPredicate& termination_predicate,
-                              Envoy::Stats::Scope& scope) const PURE;
+                              BenchmarkClient& benchmark_client,
+                              TerminationPredicatePtr&& termination_predicate,
+                              Envoy::Stats::Scope& scope,
+                              const Envoy::MonotonicTime scheduled_starting_time) const PURE;
 };
 
 class StoreFactory {
@@ -74,8 +75,8 @@ class RequestSourceFactory {
 class TerminationPredicateFactory {
 public:
   virtual ~TerminationPredicateFactory() = default;
-  virtual TerminationPredicatePtr create(Envoy::TimeSource& time_source, Envoy::Stats::Scope& scope,
-                                         Envoy::MonotonicTime time) const PURE;
+  virtual TerminationPredicatePtr create(Envoy::TimeSource& time_source,
+                                         Envoy::Stats::Scope& scope) const PURE;
 };
 
 } // namespace Nighthawk

include/nighthawk/common/BUILD

@@ -13,6 +13,7 @@ envoy_basic_cc_library(
     hdrs = [
         "exception.h",
         "operation_callback.h",
+        "phase.h",
         "platform_util.h",
         "rate_limiter.h",
         "request_source.h",

include/nighthawk/common/phase.h

@@ -0,0 +1,51 @@
+
+#pragma once
+
+#include <memory>
+
+#include "envoy/common/pure.h"
+
+#include "nighthawk/common/sequencer.h"
+
+namespace Nighthawk {
+
+/**
+ * Phase represents a distinct phase of a benchmmark execution, like warmup and cooldown.
+ * A phase is associated to a sequencer, which in turn can be associated to separate termination
+ * and failure predicates as well as its own rate limiter policy. The end of a phase also represents
+ * a natural boundary for reporting a snapshot of stats and latencies associated to the phase.
+ * High level, a worker statically configure a vector of phases, and will transfer the hot
+ * connection pool when transitioning between them. At this time, nothing is stopping us from
+ * dynamically injecting phases later, be it via grpc calls and/or live CLI input.
+ */
+class Phase {
+public:
+  virtual ~Phase() = default;
+
+  /**
+   * @return absl::string_view Contains the id of the phase. Should be unique but that is not
+   * enforced at this time so take care.
+   */
+  virtual absl::string_view id() const PURE;
+
+  /**
+   * @return Sequencer& Sequencer associated to this phase.
+   */
+  virtual Sequencer& sequencer() const PURE;
+
+  /**
+   * @return bool Indicates if latencies should be tracked for this phase.
+   */
+  virtual bool shouldMeasureLatencies() const PURE;
+
+  /**
+   * Runs the sequencer associated to this phase and blocks until completion, which means this phase
+   * has ended as well.
+   * Execution failure can be observed via the sequencer.failed_terminations counter.
+   */
+  virtual void run() const PURE;
+};
+
+using PhasePtr = std::unique_ptr<Phase>;
+
+} // namespace Nighthawk

include/nighthawk/common/termination_predicate.h

@@ -32,6 +32,14 @@ class TerminationPredicate {
    */
   virtual TerminationPredicate& link(TerminationPredicatePtr&& child) PURE;
 
+  /**
+   * Appends a predicate to the last element of the chain.
+   *
+   * @param child the child predicate to link. nullptr is not allowed.
+   * @return the dereferenced input child predicate. For convenience, so calls can be chained.
+   */
+  virtual TerminationPredicate& appendToChain(TerminationPredicatePtr&& child) PURE;
+
   /**
    * Recursively evaluates chain of linked predicates, this instance last.
    * If any linked element returns anything other then PROCEED that status will

source/client/benchmark_client_impl.cc

@@ -157,7 +157,7 @@ bool BenchmarkClientHttpImpl::tryStartRequest(CompletionCallback caller_completi
   std::string x_request_id = generator_.uuid();
   auto stream_decoder = new StreamDecoder(
       dispatcher_, api_.timeSource(), *this, std::move(caller_completion_callback),
-      *connect_statistic_, *response_statistic_, request->header(), measureLatencies(),
+      *connect_statistic_, *response_statistic_, request->header(), shouldMeasureLatencies(),
       content_length, x_request_id, http_tracer_);
   requests_initiated_++;
   pool_ptr->newStream(*stream_decoder, *stream_decoder);

source/client/benchmark_client_impl.h

@@ -135,8 +135,8 @@ class BenchmarkClientHttpImpl : public BenchmarkClient,
   // BenchmarkClient
   void terminate() override;
   StatisticPtrMap statistics() const override;
-  bool measureLatencies() const override { return measure_latencies_; }
-  void setMeasureLatencies(bool measure_latencies) override {
+  bool shouldMeasureLatencies() const override { return measure_latencies_; }
+  void setShouldMeasureLatencies(bool measure_latencies) override {
     measure_latencies_ = measure_latencies;
   }
   bool tryStartRequest(CompletionCallback caller_completion_callback) override;

source/client/client_worker_impl.cc

@@ -2,12 +2,15 @@
 
 #include "external/envoy/source/common/stats/symbol_table_impl.h"
 
-#include "common/request_source_impl.h"
+#include "common/phase_impl.h"
+#include "common/termination_predicate_impl.h"
 #include "common/utility.h"
 
 namespace Nighthawk {
 namespace Client {
 
+using namespace std::chrono_literals;
+
 ClientWorkerImpl::ClientWorkerImpl(Envoy::Api::Api& api, Envoy::ThreadLocal::Instance& tls,
                                    Envoy::Upstream::ClusterManagerPtr& cluster_manager,
                                    const BenchmarkClientFactory& benchmark_client_factory,
@@ -17,20 +20,23 @@ ClientWorkerImpl::ClientWorkerImpl(Envoy::Api::Api& api, Envoy::ThreadLocal::Ins
                                    Envoy::Stats::Store& store, const int worker_number,
                                    const Envoy::MonotonicTime starting_time,
                                    Envoy::Tracing::HttpTracerPtr& http_tracer)
-    : WorkerImpl(api, tls, store), worker_scope_(store_.createScope("cluster.")),
+    : WorkerImpl(api, tls, store), termination_predicate_factory_(termination_predicate_factory),
+      sequencer_factory_(sequencer_factory), worker_scope_(store_.createScope("cluster.")),
       worker_number_scope_(worker_scope_->createScope(fmt::format("{}.", worker_number))),
-      worker_number_(worker_number), starting_time_(starting_time), http_tracer_(http_tracer),
+      worker_number_(worker_number), http_tracer_(http_tracer),
       request_generator_(
           request_generator_factory.create(cluster_manager, *dispatcher_, *worker_number_scope_,
                                            fmt::format("{}.requestsource", worker_number))),
       benchmark_client_(benchmark_client_factory.create(
           api, *dispatcher_, *worker_number_scope_, cluster_manager, http_tracer_,
           fmt::format("{}", worker_number), *request_generator_)),
-      termination_predicate_(
-          termination_predicate_factory.create(time_source_, *worker_number_scope_, starting_time)),
-      sequencer_(sequencer_factory.create(time_source_, *dispatcher_, starting_time,
-                                          *benchmark_client_, *termination_predicate_,
-                                          *worker_number_scope_)) {}
+      phase_(std::make_unique<PhaseImpl>(
+          "main",
+          sequencer_factory_.create(
+              time_source_, *dispatcher_, *benchmark_client_,
+              termination_predicate_factory_.create(time_source_, *worker_number_scope_),
+              *worker_number_scope_, starting_time),
+          true)) {}
 
 void ClientWorkerImpl::simpleWarmup() {
   ENVOY_LOG(debug, "> worker {}: warmup start.", worker_number_);
@@ -43,11 +49,12 @@ void ClientWorkerImpl::simpleWarmup() {
 }
 
 void ClientWorkerImpl::work() {
+  benchmark_client_->setShouldMeasureLatencies(false);
   request_generator_->initOnThread();
   simpleWarmup();
-  benchmark_client_->setMeasureLatencies(true);
-  sequencer_->start();
-  sequencer_->waitForCompletion();
+  benchmark_client_->setShouldMeasureLatencies(phase_->shouldMeasureLatencies());
+  phase_->run();
+
   // Save a final snapshot of the worker-specific counter accumulations before
   // we exit the thread.
   for (const auto& stat : store_.counters()) {
@@ -72,7 +79,8 @@ void ClientWorkerImpl::shutdownThread() { benchmark_client_->terminate(); }
 StatisticPtrMap ClientWorkerImpl::statistics() const {
   StatisticPtrMap statistics;
   StatisticPtrMap s1 = benchmark_client_->statistics();
-  StatisticPtrMap s2 = sequencer_->statistics();
+  Sequencer& sequencer = phase_->sequencer();
+  StatisticPtrMap s2 = sequencer.statistics();
   statistics.insert(s1.begin(), s1.end());
   statistics.insert(s2.begin(), s2.end());
   return statistics;

source/client/client_worker_impl.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include <vector>
+
 #include "envoy/api/api.h"
 #include "envoy/event/dispatcher.h"
 #include "envoy/stats/store.h"
@@ -9,6 +11,7 @@
 #include "nighthawk/client/benchmark_client.h"
 #include "nighthawk/client/client_worker.h"
 #include "nighthawk/client/factories.h"
+#include "nighthawk/common/phase.h"
 #include "nighthawk/common/request_source.h"
 #include "nighthawk/common/sequencer.h"
 #include "nighthawk/common/termination_predicate.h"
@@ -34,23 +37,25 @@ class ClientWorkerImpl : public WorkerImpl, virtual public ClientWorker {
   const std::map<std::string, uint64_t>& thread_local_counter_values() override {
     return thread_local_counter_values_;
   }
-  const Sequencer& sequencer() const override { return *sequencer_; }
+
+  const Phase& phase() const override { return *phase_; }
+
   void shutdownThread() override;
 
 protected:
   void work() override;
 
 private:
   void simpleWarmup();
+  const TerminationPredicateFactory& termination_predicate_factory_;
+  const SequencerFactory& sequencer_factory_;
   Envoy::Stats::ScopePtr worker_scope_;
   Envoy::Stats::ScopePtr worker_number_scope_;
   const int worker_number_;
-  const Envoy::MonotonicTime starting_time_;
   Envoy::Tracing::HttpTracerPtr& http_tracer_;
   RequestSourcePtr request_generator_;
   BenchmarkClientPtr benchmark_client_;
-  TerminationPredicatePtr termination_predicate_;
-  const SequencerPtr sequencer_;
+  PhasePtr phase_;
   Envoy::LocalInfo::LocalInfoPtr local_info_;
   std::map<std::string, uint64_t> thread_local_counter_values_;
 };

source/client/factories_impl.cc

@@ -45,15 +45,14 @@ BenchmarkClientPtr BenchmarkClientFactoryImpl::create(
 SequencerFactoryImpl::SequencerFactoryImpl(const Options& options)
     : OptionBasedFactoryImpl(options) {}
 
-SequencerPtr SequencerFactoryImpl::create(Envoy::TimeSource& time_source,
-                                          Envoy::Event::Dispatcher& dispatcher,
-                                          Envoy::MonotonicTime start_time,
-                                          BenchmarkClient& benchmark_client,
-                                          TerminationPredicate& termination_predicate,
-                                          Envoy::Stats::Scope& scope) const {
+SequencerPtr SequencerFactoryImpl::create(
+    Envoy::TimeSource& time_source, Envoy::Event::Dispatcher& dispatcher,
+    BenchmarkClient& benchmark_client, TerminationPredicatePtr&& termination_predicate,
+    Envoy::Stats::Scope& scope, const Envoy::MonotonicTime scheduled_starting_time) const {
   StatisticFactoryImpl statistic_factory(options_);
-  RateLimiterPtr rate_limiter =
-      std::make_unique<LinearRateLimiter>(time_source, Frequency(options_.requestsPerSecond()));
+  Frequency frequency(options_.requestsPerSecond());
+  RateLimiterPtr rate_limiter = std::make_unique<ScheduledStartingRateLimiter>(
+      std::make_unique<LinearRateLimiter>(time_source, frequency), scheduled_starting_time);
   const uint64_t burst_size = options_.burstSize();
 
   if (burst_size) {
@@ -71,9 +70,9 @@ SequencerPtr SequencerFactoryImpl::create(Envoy::TimeSource& time_source,
     return benchmark_client.tryStartRequest(std::move(f));
   };
   return std::make_unique<SequencerImpl>(
-      platform_util_, dispatcher, time_source, start_time, std::move(rate_limiter),
-      sequencer_target, statistic_factory.create(), statistic_factory.create(),
-      options_.sequencerIdleStrategy(), termination_predicate, scope);
+      platform_util_, dispatcher, time_source, std::move(rate_limiter), sequencer_target,
+      statistic_factory.create(), statistic_factory.create(), options_.sequencerIdleStrategy(),
+      std::move(termination_predicate), scope);
 }
 
 StoreFactoryImpl::StoreFactoryImpl(const Options& options) : OptionBasedFactoryImpl(options) {}
@@ -172,11 +171,10 @@ RequestSourceFactoryImpl::create(const Envoy::Upstream::ClusterManagerPtr& clust
 TerminationPredicateFactoryImpl::TerminationPredicateFactoryImpl(const Options& options)
     : OptionBasedFactoryImpl(options) {}
 
-TerminationPredicatePtr
-TerminationPredicateFactoryImpl::create(Envoy::TimeSource& time_source, Envoy::Stats::Scope& scope,
-                                        const Envoy::MonotonicTime start) const {
+TerminationPredicatePtr TerminationPredicateFactoryImpl::create(Envoy::TimeSource& time_source,
+                                                                Envoy::Stats::Scope& scope) const {
   TerminationPredicatePtr duration_predicate =
-      std::make_unique<DurationTerminationPredicateImpl>(time_source, start, options_.duration());
+      std::make_unique<DurationTerminationPredicateImpl>(time_source, options_.duration());
   TerminationPredicate* current_predicate = duration_predicate.get();
   current_predicate = linkConfiguredPredicates(*current_predicate, options_.failurePredicates(),
                                                TerminationPredicate::Status::FAIL, scope);

source/client/factories_impl.h

@@ -39,9 +39,9 @@ class SequencerFactoryImpl : public OptionBasedFactoryImpl, public SequencerFact
 public:
   SequencerFactoryImpl(const Options& options);
   SequencerPtr create(Envoy::TimeSource& time_source, Envoy::Event::Dispatcher& dispatcher,
-                      Envoy::MonotonicTime start_time, BenchmarkClient& benchmark_client,
-                      TerminationPredicate& termination_predicate,
-                      Envoy::Stats::Scope& scope) const override;
+                      BenchmarkClient& benchmark_client,
+                      TerminationPredicatePtr&& termination_predicate, Envoy::Stats::Scope& scope,
+                      const Envoy::MonotonicTime scheduled_starting_time) const override;
 };
 
 class StoreFactoryImpl : public OptionBasedFactoryImpl, public StoreFactory {
@@ -78,8 +78,8 @@ class TerminationPredicateFactoryImpl : public OptionBasedFactoryImpl,
                                         public TerminationPredicateFactory {
 public:
   TerminationPredicateFactoryImpl(const Options& options);
-  TerminationPredicatePtr create(Envoy::TimeSource& time_source, Envoy::Stats::Scope& scope,
-                                 const Envoy::MonotonicTime start) const override;
+  TerminationPredicatePtr create(Envoy::TimeSource& time_source,
+                                 Envoy::Stats::Scope& scope) 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

@@ -454,7 +454,7 @@ bool ProcessImpl::run(OutputCollector& collector) {
   int i = 0;
   std::chrono::nanoseconds total_execution_duration = 0ns;
   for (auto& worker : workers_) {
-    auto sequencer_execution_duration = worker->sequencer().executionDuration();
+    auto sequencer_execution_duration = worker->phase().sequencer().executionDuration();
     // 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) {

source/common/BUILD

@@ -64,6 +64,7 @@ envoy_cc_library(
 envoy_cc_library(
     name = "nighthawk_common_lib",
     srcs = [
+        "phase_impl.cc",
         "rate_limiter_impl.cc",
         "sequencer_impl.cc",
         "statistic_impl.cc",
@@ -75,6 +76,7 @@ envoy_cc_library(
     ],
     hdrs = [
         "frequency.h",
+        "phase_impl.h",
         "platform_util_impl.h",
         "rate_limiter_impl.h",
         "sequencer_impl.h",

source/common/phase_impl.cc

@@ -0,0 +1,18 @@
+#include "common/phase_impl.h"
+
+namespace Nighthawk {
+
+absl::string_view PhaseImpl::id() const { return id_; }
+
+Sequencer& PhaseImpl::sequencer() const { return *sequencer_; }
+
+bool PhaseImpl::shouldMeasureLatencies() const { return should_measure_latencies_; }
+
+void PhaseImpl::run() const {
+  ENVOY_LOG(trace, "starting '{}' phase", id());
+  sequencer().start();
+  sequencer().waitForCompletion();
+  ENVOY_LOG(trace, "finished '{}' phase", id());
+}
+
+} // namespace Nighthawk