Adaptive concurrency no-op implementation

CODEOWNERS

@@ -44,5 +44,7 @@ extensions/filters/common/original_src @snowp @klarose
 /*/extensions/filters/http/dynamic_forward_proxy @mattklein123 @alyssawilk
 # omit_canary_hosts retry predicate
 /*/extensions/retry/host/omit_canary_hosts @sriduth @snowp
+# adaptive concurrency limit extension.
+/*/extensions/filters/http/adaptive_concurrency @tonya11en @mattklein123
 # http inspector
 /*/extensions/filters/listener/http_inspector @crazyxy @PiotrSikora @lizan

api/envoy/config/filter/http/adaptive_concurrency/v2alpha/BUILD

@@ -0,0 +1,11 @@
+load("@envoy_api//bazel:api_build_system.bzl", "api_proto_library_internal")
+
+licenses(["notice"]) # Apache 2
+
+api_proto_library_internal(
+ name = "adaptive_concurrency",
+ srcs = ["adaptive_concurrency.proto"],
+ deps = [
+ "//envoy/api/v2/core:base",
+ ],
+)

api/envoy/config/filter/http/adaptive_concurrency/v2alpha/adaptive_concurrency.proto

@@ -0,0 +1,11 @@
+syntax = "proto3";
+
+package envoy.config.filter.http.adaptive_concurrency.v2alpha;
+
+option java_package = "io.envoyproxy.envoy.config.filter.http.adaptive_concurrency.v2alpha";
+option java_outer_classname = "AdaptiveConcurrencyProto";
+option java_multiple_files = true;
+option go_package = "v2alpha";
+
+message AdaptiveConcurrency {
+}

source/extensions/extensions_build_config.bzl

@@ -29,6 +29,7 @@ EXTENSIONS = {
  # HTTP filters
  #
 
+ "envoy.filters.http.adaptive_concurrency": "//source/extensions/filters/http/adaptive_concurrency:config",
  "envoy.filters.http.buffer": "//source/extensions/filters/http/buffer:config",
  "envoy.filters.http.cors": "//source/extensions/filters/http/cors:config",
  "envoy.filters.http.csrf": "//source/extensions/filters/http/csrf:config",
@@ -37,9 +38,9 @@ EXTENSIONS = {
  "envoy.filters.http.ext_authz": "//source/extensions/filters/http/ext_authz:config",
  "envoy.filters.http.fault": "//source/extensions/filters/http/fault:config",
  "envoy.filters.http.grpc_http1_bridge": "//source/extensions/filters/http/grpc_http1_bridge:config",
+ "envoy.filters.http.grpc_http1_reverse_bridge": "//source/extensions/filters/http/grpc_http1_reverse_bridge:config",
  "envoy.filters.http.grpc_json_transcoder": "//source/extensions/filters/http/grpc_json_transcoder:config",
  "envoy.filters.http.grpc_web": "//source/extensions/filters/http/grpc_web:config",
- "envoy.filters.http.grpc_http1_reverse_bridge": "//source/extensions/filters/http/grpc_http1_reverse_bridge:config",
  "envoy.filters.http.gzip": "//source/extensions/filters/http/gzip:config",
  "envoy.filters.http.header_to_metadata": "//source/extensions/filters/http/header_to_metadata:config",
  "envoy.filters.http.health_check": "//source/extensions/filters/http/health_check:config",

source/extensions/filters/http/adaptive_concurrency/BUILD

@@ -0,0 +1,39 @@
+licenses(["notice"]) # Apache 2
+
+# HTTP L7 filter that dynamically adjusts the number of allowed concurrent
+# requests based on sampled latencies.
+# Public docs: TODO (tonya11en)
+
+load(
+ "//bazel:envoy_build_system.bzl",
+ "envoy_cc_library",
+ "envoy_package",
+)
+
+envoy_package()
+
+envoy_cc_library(
+ name = "adaptive_concurrency_filter_lib",
+ srcs = ["adaptive_concurrency_filter.cc"],
+ hdrs = ["adaptive_concurrency_filter.h"],
+ deps = [
+ "//include/envoy/http:filter_interface",
+ "//source/extensions/filters/http:well_known_names",
+ "//source/extensions/filters/http/adaptive_concurrency/concurrency_controller:concurrency_controller_lib",
+ "//source/extensions/filters/http/common:pass_through_filter_lib",
+ "@envoy_api//envoy/config/filter/http/adaptive_concurrency/v2alpha:adaptive_concurrency_cc",
+ ],
+)
+
+envoy_cc_library(
+ name = "config",
+ srcs = ["config.cc"],
+ hdrs = ["config.h"],
+ deps = [
+ "//include/envoy/registry",
+ "//source/common/config:filter_json_lib",
+ "//source/extensions/filters/http:well_known_names",
+ "//source/extensions/filters/http/adaptive_concurrency:adaptive_concurrency_filter_lib",
+ "//source/extensions/filters/http/common:factory_base_lib",
+ ],
+)

source/extensions/filters/http/adaptive_concurrency/adaptive_concurrency_filter.cc

@@ -0,0 +1,61 @@
+#include "extensions/filters/http/adaptive_concurrency/adaptive_concurrency_filter.h"
+
+#include <chrono>
+#include <cstdint>
+#include <string>
+#include <vector>
+
+#include "common/common/assert.h"
+
+#include "extensions/filters/http/adaptive_concurrency/concurrency_controller/concurrency_controller.h"
+#include "extensions/filters/http/well_known_names.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace AdaptiveConcurrency {
+
+AdaptiveConcurrencyFilterConfig::AdaptiveConcurrencyFilterConfig(
+ const envoy::config::filter::http::adaptive_concurrency::v2alpha::AdaptiveConcurrency&,
+ Runtime::Loader& runtime, std::string stats_prefix, Stats::Scope& scope,
+ TimeSource& time_source)
+ : runtime_(runtime), stats_prefix_(std::move(stats_prefix)), scope_(scope),
+ time_source_(time_source) {
+ // TODO (tonya11en): Remove these noop calls when stats/runtime values are
+ // implemented.
+ //
+ // Calling so the builds pass with these currently unused variables.
+ runtime_.snapshot();
+ scope_.constSymbolTable();
+}
+
+AdaptiveConcurrencyFilter::AdaptiveConcurrencyFilter(
+ AdaptiveConcurrencyFilterConfigSharedPtr config, ConcurrencyControllerSharedPtr controller)
+ : config_(std::move(config)), controller_(std::move(controller)) {}
+
+AdaptiveConcurrencyFilter::~AdaptiveConcurrencyFilter() = default;
+
+Http::FilterHeadersStatus AdaptiveConcurrencyFilter::decodeHeaders(Http::HeaderMap&, bool) {
+ const auto forwarding_action = controller_->forwardingDecision();
+
+ if (forwarding_action == ConcurrencyController::RequestForwardingAction::Block) {
+ // TODO (tonya11en): Remove filler words.
+ decoder_callbacks_->sendLocalReply(Http::Code::ServiceUnavailable, "filler words", nullptr,
+ absl::nullopt, "more filler words");
+ return Http::FilterHeadersStatus::StopIteration;
+ }
+
+ rq_start_time_ = config_->timeSource().monotonicTime();
+ return Http::FilterHeadersStatus::Continue;
+}
+
+void AdaptiveConcurrencyFilter::encodeComplete() {
+ const std::chrono::nanoseconds rq_latency =
+ config_->timeSource().monotonicTime() - rq_start_time_;
+ controller_->recordLatencySample(rq_latency);
+}
+
+} // namespace AdaptiveConcurrency
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/http/adaptive_concurrency/adaptive_concurrency_filter.h

@@ -0,0 +1,76 @@
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <unordered_set>
+#include <vector>
+
+#include "envoy/common/time.h"
+#include "envoy/config/filter/http/adaptive_concurrency/v2alpha/adaptive_concurrency.pb.h"
+#include "envoy/http/filter.h"
+#include "envoy/runtime/runtime.h"
+#include "envoy/stats/scope.h"
+#include "envoy/stats/stats_macros.h"
+
+#include "extensions/filters/http/adaptive_concurrency/concurrency_controller/concurrency_controller.h"
+#include "extensions/filters/http/common/pass_through_filter.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace AdaptiveConcurrency {
+
+/**
+ * Configuration for the adaptive concurrency limit filter.
+ */
+class AdaptiveConcurrencyFilterConfig {
+public:
+ AdaptiveConcurrencyFilterConfig(
+ const envoy::config::filter::http::adaptive_concurrency::v2alpha::AdaptiveConcurrency&
+ adaptive_concurrency,
+ Runtime::Loader& runtime, std::string stats_prefix, Stats::Scope& scope,
+ TimeSource& time_source);
+
+ TimeSource& timeSource() const { return time_source_; }
+
+private:
+ Runtime::Loader& runtime_;
+ const std::string stats_prefix_;
+ Stats::Scope& scope_;
+ TimeSource& time_source_;
+};
+
+using AdaptiveConcurrencyFilterConfigSharedPtr =
+ std::shared_ptr<const AdaptiveConcurrencyFilterConfig>;
+using ConcurrencyControllerSharedPtr =
+ std::shared_ptr<ConcurrencyController::ConcurrencyController>;
+
+/**
+ * A filter that samples request latencies and dynamically adjusts the request
+ * concurrency window.
+ */
+class AdaptiveConcurrencyFilter : public Http::PassThroughFilter,
+ Logger::Loggable<Logger::Id::filter> {
+public:
+ AdaptiveConcurrencyFilter(AdaptiveConcurrencyFilterConfigSharedPtr config,
+ ConcurrencyControllerSharedPtr controller);
+ ~AdaptiveConcurrencyFilter() override;
+
+ // Http::StreamDecoderFilter
+ Http::FilterHeadersStatus decodeHeaders(Http::HeaderMap&, bool) override;
+
+ // Http::StreamEncoderFilter
+ void encodeComplete() override;
+
+private:
+ AdaptiveConcurrencyFilterConfigSharedPtr config_;
+ const ConcurrencyControllerSharedPtr controller_;
+ MonotonicTime rq_start_time_;
+ std::unique_ptr<ConcurrencyController::RequestForwardingAction> forwarding_action_;
+};
+
+} // namespace AdaptiveConcurrency
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/http/adaptive_concurrency/concurrency_controller/BUILD

@@ -0,0 +1,24 @@
+licenses(["notice"]) # Apache 2
+
+# HTTP L7 filter that dynamically adjusts the number of allowed concurrent
+# requests based on sampled latencies.
+# Public docs: TODO (tonya11en)
+
+load(
+ "//bazel:envoy_build_system.bzl",
+ "envoy_cc_library",
+ "envoy_package",
+)
+
+envoy_package()
+
+envoy_cc_library(
+ name = "concurrency_controller_lib",
+ srcs = [],
+ hdrs = [
+ "concurrency_controller.h",
+ "noop_controller.h",
+ ],
+ deps = [
+ ],
+)

source/extensions/filters/http/adaptive_concurrency/concurrency_controller/concurrency_controller.h

@@ -0,0 +1,53 @@
+#pragma once
+
+#include <chrono>
+
+#include "envoy/common/pure.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace AdaptiveConcurrency {
+namespace ConcurrencyController {
+
+/**
+ * The controller's decision on whether a request will be forwarded.
+ */
+enum class RequestForwardingAction {
+ // The concurrency limit is exceeded, so the request cannot be forwarded.
+ Block,
+
+ // The controller has allowed the request through and changed its internal
+ // state. The request must be forwarded.
+ MustForward
+};
+
+/**
+ * Adaptive concurrency controller interface. All implementations of this
+ * interface must be thread-safe.
+ */
+class ConcurrencyController {
+public:
+ virtual ~ConcurrencyController() = default;
+
+ /**
+ * Called during decoding when the adaptive concurrency filter is attempting
+ * to forward a request. Returns its decision on whether to forward a request.
+ */
+ virtual RequestForwardingAction forwardingDecision() PURE;
+
+ /**
+ * Called during encoding when the request latency is known. Records the
+ * request latency to update the internal state of the controller for
+ * concurrency limit calculations.
+ *
+ * @param rq_latency is the clocked round-trip time for the request.
+ */
+ virtual void recordLatencySample(const std::chrono::nanoseconds& rq_latency) PURE;
+};
+
+} // namespace ConcurrencyController
+} // namespace AdaptiveConcurrency
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/http/adaptive_concurrency/concurrency_controller/noop_controller.h

@@ -0,0 +1,29 @@
+#pragma once
+
+#include <chrono>
+
+#include "extensions/filters/http/adaptive_concurrency/concurrency_controller/concurrency_controller.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace AdaptiveConcurrency {
+namespace ConcurrencyController {
+
+/**
+ * Adaptive concurrency controller that does nothing.
+ */
+class NoopController : public ConcurrencyController {
+public:
+ // ConcurrencyController.
+ RequestForwardingAction forwardingDecision() override {
+ return RequestForwardingAction::MustForward;
+ }
+ void recordLatencySample(const std::chrono::nanoseconds&) override {}
+};
+
+} // namespace ConcurrencyController
+} // namespace AdaptiveConcurrency
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/http/adaptive_concurrency/config.cc

@@ -0,0 +1,42 @@
+#include "extensions/filters/http/adaptive_concurrency/config.h"
+
+#include "envoy/config/filter/http/adaptive_concurrency/v2alpha/adaptive_concurrency.pb.validate.h"
+#include "envoy/registry/registry.h"
+
+#include "common/config/filter_json.h"
+
+#include "extensions/filters/http/adaptive_concurrency/adaptive_concurrency_filter.h"
+#include "extensions/filters/http/adaptive_concurrency/concurrency_controller/noop_controller.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace AdaptiveConcurrency {
+
+Http::FilterFactoryCb AdaptiveConcurrencyFilterFactory::createFilterFactoryFromProtoTyped(
+ const envoy::config::filter::http::adaptive_concurrency::v2alpha::AdaptiveConcurrency& config,
+ const std::string& stats_prefix, Server::Configuration::FactoryContext& context) {
+
+ // TODO (tonya11en): Noop controller needs to be replaced with an actual
+ // implementation in a future patch.
+ std::shared_ptr<ConcurrencyController::ConcurrencyController> noop_ctl =
+ std::make_shared<ConcurrencyController::NoopController>();
+
+ AdaptiveConcurrencyFilterConfigSharedPtr filter_config(new AdaptiveConcurrencyFilterConfig(
+ config, context.runtime(), stats_prefix, context.scope(), context.timeSource()));
+
+ return [filter_config, noop_ctl](Http::FilterChainFactoryCallbacks& callbacks) -> void {
+ callbacks.addStreamFilter(std::make_shared<AdaptiveConcurrencyFilter>(filter_config, noop_ctl));
+ };
+}
+
+/**
+ * Static registration for the adaptive_concurrency filter. @see RegisterFactory.
+ */
+REGISTER_FACTORY(AdaptiveConcurrencyFilterFactory,
+ Server::Configuration::NamedHttpFilterConfigFactory);
+
+} // namespace AdaptiveConcurrency
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy