Adaptive Load library for calling Nighthawk Service

api/client/BUILD

@@ -23,6 +23,7 @@ cc_grpc_library(
     srcs = [
         ":base",
     ],
+    generate_mocks = True,
     grpc_only = True,
     proto_only = False,
     use_external = False,

include/nighthawk/adaptive_load/BUILD

@@ -67,6 +67,22 @@ envoy_basic_cc_library(
     ],
 )
 
+envoy_basic_cc_library(
+    name = "nighthawk_service_client",
+    hdrs = [
+        "nighthawk_service_client.h",
+    ],
+    include_prefix = "nighthawk/adaptive_load",
+    deps = [
+        "//api/client:base_cc_proto",
+        "//api/client:grpc_service_lib",
+        "@envoy//include/envoy/common:base_includes",
+        "@envoy//source/common/common:assert_lib_with_external_headers",
+        "@envoy//source/common/common:statusor_lib_with_external_headers",
+        "@envoy//source/common/protobuf:protobuf_with_external_headers",
+    ],
+)
+
 envoy_basic_cc_library(
     name = "scoring_function",
     hdrs = [

include/nighthawk/adaptive_load/nighthawk_service_client.h

@@ -0,0 +1,38 @@
+#pragma once
+#include "envoy/common/pure.h"
+
+#include "external/envoy/source/common/common/statusor.h"
+#include "external/envoy/source/common/protobuf/protobuf.h"
+
+#include "api/client/options.pb.h"
+#include "api/client/service.grpc.pb.h"
+
+namespace Nighthawk {
+
+/**
+ * An interface for interacting with a Nighthawk Service gRPC stub.
+ */
+class NighthawkServiceClient {
+public:
+  virtual ~NighthawkServiceClient() = default;
+
+  /**
+   * Runs a single benchmark using a Nighthawk Service.
+   *
+   * @param nighthawk_service_stub Nighthawk Service gRPC stub.
+   * @param command_line_options Nighthawk Service benchmark request proto generated by the
+   * StepController, without the duration set.
+   * @param duration Duration to insert into the benchmark request.
+   *
+   * @return StatusOr<ExecutionResponse> If we reached the Nighthawk Service, this is the raw
+   * ExecutionResponse proto, containing the benchmark data or possibly an error message from
+   * Nighthawk Service; if we had trouble communicating with the Nighthawk Service, we return an
+   * error status.
+   */
+  virtual absl::StatusOr<nighthawk::client::ExecutionResponse> PerformNighthawkBenchmark(
+      nighthawk::client::NighthawkService::StubInterface* nighthawk_service_stub,
+      const nighthawk::client::CommandLineOptions& command_line_options,
+      const Envoy::Protobuf::Duration& duration) PURE;
+};
+
+} // namespace Nighthawk

source/adaptive_load/BUILD

@@ -61,6 +61,26 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "nighthawk_service_client_impl",
+    srcs = [
+        "nighthawk_service_client_impl.cc",
+    ],
+    hdrs = [
+        "nighthawk_service_client_impl.h",
+    ],
+    repository = "@envoy",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//api/client:base_cc_proto",
+        "//api/client:grpc_service_lib",
+        "//include/nighthawk/adaptive_load:nighthawk_service_client",
+        "@envoy//source/common/common:assert_lib_with_external_headers",
+        "@envoy//source/common/common:statusor_lib_with_external_headers",
+        "@envoy//source/common/protobuf:protobuf_with_external_headers",
+    ],
+)
+
 envoy_cc_library(
     name = "plugin_loader",
     srcs = [

source/adaptive_load/nighthawk_service_client_impl.cc

@@ -0,0 +1,46 @@
+#include "adaptive_load/nighthawk_service_client_impl.h"
+
+#include "external/envoy/source/common/common/assert.h"
+
+namespace Nighthawk {
+
+absl::StatusOr<nighthawk::client::ExecutionResponse>
+NighthawkServiceClientImpl::PerformNighthawkBenchmark(
+    nighthawk::client::NighthawkService::StubInterface* nighthawk_service_stub,
+    const nighthawk::client::CommandLineOptions& command_line_options,
+    const Envoy::Protobuf::Duration& duration) {
+  nighthawk::client::CommandLineOptions options = command_line_options;
+  *options.mutable_duration() = duration;
+
+  nighthawk::client::ExecutionRequest request;
+  nighthawk::client::ExecutionResponse response;
+  *request.mutable_start_request()->mutable_options() = options;
+
+  ::grpc::ClientContext context;
+  std::shared_ptr<::grpc::ClientReaderWriterInterface<nighthawk::client::ExecutionRequest,
+                                                      nighthawk::client::ExecutionResponse>>
+      stream(nighthawk_service_stub->ExecutionStream(&context));
+
+  if (!stream->Write(request)) {
+    return absl::UnknownError("Failed to write request to the Nighthawk Service gRPC channel.");
+  } else if (!stream->WritesDone()) {
+    return absl::UnknownError("WritesDone() failed on the Nighthawk Service gRPC channel.");
+  }
+
+  bool got_response = false;
+  while (stream->Read(&response)) {
+    RELEASE_ASSERT(!got_response,
+                   "Nighthawk Service has started responding with more than one message.");
+    got_response = true;
+  }
+  if (!got_response) {
+    return absl::UnknownError("Nighthawk Service did not send a gRPC response.");
+  }
+  ::grpc::Status status = stream->Finish();
+  if (!status.ok()) {
+    return absl::Status(static_cast<absl::StatusCode>(status.error_code()), status.error_message());
+  }
+  return response;
+}
+
+} // namespace Nighthawk

source/adaptive_load/nighthawk_service_client_impl.h

@@ -0,0 +1,19 @@
+#include "nighthawk/adaptive_load/nighthawk_service_client.h"
+
+#include "external/envoy/source/common/common/statusor.h"
+#include "external/envoy/source/common/protobuf/protobuf.h"
+
+#include "api/client/options.pb.h"
+#include "api/client/service.grpc.pb.h"
+
+namespace Nighthawk {
+
+class NighthawkServiceClientImpl : public NighthawkServiceClient {
+public:
+  absl::StatusOr<nighthawk::client::ExecutionResponse> PerformNighthawkBenchmark(
+      nighthawk::client::NighthawkService::StubInterface* nighthawk_service_stub,
+      const nighthawk::client::CommandLineOptions& command_line_options,
+      const Envoy::Protobuf::Duration& duration) override;
+};
+
+} // namespace Nighthawk

test/adaptive_load/BUILD

@@ -40,6 +40,16 @@ envoy_cc_test(
     ],
 )
 
+envoy_cc_test(
+    name = "nighthawk_service_client_test",
+    srcs = ["nighthawk_service_client_test.cc"],
+    repository = "@envoy",
+    deps = [
+        "//source/adaptive_load:nighthawk_service_client_impl",
+        "@com_github_grpc_grpc//:grpc++_test",
+    ],
+)
+
 envoy_cc_test(
     name = "plugin_loader_test",
     srcs = ["plugin_loader_test.cc"],

test/adaptive_load/nighthawk_service_client_test.cc

@@ -0,0 +1,177 @@
+#include "external/envoy/source/common/protobuf/protobuf.h"
+
+#include "api/client/options.pb.h"
+#include "api/client/service.grpc.pb.h"
+#include "api/client/service_mock.grpc.pb.h"
+
+#include "grpcpp/test/mock_stream.h"
+
+#include "adaptive_load/nighthawk_service_client_impl.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace Nighthawk {
+
+namespace {
+
+using ::Envoy::Protobuf::util::MessageDifferencer;
+using ::nighthawk::client::CommandLineOptions;
+using ::nighthawk::client::ExecutionRequest;
+using ::nighthawk::client::ExecutionResponse;
+using ::testing::_;
+using ::testing::DoAll;
+using ::testing::HasSubstr;
+using ::testing::Return;
+using ::testing::SaveArg;
+using ::testing::SetArgPointee;
+
+TEST(PerformNighthawkBenchmark, UsesSpecifiedDuration) {
+  const int kExpectedSeconds = 123;
+  ExecutionRequest request;
+  nighthawk::client::MockNighthawkServiceStub mock_nighthawk_service_stub;
+  EXPECT_CALL(mock_nighthawk_service_stub, ExecutionStreamRaw)
+      .WillOnce([&request](grpc_impl::ClientContext*) {
+        auto* mock_reader_writer =
+            new grpc::testing::MockClientReaderWriter<ExecutionRequest, ExecutionResponse>();
+        EXPECT_CALL(*mock_reader_writer, Read(_)).WillOnce(Return(true)).WillOnce(Return(false));
+        EXPECT_CALL(*mock_reader_writer, Write(_, _))
+            .WillOnce(::testing::DoAll(::testing::SaveArg<0>(&request), Return(true)));
+        EXPECT_CALL(*mock_reader_writer, WritesDone()).WillOnce(Return(true));
+        EXPECT_CALL(*mock_reader_writer, Finish()).WillOnce(Return(::grpc::Status::OK));
+        return mock_reader_writer;
+      });
+  Envoy::Protobuf::Duration duration;
+  duration.set_seconds(kExpectedSeconds);
+  NighthawkServiceClientImpl client;
+  absl::StatusOr<ExecutionResponse> response_or = client.PerformNighthawkBenchmark(
+      &mock_nighthawk_service_stub, CommandLineOptions(), duration);
+  EXPECT_TRUE(response_or.ok());
+  EXPECT_EQ(request.start_request().options().duration().seconds(), kExpectedSeconds);
+}
+
+TEST(PerformNighthawkBenchmark, UsesSpecifiedCommandLineOptions) {
+  const int kExpectedRps = 456;
+  ExecutionRequest request;
+  nighthawk::client::MockNighthawkServiceStub mock_nighthawk_service_stub;
+  EXPECT_CALL(mock_nighthawk_service_stub, ExecutionStreamRaw)
+      .WillOnce([&request](grpc_impl::ClientContext*) {
+        auto* mock_reader_writer =
+            new grpc::testing::MockClientReaderWriter<ExecutionRequest, ExecutionResponse>();
+        EXPECT_CALL(*mock_reader_writer, Read(_)).WillOnce(Return(true)).WillOnce(Return(false));
+        EXPECT_CALL(*mock_reader_writer, Write(_, _))
+            .WillOnce(::testing::DoAll(::testing::SaveArg<0>(&request), Return(true)));
+        EXPECT_CALL(*mock_reader_writer, WritesDone()).WillOnce(Return(true));
+        EXPECT_CALL(*mock_reader_writer, Finish()).WillOnce(Return(::grpc::Status::OK));
+        return mock_reader_writer;
+      });
+  CommandLineOptions command_line_options;
+  command_line_options.mutable_requests_per_second()->set_value(kExpectedRps);
+  NighthawkServiceClientImpl client;
+  absl::StatusOr<ExecutionResponse> response_or = client.PerformNighthawkBenchmark(
+      &mock_nighthawk_service_stub, command_line_options, Envoy::Protobuf::Duration());
+  EXPECT_TRUE(response_or.ok());
+  EXPECT_EQ(request.start_request().options().requests_per_second().value(), kExpectedRps);
+}
+
+TEST(PerformNighthawkBenchmark, ReturnsNighthawkResponseSuccessfully) {
+  ExecutionResponse expected_response;
+  nighthawk::client::MockNighthawkServiceStub mock_nighthawk_service_stub;
+  EXPECT_CALL(mock_nighthawk_service_stub, ExecutionStreamRaw)
+      .WillOnce([&expected_response](grpc_impl::ClientContext*) {
+        auto* mock_reader_writer =
+            new grpc::testing::MockClientReaderWriter<ExecutionRequest, ExecutionResponse>();
+        EXPECT_CALL(*mock_reader_writer, Read(_))
+            .WillOnce(DoAll(SetArgPointee<0>(expected_response), Return(true)))
+            .WillOnce(Return(false));
+        EXPECT_CALL(*mock_reader_writer, Write(_, _)).WillOnce(Return(true));
+        EXPECT_CALL(*mock_reader_writer, WritesDone()).WillOnce(Return(true));
+        EXPECT_CALL(*mock_reader_writer, Finish()).WillOnce(Return(::grpc::Status::OK));
+        return mock_reader_writer;
+      });
+  NighthawkServiceClientImpl client;
+  absl::StatusOr<ExecutionResponse> response_or = client.PerformNighthawkBenchmark(
+      &mock_nighthawk_service_stub, CommandLineOptions(), Envoy::Protobuf::Duration());
+  EXPECT_TRUE(response_or.ok());
+  ExecutionResponse actual_response = response_or.value();
+  EXPECT_TRUE(MessageDifferencer::Equivalent(actual_response, expected_response));
+  EXPECT_EQ(actual_response.DebugString(), expected_response.DebugString());
+}
+
+TEST(PerformNighthawkBenchmark, ReturnsErrorIfNighthawkServiceDoesNotSendResponse) {
+  nighthawk::client::MockNighthawkServiceStub mock_nighthawk_service_stub;
+  EXPECT_CALL(mock_nighthawk_service_stub, ExecutionStreamRaw)
+      .WillOnce([](grpc_impl::ClientContext*) {
+        auto* mock_reader_writer =
+            new grpc::testing::MockClientReaderWriter<ExecutionRequest, ExecutionResponse>();
+        EXPECT_CALL(*mock_reader_writer, Read(_)).WillOnce(Return(false));
+        EXPECT_CALL(*mock_reader_writer, Write(_, _)).WillOnce(Return(true));
+        EXPECT_CALL(*mock_reader_writer, WritesDone()).WillOnce(Return(true));
+        return mock_reader_writer;
+      });
+  NighthawkServiceClientImpl client;
+  absl::StatusOr<ExecutionResponse> response_or = client.PerformNighthawkBenchmark(
+      &mock_nighthawk_service_stub, CommandLineOptions(), Envoy::Protobuf::Duration());
+  ASSERT_FALSE(response_or.ok());
+  EXPECT_EQ(response_or.status().code(), absl::StatusCode::kUnknown);
+  EXPECT_THAT(response_or.status().message(),
+              HasSubstr("Nighthawk Service did not send a gRPC response."));
+}
+
+TEST(PerformNighthawkBenchmark, ReturnsErrorIfNighthawkServiceWriteFails) {
+  nighthawk::client::MockNighthawkServiceStub mock_nighthawk_service_stub;
+  EXPECT_CALL(mock_nighthawk_service_stub, ExecutionStreamRaw)
+      .WillOnce([](grpc_impl::ClientContext*) {
+        auto* mock_reader_writer =
+            new grpc::testing::MockClientReaderWriter<ExecutionRequest, ExecutionResponse>();
+        EXPECT_CALL(*mock_reader_writer, Write(_, _)).WillOnce(Return(false));
+        return mock_reader_writer;
+      });
+  NighthawkServiceClientImpl client;
+  absl::StatusOr<ExecutionResponse> response_or = client.PerformNighthawkBenchmark(
+      &mock_nighthawk_service_stub, CommandLineOptions(), Envoy::Protobuf::Duration());
+  ASSERT_FALSE(response_or.ok());
+  EXPECT_EQ(response_or.status().code(), absl::StatusCode::kUnknown);
+  EXPECT_THAT(response_or.status().message(), HasSubstr("Failed to write"));
+}
+
+TEST(PerformNighthawkBenchmark, ReturnsErrorIfNighthawkServiceWritesDoneFails) {
+  nighthawk::client::MockNighthawkServiceStub mock_nighthawk_service_stub;
+  EXPECT_CALL(mock_nighthawk_service_stub, ExecutionStreamRaw)
+      .WillOnce([](grpc_impl::ClientContext*) {
+        auto* mock_reader_writer =
+            new grpc::testing::MockClientReaderWriter<ExecutionRequest, ExecutionResponse>();
+        EXPECT_CALL(*mock_reader_writer, Write(_, _)).WillOnce(Return(true));
+        EXPECT_CALL(*mock_reader_writer, WritesDone()).WillOnce(Return(false));
+        return mock_reader_writer;
+      });
+  NighthawkServiceClientImpl client;
+  absl::StatusOr<ExecutionResponse> response_or = client.PerformNighthawkBenchmark(
+      &mock_nighthawk_service_stub, CommandLineOptions(), Envoy::Protobuf::Duration());
+  ASSERT_FALSE(response_or.ok());
+  EXPECT_EQ(response_or.status().code(), absl::StatusCode::kUnknown);
+  EXPECT_THAT(response_or.status().message(), HasSubstr("WritesDone() failed"));
+}
+
+TEST(PerformNighthawkBenchmark, ReturnsErrorIfNighthawkServiceGrpcStreamClosesAbnormally) {
+  nighthawk::client::MockNighthawkServiceStub mock_nighthawk_service_stub;
+  EXPECT_CALL(mock_nighthawk_service_stub, ExecutionStreamRaw)
+      .WillOnce([](grpc_impl::ClientContext*) {
+        auto* mock_reader_writer =
+            new grpc::testing::MockClientReaderWriter<ExecutionRequest, ExecutionResponse>();
+        EXPECT_CALL(*mock_reader_writer, Read(_)).WillOnce(Return(true)).WillOnce(Return(false));
+        EXPECT_CALL(*mock_reader_writer, Write(_, _)).WillOnce(Return(true));
+        EXPECT_CALL(*mock_reader_writer, WritesDone()).WillOnce(Return(true));
+        EXPECT_CALL(*mock_reader_writer, Finish())
+            .WillOnce(Return(::grpc::Status(::grpc::UNKNOWN, "Finish failure status message")));
+        return mock_reader_writer;
+      });
+  NighthawkServiceClientImpl client;
+  absl::StatusOr<ExecutionResponse> response_or = client.PerformNighthawkBenchmark(
+      &mock_nighthawk_service_stub, CommandLineOptions(), Envoy::Protobuf::Duration());
+  ASSERT_FALSE(response_or.ok());
+  EXPECT_EQ(response_or.status().code(), absl::StatusCode::kUnknown);
+  EXPECT_THAT(response_or.status().message(), HasSubstr("Finish failure status message"));
+}
+
+} // namespace
+} // namespace Nighthawk

tools/check_format.sh

@@ -8,7 +8,7 @@ TO_CHECK="${2:-$PWD}"
 bazel run @envoy//tools:code_format/check_format.py -- \
   --skip_envoy_build_rule_check  --namespace_check Nighthawk \
   --build_fixer_check_excluded_paths=$(realpath ".") \
-  --include_dir_order envoy,nighthawk,external/source/envoy,external,api,common,source,exe,server,client,test_common,test \
+  --include_dir_order envoy,nighthawk,external/source/envoy,external,api,common,source,exe,server,client,grpcpp,test_common,test \
   $1 $TO_CHECK
 
 # The include checker doesn't support per-file checking, so we only