network filters: add tcp bandwidth limit

Signed-off-by: Anton Kanugalawattage <[email protected]>

Author: AntonKanug

CODEOWNERS

@@ -171,6 +171,7 @@ extensions/filters/common/original_src @klarose @mattklein123
 /*/extensions/retry/priority/previous_priorities @ravenblackx @mattklein123
 /*/extensions/retry/host @ravenblackx @mattklein123
 /*/extensions/filters/network/http_connection_manager @yanavlasov @mattklein123
+/*/extensions/filters/network/tcp_bandwidth_limit @UNOWNED @UNOWNED
 /*/extensions/filters/network/tcp_proxy @zuercher @ggreenway
 /*/extensions/filters/network/echo @yanavlasov @adisuissa
 /*/extensions/filters/udp/dns_filter @mattklein123 @yanjunxiang-google

api/envoy/extensions/filters/network/tcp_bandwidth_limit/v3/BUILD

@@ -0,0 +1,12 @@
+# DO NOT EDIT. This file is generated by tools/proto_format/proto_sync.py.
+
+load("@envoy_api//bazel:api_build_system.bzl", "api_proto_package")
+
+licenses(["notice"])  # Apache 2
+
+api_proto_package(
+    deps = [
+        "//envoy/config/core/v3:pkg",
+        "@com_github_cncf_xds//udpa/annotations:pkg",
+    ],
+)

api/envoy/extensions/filters/network/tcp_bandwidth_limit/v3/tcp_bandwidth_limit.proto

@@ -0,0 +1,44 @@
+syntax = "proto3";
+
+package envoy.extensions.filters.network.tcp_bandwidth_limit.v3;
+
+import "envoy/config/core/v3/base.proto";
+import "google/protobuf/duration.proto";
+import "google/protobuf/wrappers.proto";
+
+import "udpa/annotations/status.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.filters.network.tcp_bandwidth_limit.v3";
+option java_outer_classname = "TcpBandwidthLimitProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/tcp_bandwidth_limit/v3;tcp_bandwidth_limitv3";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: TCP Bandwidth Limit]
+// TCP Bandwidth Limit :ref:`configuration overview <config_network_filters_tcp_bandwidth_limit>`.
+// [#extension: envoy.filters.network.tcp_bandwidth_limit]
+
+message TcpBandwidthLimit {
+  // The human readable prefix to use when emitting stats.
+  string stat_prefix = 1 [(validate.rules).string = {min_len: 1}];
+
+  // The limit for download bandwidth in KiB/s.
+  // If not set, no limit is applied (unlimited bandwidth).
+  google.protobuf.UInt64Value download_limit_kbps = 2;
+
+  // The limit for upload bandwidth in KiB/s.
+  // If not set, no limit is applied (unlimited bandwidth).
+  google.protobuf.UInt64Value upload_limit_kbps = 3;
+
+  // Optional Fill interval in milliseconds for the token refills. Defaults to 50ms.
+  // It must be at least 20ms to avoid too aggressive refills.
+  google.protobuf.Duration fill_interval = 4 [(validate.rules).duration = {
+    lte {seconds: 1}
+    gte {nanos: 20000000}
+  }];
+
+  // Runtime flag that controls whether the filter is enabled or not. If not specified, defaults
+  // to enabled.
+  config.core.v3.RuntimeFeatureFlag runtime_enabled = 5;
+}

api/versioning/BUILD

@@ -211,6 +211,7 @@ proto_library(
         "//envoy/extensions/filters/network/set_filter_state/v3:pkg",
         "//envoy/extensions/filters/network/sni_cluster/v3:pkg",
         "//envoy/extensions/filters/network/sni_dynamic_forward_proxy/v3:pkg",
+        "//envoy/extensions/filters/network/tcp_bandwidth_limit/v3:pkg",
         "//envoy/extensions/filters/network/tcp_proxy/v3:pkg",
         "//envoy/extensions/filters/network/thrift_proxy/filters/header_to_metadata/v3:pkg",
         "//envoy/extensions/filters/network/thrift_proxy/filters/payload_to_metadata/v3:pkg",

source/extensions/extensions_build_config.bzl

@@ -249,6 +249,7 @@ EXTENSIONS = {
     "envoy.filters.network.http_connection_manager":              "//source/extensions/filters/network/http_connection_manager:config",
     "envoy.filters.network.local_ratelimit":                      "//source/extensions/filters/network/local_ratelimit:config",
     "envoy.filters.network.mongo_proxy":                          "//source/extensions/filters/network/mongo_proxy:config",
+    "envoy.filters.network.tcp_bandwidth_limit":                  "//source/extensions/filters/network/tcp_bandwidth_limit:config",
     "envoy.filters.network.ratelimit":                            "//source/extensions/filters/network/ratelimit:config",
     "envoy.filters.network.rbac":                                 "//source/extensions/filters/network/rbac:config",
     "envoy.filters.network.redis_proxy":                          "//source/extensions/filters/network/redis_proxy:config",

source/extensions/extensions_metadata.yaml

@@ -896,6 +896,13 @@ envoy.filters.network.sni_dynamic_forward_proxy:
   status: alpha
   type_urls:
   - envoy.extensions.filters.network.sni_dynamic_forward_proxy.v3.FilterConfig
+envoy.filters.network.tcp_bandwidth_limit:
+  categories:
+  - envoy.filters.network
+  security_posture: robust_to_untrusted_downstream
+  status: alpha
+  type_urls:
+  - envoy.extensions.filters.network.tcp_bandwidth_limit.v3.TcpBandwidthLimit
 envoy.filters.network.tcp_proxy:
   categories:
   - envoy.filters.network

source/extensions/filters/network/tcp_bandwidth_limit/BUILD

@@ -0,0 +1,47 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_extension",
+    "envoy_cc_library",
+    "envoy_extension_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_extension_package()
+
+envoy_cc_library(
+    name = "tcp_bandwidth_limit",
+    srcs = ["tcp_bandwidth_limit.cc"],
+    hdrs = ["tcp_bandwidth_limit.h"],
+    deps = [
+        "//envoy/buffer:buffer_interface",
+        "//envoy/common:time_interface",
+        "//envoy/event:dispatcher_interface",
+        "//envoy/event:timer_interface",
+        "//envoy/network:connection_interface",
+        "//envoy/network:filter_interface",
+        "//envoy/runtime:runtime_interface",
+        "//envoy/stats:stats_interface",
+        "//envoy/stats:stats_macros",
+        "//source/common/buffer:buffer_lib",
+        "//source/common/common:assert_lib",
+        "//source/common/common:logger_lib",
+        "//source/common/common:shared_token_bucket_impl_lib",
+        "//source/common/runtime:runtime_lib",
+        "@envoy_api//envoy/extensions/filters/network/tcp_bandwidth_limit/v3:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_extension(
+    name = "config",
+    srcs = ["config.cc"],
+    hdrs = ["config.h"],
+    deps = [
+        ":tcp_bandwidth_limit",
+        "//envoy/registry",
+        "//envoy/server:filter_config_interface",
+        "//source/extensions/filters/network:well_known_names",
+        "//source/extensions/filters/network/common:factory_base_lib",
+        "@envoy_api//envoy/extensions/filters/network/tcp_bandwidth_limit/v3:pkg_cc_proto",
+    ],
+)

source/extensions/filters/network/tcp_bandwidth_limit/config.cc

@@ -0,0 +1,34 @@
+#include "source/extensions/filters/network/tcp_bandwidth_limit/config.h"
+
+#include "envoy/extensions/filters/network/tcp_bandwidth_limit/v3/tcp_bandwidth_limit.pb.h"
+#include "envoy/extensions/filters/network/tcp_bandwidth_limit/v3/tcp_bandwidth_limit.pb.validate.h"
+
+#include "source/extensions/filters/network/tcp_bandwidth_limit/tcp_bandwidth_limit.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace TcpBandwidthLimit {
+
+Network::FilterFactoryCb TcpBandwidthLimitConfigFactory::createFilterFactoryFromProtoTyped(
+    const envoy::extensions::filters::network::tcp_bandwidth_limit::v3::TcpBandwidthLimit&
+        proto_config,
+    Server::Configuration::FactoryContext& context) {
+
+  auto config = std::make_shared<FilterConfig>(proto_config, context.scope(),
+                                               context.serverFactoryContext().runtime(),
+                                               context.serverFactoryContext().timeSource());
+
+  return [config, &context](Network::FilterManager& filter_manager) -> void {
+    filter_manager.addFilter(std::make_shared<TcpBandwidthLimitFilter>(
+        config, context.serverFactoryContext().mainThreadDispatcher()));
+  };
+}
+
+REGISTER_FACTORY(TcpBandwidthLimitConfigFactory,
+                 Server::Configuration::NamedNetworkFilterConfigFactory);
+
+} // namespace TcpBandwidthLimit
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/network/tcp_bandwidth_limit/config.h

@@ -0,0 +1,29 @@
+#pragma once
+
+#include "envoy/extensions/filters/network/tcp_bandwidth_limit/v3/tcp_bandwidth_limit.pb.h"
+#include "envoy/extensions/filters/network/tcp_bandwidth_limit/v3/tcp_bandwidth_limit.pb.validate.h"
+
+#include "source/extensions/filters/network/common/factory_base.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace TcpBandwidthLimit {
+
+class TcpBandwidthLimitConfigFactory
+    : public Common::FactoryBase<
+          envoy::extensions::filters::network::tcp_bandwidth_limit::v3::TcpBandwidthLimit> {
+public:
+  TcpBandwidthLimitConfigFactory() : FactoryBase("envoy.filters.network.tcp_bandwidth_limit") {}
+
+private:
+  Network::FilterFactoryCb createFilterFactoryFromProtoTyped(
+      const envoy::extensions::filters::network::tcp_bandwidth_limit::v3::TcpBandwidthLimit&
+          proto_config,
+      Server::Configuration::FactoryContext& context) override;
+};
+
+} // namespace TcpBandwidthLimit
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/network/tcp_bandwidth_limit/tcp_bandwidth_limit.cc

@@ -0,0 +1,179 @@
+#include "source/extensions/filters/network/tcp_bandwidth_limit/tcp_bandwidth_limit.h"
+
+#include "envoy/buffer/buffer.h"
+#include "envoy/event/dispatcher.h"
+#include "envoy/network/connection.h"
+
+#include "source/common/common/assert.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace NetworkFilters {
+namespace TcpBandwidthLimit {
+
+FilterConfig::FilterConfig(
+    const envoy::extensions::filters::network::tcp_bandwidth_limit::v3::TcpBandwidthLimit& config,
+    Stats::Scope& scope, Runtime::Loader& runtime, TimeSource& time_source)
+    : runtime_(runtime), time_source_(time_source),
+      has_download_limit_(config.has_download_limit_kbps()),
+      has_upload_limit_(config.has_upload_limit_kbps()),
+      download_limit_kbps_(PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, download_limit_kbps, 0)),
+      upload_limit_kbps_(PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, upload_limit_kbps, 0)),
+      fill_interval_(std::chrono::milliseconds(
+          PROTOBUF_GET_MS_OR_DEFAULT(config, fill_interval, 50))), // Default 50ms
+      enabled_(config.runtime_enabled(), runtime),
+      stats_(generateStats(config.stat_prefix(), scope)) {
+
+  if (has_download_limit_) {
+    // The token bucket is configured with a max token count of the number of
+    // bytes per second, and refills at the same rate, so that we have a per
+    // second limit which refills gradually over the fill interval.
+    uint64_t limit_bytes_per_sec = kiloBytesToBytes(download_limit_kbps_);
+    download_token_bucket_ = std::make_shared<SharedTokenBucketImpl>(
+        limit_bytes_per_sec, time_source_, static_cast<double>(limit_bytes_per_sec));
+  }
+
+  if (has_upload_limit_) {
+    uint64_t limit_bytes_per_sec = kiloBytesToBytes(upload_limit_kbps_);
+    upload_token_bucket_ = std::make_shared<SharedTokenBucketImpl>(
+        limit_bytes_per_sec, time_source_, static_cast<double>(limit_bytes_per_sec));
+  }
+}
+
+TcpBandwidthLimitStats FilterConfig::generateStats(const std::string& prefix, Stats::Scope& scope) {
+  const std::string final_prefix = prefix + ".tcp_bandwidth_limit";
+  return {ALL_TCP_BANDWIDTH_LIMIT_STATS(POOL_COUNTER_PREFIX(scope, final_prefix))};
+}
+
+TcpBandwidthLimitFilter::TcpBandwidthLimitFilter(FilterConfigSharedPtr config,
+                                                 Event::Dispatcher& dispatcher)
+    : config_(config), dispatcher_(dispatcher) {}
+
+Network::FilterStatus TcpBandwidthLimitFilter::onData(Buffer::Instance& data, bool) {
+  if (!config_->enabled() || !config_->hasDownloadLimit()) {
+    return Network::FilterStatus::Continue;
+  }
+
+  config_->stats().download_enabled_.inc();
+
+  uint64_t data_size = data.length();
+  uint64_t consumed = config_->downloadTokenBucket()->consume(data_size, true);
+
+  if (consumed < data_size) {
+    config_->stats().download_throttled_.inc();
+
+    if (consumed > 0) {
+      Buffer::OwnedImpl passthrough;
+      passthrough.move(data, consumed);
+      read_callbacks_->injectReadDataToFilterChain(passthrough, false);
+    }
+
+    download_buffer_.move(data);
+
+    if (!read_disabled_) {
+      read_callbacks_->connection().readDisable(true);
+      read_disabled_ = true;
+    }
+
+    if (!download_timer_) {
+      download_timer_ = dispatcher_.createTimer([this]() { onDownloadTokenTimer(); });
+      download_timer_->enableTimer(config_->fillInterval());
+    }
+
+    return Network::FilterStatus::StopIteration;
+  }
+
+  return Network::FilterStatus::Continue;
+}
+
+Network::FilterStatus TcpBandwidthLimitFilter::onWrite(Buffer::Instance& data, bool) {
+  if (!config_->enabled() || !config_->hasUploadLimit()) {
+    return Network::FilterStatus::Continue;
+  }
+
+  config_->stats().upload_enabled_.inc();
+
+  uint64_t data_size = data.length();
+  uint64_t consumed = config_->uploadTokenBucket()->consume(data_size, true);
+
+  if (consumed < data_size) {
+    config_->stats().upload_throttled_.inc();
+
+    if (consumed > 0) {
+      Buffer::OwnedImpl to_send;
+      to_send.move(data, consumed);
+      write_callbacks_->connection().write(to_send, false);
+    }
+
+    upload_buffer_.move(data);
+
+    if (!upload_timer_) {
+      upload_timer_ = dispatcher_.createTimer([this]() { onUploadTokenTimer(); });
+      upload_timer_->enableTimer(config_->fillInterval());
+    }
+
+    return Network::FilterStatus::StopIteration;
+  }
+
+  return Network::FilterStatus::Continue;
+}
+
+void TcpBandwidthLimitFilter::onDownloadTokenTimer() {
+  processBufferedDownloadData();
+
+  if (download_buffer_.length() > 0) {
+    download_timer_->enableTimer(config_->fillInterval());
+  } else {
+    download_timer_.reset();
+
+    if (read_disabled_) {
+      read_callbacks_->connection().readDisable(false);
+      read_disabled_ = false;
+    }
+  }
+}
+
+void TcpBandwidthLimitFilter::onUploadTokenTimer() {
+  processBufferedUploadData();
+
+  if (upload_buffer_.length() > 0) {
+    upload_timer_->enableTimer(config_->fillInterval());
+  } else {
+    upload_timer_.reset();
+  }
+}
+
+void TcpBandwidthLimitFilter::processBufferedDownloadData() {
+  if (download_buffer_.length() == 0 || !config_->downloadTokenBucket()) {
+    return;
+  }
+
+  uint64_t buffer_size = download_buffer_.length();
+  uint64_t consumed = config_->downloadTokenBucket()->consume(buffer_size, true);
+
+  if (consumed > 0) {
+    Buffer::OwnedImpl data_to_send;
+    data_to_send.move(download_buffer_, consumed);
+    read_callbacks_->injectReadDataToFilterChain(data_to_send, false);
+  }
+}
+
+void TcpBandwidthLimitFilter::processBufferedUploadData() {
+  if (upload_buffer_.length() == 0 || !config_->uploadTokenBucket()) {
+    return;
+  }
+
+  uint64_t buffer_size = upload_buffer_.length();
+  uint64_t consumed = config_->uploadTokenBucket()->consume(buffer_size, true);
+
+  if (consumed > 0) {
+    Buffer::OwnedImpl data_to_send;
+    data_to_send.move(upload_buffer_, consumed);
+    write_callbacks_->connection().write(data_to_send, false);
+  }
+}
+
+} // namespace TcpBandwidthLimit
+} // namespace NetworkFilters
+} // namespace Extensions
+} // namespace Envoy