Introduce Phases

include/nighthawk/client/client_worker.h

@@ -1,12 +1,13 @@
 #pragma once
 
 #include <memory>
+#include <vector>
 
 #include "envoy/common/pure.h"
 #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 +32,9 @@ class ClientWorker : virtual public Worker {
   virtual const std::map<std::string, uint64_t>& thread_local_counter_values() PURE;
 
   /**
-   * @return const Sequencer&
+   * @return const std::vector<PhasePtr>& vector of phases associated to this worker.
    */
-  virtual const Sequencer& sequencer() const PURE;
+  virtual const std::vector<PhasePtr>& phases() const PURE;
 };
 
 using ClientWorkerPtr = std::unique_ptr<ClientWorker>;

include/nighthawk/client/factories.h

@@ -37,9 +37,9 @@ 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 bool warmup) const PURE;
 };
 
 class StoreFactory {
@@ -72,8 +72,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,50 @@
+
+#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 measureLatencies() const PURE;
+
+  /**
+   * Runs the sequencer associated to this phase and blocks until completion, which means this phase
+   * has ended as well.
+   */
+  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/client_worker_impl.cc

@@ -1,13 +1,18 @@
 #include "client/client_worker_impl.h"
 
+#include <vector>
+
 #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,34 +22,55 @@ 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),
       request_generator_(request_generator_factory.create()),
       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_)) {}
+          fmt::format("{}", worker_number), *request_generator_)) {}
 
 void ClientWorkerImpl::simpleWarmup() {
-  ENVOY_LOG(debug, "> worker {}: warmup start.", worker_number_);
-  if (benchmark_client_->tryStartRequest([this](bool, bool) { dispatcher_->exit(); })) {
-    dispatcher_->run(Envoy::Event::Dispatcher::RunType::RunUntilExit);
-  } else {
-    ENVOY_LOG(warn, "> worker {}: failed to initiate warmup request.", worker_number_);
-  }
-  ENVOY_LOG(debug, "> worker {}: warmup done.", worker_number_);
+  // We add a short warmup phase, which ends when either the first successful response
+  // is observed or two seconds have passed, whichever comes first. These warmup conditions are
+  // registered on top of the original configured predicates.
+  // TODO(oschaaf): allow configuration of this phase once the ramping rate limiters land,
+  // or a generic configuration of multiple phases (id, duration, termination predicates, rate
+  // limiting options, etc).
+  TerminationPredicatePtr warmup_predicates =
+      termination_predicate_factory_.create(time_source_, *worker_number_scope_);
+  warmup_predicates
+      ->appendToChain(std::make_unique<StatsCounterAbsoluteThresholdTerminationPredicateImpl>(
+          worker_number_scope_->counter("benchmark.http_2xx"), 0,
+          TerminationPredicate::Status::TERMINATE))
+      .appendToChain(std::make_unique<DurationTerminationPredicateImpl>(time_source_, 2s));
+  phases_.emplace_back(std::make_unique<PhaseImpl>(
+      "warmup",
+      sequencer_factory_.create(time_source_, *dispatcher_, *benchmark_client_,
+                                std::move(warmup_predicates), *worker_number_scope_, true),
+      false));
 }
 
 void ClientWorkerImpl::work() {
   simpleWarmup();
-  benchmark_client_->setMeasureLatencies(true);
-  sequencer_->start();
-  sequencer_->waitForCompletion();
+
+  phases_.emplace_back(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_, false),
+      true));
+
+  for (auto& phase : phases_) {
+    benchmark_client_->setMeasureLatencies(phase->measureLatencies());
+    phase->run();
+    if (worker_number_scope_->counter("sequencer.failed_terminations").value() != 0) {
+      break;
+    }
+  }
+
   // Save a final snapshot of the worker-specific counter accumulations before
   // we exit the thread.
   for (const auto& stat : store_.counters()) {
@@ -69,7 +95,8 @@ void ClientWorkerImpl::shutdownThread() { benchmark_client_->terminate(); }
 StatisticPtrMap ClientWorkerImpl::statistics() const {
   StatisticPtrMap statistics;
   StatisticPtrMap s1 = benchmark_client_->statistics();
-  StatisticPtrMap s2 = sequencer_->statistics();
+  auto& sequencer = phases_.back()->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,26 @@ 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 std::vector<PhasePtr>& phases() const override { return phases_; }
+
   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_;
+  std::vector<PhasePtr> phases_;
   Envoy::LocalInfo::LocalInfoPtr local_info_;
   std::map<std::string, uint64_t> thread_local_counter_values_;
 };

source/client/factories_impl.cc

@@ -47,16 +47,15 @@ SequencerFactoryImpl::SequencerFactoryImpl(const Options& 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 {
+                                          TerminationPredicatePtr&& termination_predicate,
+                                          Envoy::Stats::Scope& scope, const bool warmup) const {
   StatisticFactoryImpl statistic_factory(options_);
-  RateLimiterPtr rate_limiter =
-      std::make_unique<LinearRateLimiter>(time_source, Frequency(options_.requestsPerSecond()));
+  Frequency frequency(warmup ? 1 : options_.requestsPerSecond());
+  RateLimiterPtr rate_limiter = std::make_unique<LinearRateLimiter>(time_source, frequency);
   const uint64_t burst_size = options_.burstSize();
 
-  if (burst_size) {
+  if (!warmup && burst_size) {
     rate_limiter = std::make_unique<BurstingRateLimiter>(std::move(rate_limiter), 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) {}
@@ -144,11 +143,10 @@ RequestSourcePtr RequestSourceFactoryImpl::create() const {
 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 bool warmup) const override;
 };
 
 class StoreFactoryImpl : public OptionBasedFactoryImpl, public StoreFactory {
@@ -76,8 +76,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

@@ -399,7 +399,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->phases().back()->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

@@ -41,6 +41,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",
@@ -51,6 +52,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::measureLatencies() const { return measure_latencies_; }
+
+void PhaseImpl::run() const {
+  ENVOY_LOG(trace, "starting '{}' phase", id_);
+  sequencer().start();
+  sequencer().waitForCompletion();
+  ENVOY_LOG(trace, "finished '{}' phase", id_);
+}
+
+} // namespace Nighthawk