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/filters/http/adaptive_concurrency/BUILD

@@ -0,0 +1,38 @@
+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",
+ "@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,64 @@
+#include "extensions/filters/http/adaptive_concurrency/adaptive_concurrency_filter.h"
+
+#include <chrono>
+#include <cstdint>
+#include <string>
+#include <vector>
+
+#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) {}
+
+AdaptiveConcurrencyFilter::AdaptiveConcurrencyFilter(
+ AdaptiveConcurrencyFilterConfigSharedPtr config, ConcurrencyControllerSharedPtr controller)
+ : config_(std::move(config)), controller_(std::move(controller)) {}
+
+AdaptiveConcurrencyFilter::~AdaptiveConcurrencyFilter() = default;
+
+Http::FilterHeadersStatus AdaptiveConcurrencyFilter::decodeHeaders(Http::HeaderMap&,
+ bool end_stream) {
+ if (!end_stream) {
+ return Http::FilterHeadersStatus::Continue;
+ }
+
+ if (controller_->tryForwardRequest()) {
+ rq_start_time_ = config_->timeSource().monotonicTime();
+ return Http::FilterHeadersStatus::Continue;
+ }
+
+ // TODO (tonya11en): Remove filler words.
+ decoder_callbacks_->sendLocalReply(Http::Code::ServiceUnavailable, "filler words", nullptr,
+ absl::nullopt, "more filler words");
+ return Http::FilterHeadersStatus::StopIteration;
+}
+
+void AdaptiveConcurrencyFilter::onDestroy() {
+ // TODO (tonya11en).
+}
+
+Http::FilterHeadersStatus AdaptiveConcurrencyFilter::encodeHeaders(Http::HeaderMap&,
+ bool end_stream) {
+ if (end_stream) {
+ const std::chrono::nanoseconds rq_latency =
+ config_->timeSource().monotonicTime() - rq_start_time_;
+ controller_->recordLatencySample(rq_latency);
+ }
+
+ return Http::FilterHeadersStatus::Continue;
+}
+
+} // namespace AdaptiveConcurrency
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

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

@@ -0,0 +1,104 @@
+#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"
+
+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);
+
+ Runtime::Loader& runtime() { return runtime_; }
+ const std::string& statsPrefix() const { return stats_prefix_; }
+ Stats::Scope& scope() { return scope_; }
+ TimeSource& timeSource() { return time_source_; }
+
+private:
+ Runtime::Loader& runtime_;
+ const std::string stats_prefix_;
+ Stats::Scope& scope_;
+ TimeSource& time_source_;
+};
+
+using AdaptiveConcurrencyFilterConfigSharedPtr = std::shared_ptr<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::StreamFilter, Logger::Loggable<Logger::Id::filter> {
+public:
+ AdaptiveConcurrencyFilter(AdaptiveConcurrencyFilterConfigSharedPtr config,
+ ConcurrencyControllerSharedPtr controller);
+ ~AdaptiveConcurrencyFilter() override;
+
+ // Http::StreamFilterBase
+ void onDestroy() override;
+
+ // Http::StreamDecoderFilter
+ Http::FilterHeadersStatus decodeHeaders(Http::HeaderMap& headers, bool end_stream) override;
+ Http::FilterDataStatus decodeData(Buffer::Instance&, bool) override {
+ return Http::FilterDataStatus::Continue;
+ }
+ Http::FilterTrailersStatus decodeTrailers(Http::HeaderMap&) override {
+ return Http::FilterTrailersStatus::Continue;
+ }
+ void setDecoderFilterCallbacks(Http::StreamDecoderFilterCallbacks& callbacks) override {
+ decoder_callbacks_ = &callbacks;
+ }
+
+ // Http::StreamEncoderFilter
+ Http::FilterHeadersStatus encode100ContinueHeaders(Http::HeaderMap&) override {
+ return Http::FilterHeadersStatus::Continue;
+ }
+ Http::FilterHeadersStatus encodeHeaders(Http::HeaderMap&, bool) override;
+ Http::FilterDataStatus encodeData(Buffer::Instance&, bool) override {
+ return Http::FilterDataStatus::Continue;
+ }
+ Http::FilterTrailersStatus encodeTrailers(Http::HeaderMap&) override {
+ return Http::FilterTrailersStatus::Continue;
+ }
+ Http::FilterMetadataStatus encodeMetadata(Http::MetadataMap&) override {
+ return Http::FilterMetadataStatus::Continue;
+ }
+ void setEncoderFilterCallbacks(Http::StreamEncoderFilterCallbacks& callbacks) override {
+ encoder_callbacks_ = &callbacks;
+ }
+
+private:
+ AdaptiveConcurrencyFilterConfigSharedPtr config_;
+ Http::StreamDecoderFilterCallbacks* decoder_callbacks_{};
+ Http::StreamEncoderFilterCallbacks* encoder_callbacks_{};
+ ConcurrencyControllerSharedPtr controller_;
+ MonotonicTime rq_start_time_;
+};
+
+} // 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,42 @@
+#pragma once
+
+#include <chrono>
+
+#include "envoy/common/pure.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace AdaptiveConcurrency {
+namespace ConcurrencyController {
+
+/**
+ * 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 true once the controller's internal state is
+ * updated and the request can be forwarded.
+ */
+ virtual bool tryForwardRequest() 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,27 @@
+#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.
+ bool tryForwardRequest() override { return true; }
+ 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