Adaptive Load Controller main loop

include/nighthawk/adaptive_load/adaptive_load_controller.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "envoy/common/time.h"
+#include "envoy/common/pure.h"
 
 #include "external/envoy/source/common/common/statusor.h"
 
@@ -10,26 +10,35 @@
 namespace Nighthawk {
 
 /**
- * Performs an adaptive load session consisting of the Adjusting Stage and the
- * Testing Stage. Adjusting Stage: Runs a series of short benchmarks, checks metrics according to
- * MetricSpecs, and adjusts load up or down based on the result; returns an error if convergence is
- * not detected before the deadline in the spec. Load adjustments and convergence detection are
- * computed by a StepController plugin. Metric values are obtained through MetricsPlugins. Testing
- * Stage: When the optimal load is found, runs one long benchmark to validate it.
- *
- * @param nighthawk_service_stub A Nighthawk Service gRPC stub.
- * @param spec A proto that defines all aspects of the adaptive load session, including metrics,
- * threshold, duration of adjusting stage benchmarks, and underlying Nighthawk traffic parameters.
- * @param time_source An abstraction of the system clock. Normally, just construct an
- * Envoy::Event::RealTimeSystem and pass it. NO_CHECK_FORMAT(real_time). If calling from an
- * Envoy-based process, there may be an existing TimeSource or TimeSystem to use. If calling
- * from a test, pass a fake TimeSource.
- *
- * @return StatusOr<AdaptiveLoadSessionOutput> A proto logging the result of all traffic attempted
- * and all corresponding metric values and scores, or an overall error status if the session failed.
+ * Contains the main loop of the adaptive load controller. Consults a StepController for load
+ * decisions, interacts with Nighthawk Service and MetricsPlugins.
  */
-absl::StatusOr<nighthawk::adaptive_load::AdaptiveLoadSessionOutput> PerformAdaptiveLoadSession(
-    nighthawk::client::NighthawkService::StubInterface* nighthawk_service_stub,
-    const nighthawk::adaptive_load::AdaptiveLoadSessionSpec& spec, Envoy::TimeSource& time_source);
+class AdaptiveLoadController {
+public:
+  virtual ~AdaptiveLoadController() = default;
+  /**
+   * Performs an adaptive load session consisting of the Adjusting Stage and the
+   * Testing Stage.
+   *
+   * Adjusting Stage: Runs a series of short benchmarks, checks metrics according to MetricSpecs,
+   * and adjusts load up or down based on the result. Returns an error if convergence is not
+   * detected before the deadline in the spec. Load adjustments and convergence detection are
+   * computed by a StepController plugin. Metric values are obtained through MetricsPlugins.
+   *
+   * Testing Stage: When the optimal load is found, runs one long benchmark to validate it.
+   *
+   * @param nighthawk_service_stub A Nighthawk Service gRPC stub.
+   * @param spec A proto that defines all aspects of the adaptive load session, including metrics,
+   * threshold, duration of adjusting stage benchmarks, and underlying Nighthawk traffic parameters.
+   *
+   * @return StatusOr<AdaptiveLoadSessionOutput> A proto logging the result of all traffic attempted
+   * and all corresponding metric values and scores, or an overall error status if the session
+   * failed.
+   */
+  virtual absl::StatusOr<nighthawk::adaptive_load::AdaptiveLoadSessionOutput>
+  PerformAdaptiveLoadSession(
+      nighthawk::client::NighthawkService::StubInterface* nighthawk_service_stub,
+      const nighthawk::adaptive_load::AdaptiveLoadSessionSpec& spec) PURE;
+};
 
 } // namespace Nighthawk

source/adaptive_load/BUILD

@@ -8,6 +8,30 @@ licenses(["notice"])  # Apache 2
 
 envoy_package()
 
+envoy_cc_library(
+    name = "adaptive_load_controller_impl",
+    srcs = [
+        "adaptive_load_controller_impl.cc",
+    ],
+    hdrs = [
+        "adaptive_load_controller_impl.h",
+    ],
+    repository = "@envoy",
+    visibility = ["//visibility:public"],
+    deps = [
+        ":metrics_plugin_impl",
+        ":plugin_loader",
+        "//api/adaptive_load:adaptive_load_proto_cc_proto",
+        "//api/client:base_cc_proto",
+        "//include/nighthawk/adaptive_load:adaptive_load_controller",
+        "//include/nighthawk/adaptive_load:metrics_evaluator",
+        "//include/nighthawk/adaptive_load:scoring_function",
+        "//include/nighthawk/adaptive_load:session_spec_proto_helper",
+        "//include/nighthawk/adaptive_load:step_controller",
+        "//include/nighthawk/common:nighthawk_service_client",
+    ],
+)
+
 envoy_cc_library(
     name = "config_validator_impl",
     srcs = [

source/adaptive_load/adaptive_load_controller_impl.cc

@@ -0,0 +1,205 @@
+#include "adaptive_load/adaptive_load_controller_impl.h"
+
+#include <chrono>
+
+#include "envoy/common/exception.h"
+#include "envoy/config/core/v3/base.pb.h"
+
+#include "nighthawk/adaptive_load/adaptive_load_controller.h"
+#include "nighthawk/adaptive_load/metrics_plugin.h"
+#include "nighthawk/adaptive_load/scoring_function.h"
+#include "nighthawk/adaptive_load/step_controller.h"
+
+#include "external/envoy/source/common/common/logger.h"
+#include "external/envoy/source/common/common/statusor.h"
+#include "external/envoy/source/common/protobuf/protobuf.h"
+
+#include "api/adaptive_load/adaptive_load.pb.h"
+#include "api/adaptive_load/benchmark_result.pb.h"
+#include "api/adaptive_load/metric_spec.pb.h"
+#include "api/client/options.pb.h"
+#include "api/client/output.pb.h"
+#include "api/client/service.grpc.pb.h"
+
+#include "absl/container/flat_hash_map.h"
+#include "absl/status/status.h"
+#include "absl/strings/str_format.h"
+#include "absl/strings/str_join.h"
+#include "adaptive_load/metrics_plugin_impl.h"
+#include "adaptive_load/plugin_loader.h"
+
+namespace Nighthawk {
+
+namespace {
+
+using nighthawk::adaptive_load::AdaptiveLoadSessionOutput;
+using nighthawk::adaptive_load::AdaptiveLoadSessionSpec;
+using nighthawk::adaptive_load::BenchmarkResult;
+using nighthawk::adaptive_load::MetricEvaluation;
+using nighthawk::adaptive_load::MetricSpec;
+using nighthawk::adaptive_load::MetricSpecWithThreshold;
+using nighthawk::adaptive_load::ThresholdSpec;
+
+/**
+ * Loads and initializes MetricsPlugins requested in the session spec. Assumes the spec has already
+ * been validated; crashes the process otherwise.
+ *
+ * @param spec Adaptive load session spec proto that has already been validated.
+ *
+ * @return Map from MetricsPlugin names to initialized plugins, to be used in the course of a single
+ * adaptive load session based on the session spec.
+ */
+absl::flat_hash_map<std::string, MetricsPluginPtr>
+LoadMetricsPlugins(const AdaptiveLoadSessionSpec& spec) {
+  absl::flat_hash_map<std::string, MetricsPluginPtr> name_to_custom_metrics_plugin_map;
+  for (const envoy::config::core::v3::TypedExtensionConfig& config :
+       spec.metrics_plugin_configs()) {
+    absl::StatusOr<MetricsPluginPtr> metrics_plugin_or = LoadMetricsPlugin(config);
+    RELEASE_ASSERT(
+        metrics_plugin_or.ok(),
+        absl::StrCat(
+            "MetricsPlugin loading error should have been caught during input validation: ",
+            metrics_plugin_or.status().message()));
+    name_to_custom_metrics_plugin_map[config.name()] = std::move(metrics_plugin_or.value());
+  }
+  return name_to_custom_metrics_plugin_map;
+}
+
+/**
+ * Loads and initializes a StepController plugin requested in the session spec. Assumes
+ * the spec has already been validated; crashes the process otherwise.
+ *
+ * @param spec Adaptive load session spec proto that has already been validated.
+ *
+ * @return unique_ptr<StepController> Initialized StepController plugin.
+ */
+StepControllerPtr LoadStepControllerPluginFromSpec(const AdaptiveLoadSessionSpec& spec) {
+  absl::StatusOr<StepControllerPtr> step_controller_or =
+      LoadStepControllerPlugin(spec.step_controller_config(), spec.nighthawk_traffic_template());
+  RELEASE_ASSERT(
+      step_controller_or.ok(),
+      absl::StrCat(
+          "StepController plugin loading error should have been caught during input validation: ",
+          step_controller_or.status().message()));
+  return std::move(step_controller_or.value());
+}
+
+} // namespace
+
+AdaptiveLoadControllerImpl::AdaptiveLoadControllerImpl(
+    const NighthawkServiceClient& nighthawk_service_client,
+    const MetricsEvaluator& metrics_evaluator,
+    const AdaptiveLoadSessionSpecProtoHelper& session_spec_proto_helper,
+    Envoy::TimeSource& time_source)
+    : nighthawk_service_client_{nighthawk_service_client}, metrics_evaluator_{metrics_evaluator},
+      session_spec_proto_helper_{session_spec_proto_helper}, time_source_{time_source} {}
+
+absl::StatusOr<BenchmarkResult> AdaptiveLoadControllerImpl::PerformAndAnalyzeNighthawkBenchmark(
+    nighthawk::client::NighthawkService::StubInterface* nighthawk_service_stub,
+    const AdaptiveLoadSessionSpec& spec,
+    const absl::flat_hash_map<std::string, MetricsPluginPtr>& name_to_custom_plugin_map,
+    StepController& step_controller, Envoy::ProtobufWkt::Duration duration) {
+  absl::StatusOr<nighthawk::client::CommandLineOptions> command_line_options_or =
+      step_controller.GetCurrentCommandLineOptions();
+  if (!command_line_options_or.ok()) {
+    ENVOY_LOG_MISC(error, "Error constructing Nighthawk input: {}: {}",
+                   command_line_options_or.status().code(),
+                   command_line_options_or.status().message());
+    return command_line_options_or.status();
+  }
+  nighthawk::client::CommandLineOptions command_line_options = command_line_options_or.value();
+  // Overwrite the duration in the traffic template with the specified duration of the adjusting
+  // or testing stage.
+  *command_line_options.mutable_duration() = std::move(duration);
+
+  ENVOY_LOG_MISC(info, "Sending load: {}", command_line_options.DebugString());
+  absl::StatusOr<nighthawk::client::ExecutionResponse> nighthawk_response_or =
+      nighthawk_service_client_.PerformNighthawkBenchmark(nighthawk_service_stub,
+                                                          command_line_options);
+  if (!nighthawk_response_or.ok()) {
+    ENVOY_LOG_MISC(error, "Nighthawk Service error: {}: {}", nighthawk_response_or.status().code(),
+                   nighthawk_response_or.status().message());
+    return nighthawk_response_or.status();
+  }
+  nighthawk::client::ExecutionResponse nighthawk_response = nighthawk_response_or.value();
+
+  absl::StatusOr<BenchmarkResult> benchmark_result_or =
+      metrics_evaluator_.AnalyzeNighthawkBenchmark(nighthawk_response, spec,
+                                                   name_to_custom_plugin_map);
+  if (!benchmark_result_or.ok()) {
+    ENVOY_LOG_MISC(error, "Benchmark scoring error: {}: {}", benchmark_result_or.status().code(),
+                   benchmark_result_or.status().message());
+    return benchmark_result_or.status();
+  }
+  BenchmarkResult benchmark_result = benchmark_result_or.value();
+  for (const MetricEvaluation& evaluation : benchmark_result.metric_evaluations()) {
+    ENVOY_LOG_MISC(info, "Evaluation: {}", evaluation.DebugString());
+  }
+  step_controller.UpdateAndRecompute(benchmark_result);
+  return benchmark_result;
+}
+
+absl::StatusOr<AdaptiveLoadSessionOutput> AdaptiveLoadControllerImpl::PerformAdaptiveLoadSession(
+    nighthawk::client::NighthawkService::StubInterface* nighthawk_service_stub,
+    const AdaptiveLoadSessionSpec& input_spec) {
+  AdaptiveLoadSessionSpec spec = session_spec_proto_helper_.SetSessionSpecDefaults(input_spec);
+  absl::Status validation_status = session_spec_proto_helper_.CheckSessionSpec(spec);
+  if (!validation_status.ok()) {
+    ENVOY_LOG_MISC(error, "Validation failed: {}", validation_status.message());
+    return validation_status;
+  }
+  absl::flat_hash_map<std::string, MetricsPluginPtr> name_to_custom_metrics_plugin_map =
+      LoadMetricsPlugins(spec);
+  StepControllerPtr step_controller = LoadStepControllerPluginFromSpec(spec);
+  AdaptiveLoadSessionOutput output;
+
+  // Threshold specs are reproduced in the output proto for convenience.
+  for (const nighthawk::adaptive_load::MetricSpecWithThreshold& threshold :
+       spec.metric_thresholds()) {
+    *output.mutable_metric_thresholds()->Add() = threshold;
+  }
+
+  // Perform adjusting stage:
+  Envoy::MonotonicTime start_time = time_source_.monotonicTime();
+  std::string doom_reason;
+  do {
+    absl::StatusOr<BenchmarkResult> result_or = PerformAndAnalyzeNighthawkBenchmark(
+        nighthawk_service_stub, spec, name_to_custom_metrics_plugin_map, *step_controller,
+        spec.measuring_period());
+    if (!result_or.ok()) {
+      return result_or.status();
+    }
+    BenchmarkResult result = result_or.value();
+    *output.mutable_adjusting_stage_results()->Add() = result;
+
+    const std::chrono::nanoseconds time_limit_ns(
+        Envoy::Protobuf::util::TimeUtil::DurationToNanoseconds(spec.convergence_deadline()));
+    const auto elapsed_time_ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
+        time_source_.monotonicTime() - start_time);
+    if (elapsed_time_ns > time_limit_ns) {
+      std::string message = absl::StrFormat("Failed to converge before deadline of %.2f seconds.",
+                                            time_limit_ns.count() / 1e9);
+      ENVOY_LOG_MISC(error, message);
+      return absl::DeadlineExceededError(message);
+    }
+  } while (!step_controller->IsConverged() && !step_controller->IsDoomed(doom_reason));
+
+  if (step_controller->IsDoomed(doom_reason)) {
+    std::string message =
+        absl::StrCat("Step controller determined that it can never converge: ", doom_reason);
+    ENVOY_LOG_MISC(error, message);
+    return absl::AbortedError(message);
+  }
+
+  // Perform testing stage:
+  absl::StatusOr<BenchmarkResult> result_or = PerformAndAnalyzeNighthawkBenchmark(
+      nighthawk_service_stub, spec, name_to_custom_metrics_plugin_map, *step_controller,
+      spec.testing_stage_duration());
+  if (!result_or.ok()) {
+    return result_or.status();
+  }
+  *output.mutable_testing_stage_result() = result_or.value();
+  return output;
+}
+
+} // namespace Nighthawk

source/adaptive_load/adaptive_load_controller_impl.h

@@ -0,0 +1,83 @@
+#include "envoy/common/time.h"
+
+#include "nighthawk/adaptive_load/adaptive_load_controller.h"
+#include "nighthawk/adaptive_load/metrics_evaluator.h"
+#include "nighthawk/adaptive_load/session_spec_proto_helper.h"
+#include "nighthawk/adaptive_load/step_controller.h"
+#include "nighthawk/common/nighthawk_service_client.h"
+
+namespace Nighthawk {
+
+class AdaptiveLoadControllerImpl : public AdaptiveLoadController {
+public:
+  /**
+   * Constructs an implementation of the adaptive load controller main loop that relies on logic in
+   * several helper objects. Through helpers, it performs Nighthawk Service benchmarks, obtains
+   * metrics from MetricsPlugins, scores the results, and consults a StepController plugin to
+   * determine the next load and detect convergence. All plugins are specified through the
+   * AdaptiveLoadSessionSpec proto.
+   *
+   * This class is thread-safe, but Nighthawk Service itself does not support multiple simultaneous
+   * benchmarks.
+   *
+   * Usage:
+   *
+   *   AdaptiveLoadControllerImpl controller(
+   *       NighthawkServiceClientImpl(),
+   *       MetricsEvaluatorImpl(),
+   *       AdaptiveLoadSessionSpecProtoHelperImpl(),
+   *       Envoy::Event::RealTimeSystem()); // NO_CHECK_FORMAT(real_time))
+   *   AdaptiveLoadSessionSpec spec;
+   *   // (set spec fields here)
+   *   StatusOr<AdaptiveLoadSessionOutput> output =
+   *       controller.PerformAdaptiveLoadSession(&nighthawk_service_stub, spec);
+   *
+   * @param nighthawk_service_client A helper that executes Nighthawk Service benchmarks given a
+   * gRPC stub.
+   * @param metrics_evaluator A helper that obtains metrics from MetricsPlugins and Nighthawk
+   * Service responses, then scores them.
+   * @param session_spec_proto_helper A helper that sets default values and performs validation in
+   * an AdaptiveLoadSessionSpec proto.
+   * @param time_source An abstraction of the system clock. Normally, just construct an
+   * Envoy::Event::RealTimeSystem and pass it. NO_CHECK_FORMAT(real_time). If calling from an
+   * Envoy-based process, there may be an existing TimeSource or TimeSystem to use. If calling
+   * from a test, pass a fake TimeSource.
+   */
+  AdaptiveLoadControllerImpl(const NighthawkServiceClient& nighthawk_service_client,
+                             const MetricsEvaluator& metrics_evaluator,
+                             const AdaptiveLoadSessionSpecProtoHelper& session_spec_proto_helper,
+                             Envoy::TimeSource& time_source);
+
+  absl::StatusOr<nighthawk::adaptive_load::AdaptiveLoadSessionOutput> PerformAdaptiveLoadSession(
+      nighthawk::client::NighthawkService::StubInterface* nighthawk_service_stub,
+      const nighthawk::adaptive_load::AdaptiveLoadSessionSpec& spec) override;
+
+private:
+  /**
+   * Gets the current load from the StepController, performs a benchmark via a Nighthawk Service,
+   * hands the result off for analysis, and reports the scores back to the StepController.
+   *
+   * @param nighthawk_service_stub Nighthawk Service gRPC stub.
+   * @param spec Proto describing the overall adaptive load session.
+   * @param name_to_custom_plugin_map Common map from plugin names to MetricsPlugins loaded and
+   * initialized once at the beginning of the session and passed to all calls of this function.
+   * @param step_controller The active StepController specified in the session spec proto.
+   * @param duration The duration of the benchmark to insert into the traffic template, different
+   * between adjusting and testing stages.
+   *
+   * @return BenchmarkResult Proto containing either an error status or raw Nighthawk Service
+   * results, metric values, and metric scores.
+   */
+  absl::StatusOr<nighthawk::adaptive_load::BenchmarkResult> PerformAndAnalyzeNighthawkBenchmark(
+      nighthawk::client::NighthawkService::StubInterface* nighthawk_service_stub,
+      const nighthawk::adaptive_load::AdaptiveLoadSessionSpec& spec,
+      const absl::flat_hash_map<std::string, MetricsPluginPtr>& name_to_custom_plugin_map,
+      StepController& step_controller, Envoy::ProtobufWkt::Duration duration);
+
+  const NighthawkServiceClient& nighthawk_service_client_;
+  const MetricsEvaluator& metrics_evaluator_;
+  const AdaptiveLoadSessionSpecProtoHelper& session_spec_proto_helper_;
+  Envoy::TimeSource& time_source_;
+};
+
+} // namespace Nighthawk