Adaptive Load metrics evaluator library

include/nighthawk/adaptive_load/BUILD

@@ -52,6 +52,21 @@ envoy_basic_cc_library(
     ],
 )
 
+envoy_basic_cc_library(
+    name = "metrics_evaluator",
+    hdrs = [
+        "metrics_evaluator.h",
+    ],
+    include_prefix = "nighthawk/adaptive_load",
+    deps = [
+        ":metrics_plugin",
+        "//api/adaptive_load:adaptive_load_proto_cc_proto",
+        "@envoy//include/envoy/common:base_includes",
+        "@envoy//include/envoy/config:typed_config_interface",
+        "@envoy//source/common/common:statusor_lib_with_external_headers",
+    ],
+)
+
 envoy_basic_cc_library(
     name = "metrics_plugin",
     hdrs = [

include/nighthawk/adaptive_load/metrics_evaluator.h

@@ -0,0 +1,90 @@
+#include "envoy/config/core/v3/base.pb.h"
+
+#include "nighthawk/adaptive_load/metrics_plugin.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.pb.h"
+
+#include "absl/container/flat_hash_map.h"
+#include "absl/status/status.h"
+#include "absl/strings/str_join.h"
+
+namespace Nighthawk {
+
+/**
+ * A utility for calling MetricsPlugins and scoring metrics according to ThresholdSpecs.
+ *
+ * AnalyzeNighthawkBenchmark() is intended to be called repeatedly from the adaptive load controller
+ * main loop after each Nighthawk Service call. The controller maintains a set of shared
+ * MetricsPlugins that are initialized once for the whole session. AnalyzeNighthawkBenchmark() calls
+ * EvaluateMetric() and ExtractMetricSpecs() internally. The AdaptiveLoadSessionSpec is consulted
+ * for MetricSpec, ThresholdSpec, and MetricsPlugin information.
+ */
+class MetricsEvaluator {
+public:
+  virtual ~MetricsEvaluator() = default;
+
+  /**
+   * Calls a MetricPlugin to obtain the metric value defined by the MetricSpec, then scores the
+   * value according to a ThresholdSpec if one is present.
+   *
+   * @param metric_spec The MetricSpec identifying the metric by name and plugin name.
+   * @param metrics_plugin A MetricsPlugin that will be queried. The plugin must correspond to the
+   * plugin name in the MetricSpec, and it should support the requested metric name in the
+   * MetricSpec.
+   * @param threshold_spec A proto describing the threshold and scoring function. Nullptr if the
+   * metric is informational only.
+   *
+   * @return StatusOr<MetricEvaluation> A proto containing the metric value (and its score if a
+   * threshold was specified), or an error status if the metric could not be obtained from the
+   * MetricsPlugin.
+   */
+  virtual absl::StatusOr<nighthawk::adaptive_load::MetricEvaluation>
+  EvaluateMetric(const nighthawk::adaptive_load::MetricSpec& metric_spec,
+                 MetricsPlugin& metrics_plugin,
+                 const nighthawk::adaptive_load::ThresholdSpec* threshold_spec) const PURE;
+
+  /**
+   * Extracts pointers to metric descriptors and corresponding thresholds from a top-level adaptive
+   * load session spec to an ordered list and a map. Allows for uniform treatment of scored and
+   * informational metrics.
+   *
+   * @param spec The adaptive load session spec.
+   * @return Vector of pairs of pointers to MetricSpec and ThresholdSpec within |spec|. For
+   * informational metrics, the ThresholdSpec pointer is nullptr.
+   */
+  virtual const std::vector<std::pair<const nighthawk::adaptive_load::MetricSpec*,
+                                      const nighthawk::adaptive_load::ThresholdSpec*>>
+  ExtractMetricSpecs(const nighthawk::adaptive_load::AdaptiveLoadSessionSpec& spec) const PURE;
+
+  /**
+   * Analyzes a Nighthawk Service benchmark against configured MetricThresholds. For each
+   * MetricSpec, queries a MetricsPlugin for the current metric value. Assumes that the values from
+   * MetricsPlugins correspond timewise with the Nighthawk benchmark.
+   *
+   * @param nighthawk_response Proto returned from Nighthawk Service describing the latest single
+   * benchmark session. To be translated into scorable metrics by the "nighthawk.builtin"
+   * MetricsPlugin.
+   * @param spec Top-level proto defining the adaptive load session.
+   * @param name_to_custom_metrics_plugin_map Map from plugin names to initialized MetricsPlugins.
+   * Must include all MetricsPlugins referenced in the spec other than "nighthawk.builtin".
+   *
+   * @return StatusOr<BenchmarkResult> A proto containing all metric scores for this Nighthawk
+   * Service benchmark session, or an error propagated from a MetricsPlugin.
+   */
+  virtual absl::StatusOr<nighthawk::adaptive_load::BenchmarkResult>
+  AnalyzeNighthawkBenchmark(const nighthawk::client::ExecutionResponse& nighthawk_response,
+                            const nighthawk::adaptive_load::AdaptiveLoadSessionSpec& spec,
+                            const absl::flat_hash_map<std::string, MetricsPluginPtr>&
+                                name_to_custom_metrics_plugin_map) const PURE;
+};
+
+} // namespace Nighthawk

source/adaptive_load/BUILD

@@ -42,6 +42,27 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "metrics_evaluator_impl",
+    srcs = [
+        "metrics_evaluator_impl.cc",
+    ],
+    hdrs = [
+        "metrics_evaluator_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",
+    ],
+)
+
 envoy_cc_library(
     name = "metrics_plugin_impl",
     srcs = [

source/adaptive_load/metrics_evaluator_impl.cc

@@ -0,0 +1,114 @@
+#include "adaptive_load/metrics_evaluator_impl.h"
+
+#include <utility>
+
+#include "api/adaptive_load/metric_spec.pb.h"
+
+#include "adaptive_load/metrics_plugin_impl.h"
+#include "adaptive_load/plugin_loader.h"
+
+namespace Nighthawk {
+
+namespace {
+
+using ::nighthawk::adaptive_load::MetricSpec;
+using ::nighthawk::adaptive_load::MetricSpecWithThreshold;
+using ::nighthawk::adaptive_load::ThresholdSpec;
+
+} // namespace
+
+absl::StatusOr<nighthawk::adaptive_load::MetricEvaluation>
+MetricsEvaluatorImpl::EvaluateMetric(const MetricSpec& metric_spec, MetricsPlugin& metrics_plugin,
+                                     const ThresholdSpec* threshold_spec) const {
+  nighthawk::adaptive_load::MetricEvaluation evaluation;
+  evaluation.set_metric_id(
+      absl::StrCat(metric_spec.metrics_plugin_name(), "/", metric_spec.metric_name()));
+  const absl::StatusOr<double> metric_value_or =
+      metrics_plugin.GetMetricByName(metric_spec.metric_name());
+  if (!metric_value_or.ok()) {
+    return absl::Status(static_cast<absl::StatusCode>(metric_value_or.status().code()),
+                        absl::StrCat("Error calling MetricsPlugin '",
+                                     metric_spec.metrics_plugin_name(), ": ",
+                                     metric_value_or.status().message()));
+  }
+  const double metric_value = metric_value_or.value();
+  evaluation.set_metric_value(metric_value);
+  if (threshold_spec == nullptr) {
+    // Informational metric.
+    evaluation.set_weight(0.0);
+  } else {
+    evaluation.set_weight(threshold_spec->weight().value());
+    absl::StatusOr<ScoringFunctionPtr> scoring_function_or =
+        LoadScoringFunctionPlugin(threshold_spec->scoring_function());
+    RELEASE_ASSERT(scoring_function_or.ok(),
+                   absl::StrCat("ScoringFunction plugin loading error should have been caught "
+                                "during input validation: ",
+                                scoring_function_or.status().message()));
+    ScoringFunctionPtr scoring_function = std::move(scoring_function_or.value());
+    evaluation.set_threshold_score(scoring_function->EvaluateMetric(metric_value));
+  }
+  return evaluation;
+}
+
+const std::vector<std::pair<const MetricSpec*, const ThresholdSpec*>>
+MetricsEvaluatorImpl::ExtractMetricSpecs(
+    const nighthawk::adaptive_load::AdaptiveLoadSessionSpec& spec) const {
+  std::vector<std::pair<const MetricSpec*, const ThresholdSpec*>> spec_threshold_pairs;
+  for (const MetricSpecWithThreshold& metric_threshold : spec.metric_thresholds()) {
+    spec_threshold_pairs.emplace_back(&metric_threshold.metric_spec(),
+                                      &metric_threshold.threshold_spec());
+  }
+  for (const MetricSpec& metric_spec : spec.informational_metric_specs()) {
+    spec_threshold_pairs.emplace_back(&metric_spec, nullptr);
+  }
+  return spec_threshold_pairs;
+}
+
+absl::StatusOr<nighthawk::adaptive_load::BenchmarkResult>
+MetricsEvaluatorImpl::AnalyzeNighthawkBenchmark(
+    const nighthawk::client::ExecutionResponse& nighthawk_response,
+    const nighthawk::adaptive_load::AdaptiveLoadSessionSpec& spec,
+    const absl::flat_hash_map<std::string, MetricsPluginPtr>& name_to_custom_metrics_plugin_map)
+    const {
+  if (nighthawk_response.error_detail().code() != static_cast<int>(absl::StatusCode::kOk)) {
+    return absl::Status(static_cast<absl::StatusCode>(nighthawk_response.error_detail().code()),
+                        nighthawk_response.error_detail().message());
+  }
+
+  nighthawk::adaptive_load::BenchmarkResult benchmark_result;
+  *benchmark_result.mutable_nighthawk_service_output() = nighthawk_response.output();
+
+  // A map containing all available MetricsPlugins: preloaded custom plugins shared across all
+  // benchmarks, and a freshly instantiated builtin plugin for this benchmark only.
+  absl::flat_hash_map<std::string, MetricsPlugin*> name_to_plugin_map;
+  for (const auto& name_plugin_pair : name_to_custom_metrics_plugin_map) {
+    name_to_plugin_map[name_plugin_pair.first] = name_plugin_pair.second.get();
+  }
+  auto builtin_plugin =
+      std::make_unique<NighthawkStatsEmulatedMetricsPlugin>(nighthawk_response.output());
+  name_to_plugin_map["nighthawk.builtin"] = builtin_plugin.get();
+
+  const std::vector<std::pair<const MetricSpec*, const ThresholdSpec*>> spec_threshold_pairs =
+      ExtractMetricSpecs(spec);
+
+  std::vector<std::string> errors;
+  for (const std::pair<const MetricSpec*, const ThresholdSpec*>& spec_threshold_pair :
+       spec_threshold_pairs) {
+    absl::StatusOr<nighthawk::adaptive_load::MetricEvaluation> evaluation_or =
+        EvaluateMetric(*spec_threshold_pair.first,
+                       *name_to_plugin_map[spec_threshold_pair.first->metrics_plugin_name()],
+                       spec_threshold_pair.second);
+    if (!evaluation_or.ok()) {
+      errors.emplace_back(absl::StrCat("Error evaluating metric: ", evaluation_or.status().code(),
+                                       ": ", evaluation_or.status().message()));
+      continue;
+    }
+    *benchmark_result.mutable_metric_evaluations()->Add() = evaluation_or.value();
+  }
+  if (!errors.empty()) {
+    return absl::InternalError(absl::StrJoin(errors, "\n"));
+  }
+  return benchmark_result;
+}
+
+} // namespace Nighthawk

source/adaptive_load/metrics_evaluator_impl.h

@@ -0,0 +1,23 @@
+#include "nighthawk/adaptive_load/metrics_evaluator.h"
+
+namespace Nighthawk {
+
+class MetricsEvaluatorImpl : public MetricsEvaluator {
+public:
+  absl::StatusOr<nighthawk::adaptive_load::MetricEvaluation>
+  EvaluateMetric(const nighthawk::adaptive_load::MetricSpec& metric_spec,
+                 MetricsPlugin& metrics_plugin,
+                 const nighthawk::adaptive_load::ThresholdSpec* threshold_spec) const override;
+
+  const std::vector<std::pair<const nighthawk::adaptive_load::MetricSpec*,
+                              const nighthawk::adaptive_load::ThresholdSpec*>>
+  ExtractMetricSpecs(const nighthawk::adaptive_load::AdaptiveLoadSessionSpec& spec) const override;
+
+  absl::StatusOr<nighthawk::adaptive_load::BenchmarkResult>
+  AnalyzeNighthawkBenchmark(const nighthawk::client::ExecutionResponse& nighthawk_response,
+                            const nighthawk::adaptive_load::AdaptiveLoadSessionSpec& spec,
+                            const absl::flat_hash_map<std::string, MetricsPluginPtr>&
+                                name_to_custom_metrics_plugin_map) const override;
+};
+
+} // namespace Nighthawk

test/adaptive_load/BUILD

@@ -30,6 +30,18 @@ envoy_cc_test(
     ],
 )
 
+envoy_cc_test(
+    name = "metrics_evaluator_test",
+    srcs = ["metrics_evaluator_test.cc"],
+    repository = "@envoy",
+    deps = [
+        ":minimal_output",
+        "//source/adaptive_load:metrics_evaluator_impl",
+        "//source/adaptive_load:scoring_function_impl",
+        "//test/adaptive_load/fake_plugins/fake_metrics_plugin",
+    ],
+)
+
 envoy_cc_test(
     name = "metrics_plugin_test",
     srcs = ["metrics_plugin_test.cc"],