Test server dynamic delay

BUILD

@@ -30,6 +30,7 @@ envoy_cc_binary(
     name = "nighthawk_test_server",
     repository = "@envoy",
     deps = [
+        "//source/server:http_dynamic_delay_filter_config",
         "//source/server:http_test_server_filter_config",
         "@envoy//source/exe:envoy_main_entry_lib",
     ],

api/server/response_options.proto

@@ -5,6 +5,16 @@ package nighthawk.server;
 import "google/protobuf/wrappers.proto";
 import "validate/validate.proto";
 import "envoy/api/v2/core/base.proto";
+import "google/protobuf/duration.proto";
+
+message ConcurrencyBasedDelay {
+  // Minimal delay to add to replies.
+  google.protobuf.Duration minimal_delay = 1 [(validate.rules).duration.gte.nanos = 0];
+  // Factor to use when adding latency as concurrency increases.
+  google.protobuf.Duration concurrency_delay_factor = 2 [(validate.rules).duration.gte.nanos = 0];
+  // Multiplier.
+  uint64 concurrency_delay_multiplier = 3;
+}
 
 // Options that control the test server response.
 message ResponseOptions {
@@ -14,4 +24,11 @@ message ResponseOptions {
   uint32 response_body_size = 2 [(validate.rules).uint32 = {lte: 4194304}];
   // If true, then echo request headers in the response body.
   bool echo_request_headers = 3;
+
+  oneof oneof_delay_options {
+    // Static delay duration.
+    google.protobuf.Duration static_delay = 4 [(validate.rules).duration.gte.nanos = 0];
+    // Concurrency based delay configuration.
+    ConcurrencyBasedDelay concurrency_based_delay = 5;
+  }
 }

source/server/BUILD

@@ -8,15 +8,39 @@ licenses(["notice"])  # Apache 2
 
 envoy_package()
 
+envoy_cc_library(
+    name = "common_lib",
+    srcs = ["common.cc"],
+    hdrs = ["common.h"],
+    repository = "@envoy",
+    deps = [
+        "//api/server:response_options_proto_cc_proto",
+        "@envoy//source/common/protobuf:message_validator_lib_with_external_headers",
+        "@envoy//source/common/protobuf:utility_lib_with_external_headers",
+        "@envoy//source/common/singleton:const_singleton_with_external_headers",
+    ],
+)
+
 envoy_cc_library(
     name = "http_test_server_filter_lib",
     srcs = ["http_test_server_filter.cc"],
     hdrs = ["http_test_server_filter.h"],
     repository = "@envoy",
     deps = [
+        ":common_lib",
+        "//api/server:response_options_proto_cc_proto",
+        "@envoy//source/exe:envoy_common_lib_with_external_headers",
+    ],
+)
+
+envoy_cc_library(
+    name = "http_dynamic_delay_filter_lib",
+    srcs = ["http_dynamic_delay_filter.cc"],
+    hdrs = ["http_dynamic_delay_filter.h"],
+    repository = "@envoy",
+    deps = [
+        ":common_lib",
         "//api/server:response_options_proto_cc_proto",
-        "@envoy//source/common/protobuf:message_validator_lib_with_external_headers",
-        "@envoy//source/common/protobuf:utility_lib_with_external_headers",
         "@envoy//source/exe:envoy_common_lib_with_external_headers",
     ],
 )
@@ -30,3 +54,13 @@ envoy_cc_library(
         "@envoy//include/envoy/server:filter_config_interface_with_external_headers",
     ],
 )
+
+envoy_cc_library(
+    name = "http_dynamic_delay_filter_config",
+    srcs = ["http_dynamic_delay_filter_config.cc"],
+    repository = "@envoy",
+    deps = [
+        ":http_dynamic_delay_filter_lib",
+        "@envoy//include/envoy/server:filter_config_interface_with_external_headers",
+    ],
+)

source/server/common.cc

@@ -0,0 +1,43 @@
+#include "server/common.h"
+
+#include <string>
+
+#include "external/envoy/source/common/protobuf/message_validator_impl.h"
+#include "external/envoy/source/common/protobuf/utility.h"
+
+#include "api/server/response_options.pb.validate.h"
+
+#include "absl/strings/numbers.h"
+
+namespace Nighthawk {
+namespace Server {
+
+bool Utility::mergeJsonConfig(absl::string_view json, nighthawk::server::ResponseOptions& config,
+                              absl::optional<std::string>& error_message) {
+  error_message = absl::nullopt;
+  try {
+    nighthawk::server::ResponseOptions json_config;
+    auto& validation_visitor = Envoy::ProtobufMessage::getStrictValidationVisitor();
+    Envoy::MessageUtil::loadFromJson(std::string(json), json_config, validation_visitor);
+    config.MergeFrom(json_config);
+    Envoy::MessageUtil::validate(config, validation_visitor);
+  } catch (const Envoy::EnvoyException& exception) {
+    error_message.emplace(fmt::format("Error merging json config: {}", exception.what()));
+  }
+  return error_message == absl::nullopt;
+}
+
+void Utility::applyConfigToResponseHeaders(Envoy::Http::ResponseHeaderMap& response_headers,
+                                           nighthawk::server::ResponseOptions& response_options) {
+  for (const auto& header_value_option : response_options.response_headers()) {
+    const auto& header = header_value_option.header();
+    auto lower_case_key = Envoy::Http::LowerCaseString(header.key());
+    if (!header_value_option.append().value()) {
+      response_headers.remove(lower_case_key);
+    }
+    response_headers.addCopy(lower_case_key, header.value());
+  }
+}
+
+} // namespace Server
+} // namespace Nighthawk

source/server/common.h

@@ -0,0 +1,50 @@
+#pragma once
+
+#include <string>
+
+#include "envoy/http/header_map.h"
+
+#include "external/envoy/source/common/singleton/const_singleton.h"
+
+#include "api/server/response_options.pb.h"
+
+namespace Nighthawk {
+namespace Server {
+
+namespace TestServer {
+
+class HeaderNameValues {
+public:
+  const Envoy::Http::LowerCaseString TestServerConfig{"x-nighthawk-test-server-config"};
+};
+
+using HeaderNames = Envoy::ConstSingleton<HeaderNameValues>;
+
+} // namespace TestServer
+
+class Utility {
+public:
+  /**
+   * Merges a json string containing configuration into a ResponseOptions instance.
+   *
+   * @param json Json-formatted seralization of ResponseOptions to merge into the configuration.
+   * @param config The target that the json string should be merged into.
+   * @param error_message Will contain an error message iff an error occurred.
+   * @return bool false iff an error occurred.
+   */
+  static bool mergeJsonConfig(absl::string_view json, nighthawk::server::ResponseOptions& config,
+                              absl::optional<std::string>& error_message);
+
+  /**
+   * Applies ResponseOptions onto a HeaderMap containing response headers.
+   *
+   * @param response_headers Response headers to transform to reflect the passed in response
+   * options.
+   * @param response_options Configuration specifying how to transform the header map.
+   */
+  static void applyConfigToResponseHeaders(Envoy::Http::ResponseHeaderMap& response_headers,
+                                           nighthawk::server::ResponseOptions& response_options);
+};
+
+} // namespace Server
+} // namespace Nighthawk

source/server/http_dynamic_delay_filter.cc

@@ -0,0 +1,77 @@
+#include "server/http_dynamic_delay_filter.h"
+
+#include <string>
+
+#include "envoy/server/filter_config.h"
+
+#include "server/common.h"
+
+#include "absl/strings/str_cat.h"
+
+namespace Nighthawk {
+namespace Server {
+
+std::atomic<uint64_t> HttpDynamicDelayDecoderFilterConfig::instances_(0);
+
+HttpDynamicDelayDecoderFilterConfig::HttpDynamicDelayDecoderFilterConfig(
+    nighthawk::server::ResponseOptions proto_config)
+    : server_config_(std::move(proto_config)) {}
+
+HttpDynamicDelayDecoderFilter::HttpDynamicDelayDecoderFilter(
+    HttpDynamicDelayDecoderFilterConfigSharedPtr config)
+    : config_(std::move(config)) {
+  config_->incrementInstanceCount();
+}
+
+void HttpDynamicDelayDecoderFilter::onDestroy() { config_->decrementInstanceCount(); }
+
+Envoy::Http::FilterHeadersStatus
+HttpDynamicDelayDecoderFilter::decodeHeaders(Envoy::Http::RequestHeaderMap& headers, bool) {
+  base_config_ = config_->server_config();
+  const auto* request_config_header = headers.get(TestServer::HeaderNames::get().TestServerConfig);
+  if (request_config_header) {
+    Utility::mergeJsonConfig(request_config_header->value().getStringView(), base_config_,
+                             error_message_);
+  }
+  if (error_message_.has_value()) {
+    decoder_callbacks_->sendLocalReply(
+        static_cast<Envoy::Http::Code>(500),
+        fmt::format("dynamic-delay didn't understand the request: {}", *error_message_), nullptr,
+        absl::nullopt, "");
+    return Envoy::Http::FilterHeadersStatus::StopIteration;
+  }
+  absl::optional<int64_t> delay;
+  if (base_config_.has_static_delay()) {
+    delay = Envoy::Protobuf::util::TimeUtil::DurationToMilliseconds(base_config_.static_delay());
+  } else if (base_config_.has_concurrency_based_delay()) {
+    auto& concurrency = base_config_.concurrency_based_delay();
+    const uint64_t current_value = config_->approximateInstances();
+    delay = Envoy::Protobuf::util::TimeUtil::DurationToMilliseconds(
+        concurrency.minimal_delay() + (current_value * concurrency.concurrency_delay_factor()));
+    std::cerr << current_value << ":" << delay.value() << std::endl;
+  }
+  if (delay.has_value() && delay > 0) {
+    // Emit header to communicate the delay we desire to the fault filter extension.
+    const Envoy::Http::LowerCaseString key("x-envoy-fault-delay-request");
+    headers.setCopy(key, absl::StrCat(*delay));
+  }
+  return Envoy::Http::FilterHeadersStatus::Continue;
+}
+
+Envoy::Http::FilterDataStatus HttpDynamicDelayDecoderFilter::decodeData(Envoy::Buffer::Instance&,
+                                                                        bool) {
+  return Envoy::Http::FilterDataStatus::Continue;
+}
+
+Envoy::Http::FilterTrailersStatus
+HttpDynamicDelayDecoderFilter::decodeTrailers(Envoy::Http::RequestTrailerMap&) {
+  return Envoy::Http::FilterTrailersStatus::Continue;
+}
+
+void HttpDynamicDelayDecoderFilter::setDecoderFilterCallbacks(
+    Envoy::Http::StreamDecoderFilterCallbacks& callbacks) {
+  decoder_callbacks_ = &callbacks;
+}
+
+} // namespace Server
+} // namespace Nighthawk

source/server/http_dynamic_delay_filter.h

@@ -0,0 +1,51 @@
+#pragma once
+
+#include <atomic>
+#include <string>
+
+#include "envoy/server/filter_config.h"
+
+#include "api/server/response_options.pb.h"
+
+namespace Nighthawk {
+namespace Server {
+
+// Basically this is left in as a placeholder for further configuration.
+class HttpDynamicDelayDecoderFilterConfig {
+public:
+  HttpDynamicDelayDecoderFilterConfig(nighthawk::server::ResponseOptions proto_config);
+  const nighthawk::server::ResponseOptions& server_config() { return server_config_; }
+  void incrementInstanceCount() { instances_++; }
+  void decrementInstanceCount() { instances_--; }
+  uint64_t approximateInstances() const { return instances_; }
+
+private:
+  const nighthawk::server::ResponseOptions server_config_;
+  static std::atomic<uint64_t> instances_;
+};
+
+using HttpDynamicDelayDecoderFilterConfigSharedPtr =
+    std::shared_ptr<HttpDynamicDelayDecoderFilterConfig>;
+
+class HttpDynamicDelayDecoderFilter : public Envoy::Http::StreamDecoderFilter {
+public:
+  HttpDynamicDelayDecoderFilter(HttpDynamicDelayDecoderFilterConfigSharedPtr);
+
+  // Http::StreamFilterBase
+  void onDestroy() override;
+
+  // Http::StreamDecoderFilter
+  Envoy::Http::FilterHeadersStatus decodeHeaders(Envoy::Http::RequestHeaderMap&, bool) override;
+  Envoy::Http::FilterDataStatus decodeData(Envoy::Buffer::Instance&, bool) override;
+  Envoy::Http::FilterTrailersStatus decodeTrailers(Envoy::Http::RequestTrailerMap&) override;
+  void setDecoderFilterCallbacks(Envoy::Http::StreamDecoderFilterCallbacks&) override;
+
+private:
+  const HttpDynamicDelayDecoderFilterConfigSharedPtr config_;
+  Envoy::Http::StreamDecoderFilterCallbacks* decoder_callbacks_;
+  nighthawk::server::ResponseOptions base_config_;
+  absl::optional<std::string> error_message_;
+};
+
+} // namespace Server
+} // namespace Nighthawk

source/server/http_dynamic_delay_filter_config.cc

@@ -0,0 +1,55 @@
+#include <string>
+
+#include "envoy/registry/registry.h"
+
+#include "external/envoy/source/common/protobuf/message_validator_impl.h"
+
+#include "api/server/response_options.pb.h"
+#include "api/server/response_options.pb.validate.h"
+
+#include "server/http_dynamic_delay_filter.h"
+
+namespace Nighthawk {
+namespace Server {
+namespace Configuration {
+
+class HttpDynamicDelayDecoderFilterConfig
+    : public Envoy::Server::Configuration::NamedHttpFilterConfigFactory {
+public:
+  Envoy::Http::FilterFactoryCb
+  createFilterFactoryFromProto(const Envoy::Protobuf::Message& proto_config, const std::string&,
+                               Envoy::Server::Configuration::FactoryContext& context) override {
+
+    auto& validation_visitor = Envoy::ProtobufMessage::getStrictValidationVisitor();
+    return createFilter(
+        Envoy::MessageUtil::downcastAndValidate<const nighthawk::server::ResponseOptions&>(
+            proto_config, validation_visitor),
+        context);
+  }
+
+  Envoy::ProtobufTypes::MessagePtr createEmptyConfigProto() override {
+    return Envoy::ProtobufTypes::MessagePtr{new nighthawk::server::ResponseOptions()};
+  }
+
+  std::string name() const override { return "dynamic-delay"; }
+
+private:
+  Envoy::Http::FilterFactoryCb createFilter(const nighthawk::server::ResponseOptions& proto_config,
+                                            Envoy::Server::Configuration::FactoryContext&) {
+    Nighthawk::Server::HttpDynamicDelayDecoderFilterConfigSharedPtr config =
+        std::make_shared<Nighthawk::Server::HttpDynamicDelayDecoderFilterConfig>(
+            Nighthawk::Server::HttpDynamicDelayDecoderFilterConfig(proto_config));
+
+    return [config](Envoy::Http::FilterChainFactoryCallbacks& callbacks) -> void {
+      auto* filter = new Nighthawk::Server::HttpDynamicDelayDecoderFilter(config);
+      callbacks.addStreamDecoderFilter(Envoy::Http::StreamDecoderFilterSharedPtr{filter});
+    };
+  }
+};
+
+static Envoy::Registry::RegisterFactory<HttpDynamicDelayDecoderFilterConfig,
+                                        Envoy::Server::Configuration::NamedHttpFilterConfigFactory>
+    register_;
+} // namespace Configuration
+} // namespace Server
+} // namespace Nighthawk