Adaptive Load library for calling Nighthawk Service (envoyproxy#493)

A library that calls a Nighthawk Service gRPC stub with the given `CommandLineOptions`, translating all possible gRPC errors into `absl::StatusOr`.

Will need to be updated when Nighthawk Service starts returning more than one message over the stream.

Part 4 of splitting PR envoyproxy#483.

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/common/BUILD

@@ -37,6 +37,22 @@ envoy_basic_cc_library(
     ],
 )
 
+envoy_basic_cc_library(
+    name = "nighthawk_service_client",
+    hdrs = [
+        "nighthawk_service_client.h",
+    ],
+    include_prefix = "nighthawk/common",
+    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 = "request_lib",
     hdrs = [

include/nighthawk/common/nighthawk_service_client.h

@@ -0,0 +1,36 @@
+#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.
+   *
+   * @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) PURE;
+};
+
+} // namespace Nighthawk

source/common/BUILD

@@ -8,6 +8,26 @@ licenses(["notice"])  # Apache 2
 
 envoy_package()
 
+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/common: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 = "request_impl_lib",
     hdrs = [

source/common/nighthawk_service_client_impl.cc

@@ -0,0 +1,42 @@
+#include "common/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) {
+  nighthawk::client::ExecutionRequest request;
+  nighthawk::client::ExecutionResponse response;
+  *request.mutable_start_request()->mutable_options() = command_line_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::UnavailableError("Failed to write request to the Nighthawk Service gRPC channel.");
+  } else if (!stream->WritesDone()) {
+    return absl::InternalError("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::InternalError("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/common/nighthawk_service_client_impl.h

@@ -0,0 +1,25 @@
+#include "nighthawk/common/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 {
+
+/**
+ * Real implementation of a helper that opens a channel with the gRPC stub, sends the input, and
+ * translates the output or errors into a StatusOr.
+ *
+ * This class is stateless and may be called from multiple threads. Furthermore, the same gRPC stub
+ * is safe to use from multiple threads simultaneously.
+ */
+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) override;
+};
+
+} // namespace Nighthawk

test/common/BUILD

@@ -28,3 +28,13 @@ envoy_cc_test(
         ":fake_time_source",
     ],
 )
+
+envoy_cc_test(
+    name = "nighthawk_service_client_test",
+    srcs = ["nighthawk_service_client_test.cc"],
+    repository = "@envoy",
+    deps = [
+        "//source/common:nighthawk_service_client_impl",
+        "@com_github_grpc_grpc//:grpc++_test",
+    ],
+)

test/common/nighthawk_service_client_test.cc

@@ -0,0 +1,178 @@
+#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 "common/nighthawk_service_client_impl.h"
+
+#include "grpcpp/test/mock_stream.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, UsesSpecifiedCommandLineOptions) {
+  const int kExpectedRps = 456;
+  ExecutionRequest request;
+  nighthawk::client::MockNighthawkServiceStub mock_nighthawk_service_stub;
+  // Configure the mock Nighthawk Service stub to return an inner mock channel when the code under
+  // test requests a channel. Set call expectations on the inner mock channel.
+  EXPECT_CALL(mock_nighthawk_service_stub, ExecutionStreamRaw)
+      .WillOnce([&request](grpc_impl::ClientContext*) {
+        auto* mock_reader_writer =
+            new grpc::testing::MockClientReaderWriter<ExecutionRequest, ExecutionResponse>();
+        // PerformNighthawkBenchmark currently expects Read to return true exactly once.
+        EXPECT_CALL(*mock_reader_writer, Read(_)).WillOnce(Return(true)).WillOnce(Return(false));
+        // Capture the Nighthawk request PerformNighthawkBenchmark sends on the channel.
+        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);
+  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;
+  // Configure the mock Nighthawk Service stub to return an inner mock channel when the code under
+  // test requests a channel. Set call expectations on the inner mock channel.
+  EXPECT_CALL(mock_nighthawk_service_stub, ExecutionStreamRaw)
+      .WillOnce([&expected_response](grpc_impl::ClientContext*) {
+        auto* mock_reader_writer =
+            new grpc::testing::MockClientReaderWriter<ExecutionRequest, ExecutionResponse>();
+        // PerformNighthawkBenchmark currently expects Read to return true exactly once.
+        // Capture the gRPC response proto as it is written to the output parameter.
+        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());
+  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;
+  // Configure the mock Nighthawk Service stub to return an inner mock channel when the code under
+  // test requests a channel. Set call expectations on the inner mock channel.
+  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());
+  ASSERT_FALSE(response_or.ok());
+  EXPECT_EQ(response_or.status().code(), absl::StatusCode::kInternal);
+  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;
+  // Configure the mock Nighthawk Service stub to return an inner mock channel when the code under
+  // test requests a channel. Set call expectations on the inner mock channel.
+  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());
+  ASSERT_FALSE(response_or.ok());
+  EXPECT_EQ(response_or.status().code(), absl::StatusCode::kUnavailable);
+  EXPECT_THAT(response_or.status().message(), HasSubstr("Failed to write"));
+}
+
+TEST(PerformNighthawkBenchmark, ReturnsErrorIfNighthawkServiceWritesDoneFails) {
+  nighthawk::client::MockNighthawkServiceStub mock_nighthawk_service_stub;
+  // Configure the mock Nighthawk Service stub to return an inner mock channel when the code under
+  // test requests a channel. Set call expectations on the inner mock channel.
+  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());
+  ASSERT_FALSE(response_or.ok());
+  EXPECT_EQ(response_or.status().code(), absl::StatusCode::kInternal);
+  EXPECT_THAT(response_or.status().message(), HasSubstr("WritesDone() failed"));
+}
+
+TEST(PerformNighthawkBenchmark, PropagatesErrorIfNighthawkServiceGrpcStreamClosesAbnormally) {
+  nighthawk::client::MockNighthawkServiceStub mock_nighthawk_service_stub;
+  // Configure the mock Nighthawk Service stub to return an inner mock channel when the code under
+  // test requests a channel. Set call expectations on the inner mock channel.
+  EXPECT_CALL(mock_nighthawk_service_stub, ExecutionStreamRaw)
+      .WillOnce([](grpc_impl::ClientContext*) {
+        auto* mock_reader_writer =
+            new grpc::testing::MockClientReaderWriter<ExecutionRequest, ExecutionResponse>();
+        // PerformNighthawkBenchmark currently expects Read to return true exactly once.
+        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::PERMISSION_DENIED, "Finish failure status message")));
+        return mock_reader_writer;
+      });
+
+  NighthawkServiceClientImpl client;
+  absl::StatusOr<ExecutionResponse> response_or =
+      client.PerformNighthawkBenchmark(&mock_nighthawk_service_stub, CommandLineOptions());
+  ASSERT_FALSE(response_or.ok());
+  EXPECT_EQ(response_or.status().code(), absl::StatusCode::kPermissionDenied);
+  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