Support process level access log rate limiting

CODEOWNERS

@@ -274,6 +274,8 @@ extensions/upstreams/tcp @ggreenway @mattklein123
 /*/extensions/compression/zstd @rainingmaster @mattklein123
 # cel
 /*/extensions/access_loggers/filters/cel @kyessenov @douglas-reid @adisuissa
+# process rate limit
+/*/extensions/access_loggers/filters/process_ratelimit @taoxuy @kyessenov
 # health check
 /*/extensions/filters/http/health_check @mattklein123 @adisuissa
 # lua

api/BUILD

@@ -132,6 +132,7 @@ proto_library(
         "//envoy/data/tap/v3:pkg",
         "//envoy/extensions/access_loggers/file/v3:pkg",
         "//envoy/extensions/access_loggers/filters/cel/v3:pkg",
+        "//envoy/extensions/access_loggers/filters/process_ratelimit/v3:pkg",
         "//envoy/extensions/access_loggers/fluentd/v3:pkg",
         "//envoy/extensions/access_loggers/grpc/v3:pkg",
         "//envoy/extensions/access_loggers/open_telemetry/v3:pkg",

api/envoy/extensions/access_loggers/filters/process_ratelimit/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/access_loggers/filters/process_ratelimit/v3/process_ratelimit.proto

@@ -0,0 +1,32 @@
+syntax = "proto3";
+
+package envoy.extensions.access_loggers.filters.process_ratelimit.v3;
+
+import "envoy/config/core/v3/config_source.proto";
+
+import "udpa/annotations/status.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.access_loggers.filters.process_ratelimit.v3";
+option java_outer_classname = "ProcessRatelimitProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/access_loggers/filters/process_ratelimit/v3;process_ratelimitv3";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+
+// [#protodoc-title: ProcessRateLimiter]
+// [#extension: envoy.access_loggers.extension_filters.process_ratelimit]
+
+// Filters for rate limiting the access log emission using global token buckets per process and shared across all listeners.
+message ProcessRateLimitFilter {
+  // The dynamic config for the token bucket.
+  DynamicTokenBucket dynamic_config = 1;
+}
+
+message DynamicTokenBucket {
+  // the key used to find the token bucket in the singleton map.
+  string resource_name = 1 [(validate.rules).string = {min_len: 1}];
+
+  // The configuration source for the :ref:`token_bucket <envoy_v3_api_msg_type.v3.TokenBucket>`.
+  // It should stay the same through the process lifetime.
+  config.core.v3.ConfigSource config_source = 2 [(validate.rules).message = {required: true}];
+}

api/versioning/BUILD

@@ -71,6 +71,7 @@ proto_library(
         "//envoy/data/tap/v3:pkg",
         "//envoy/extensions/access_loggers/file/v3:pkg",
         "//envoy/extensions/access_loggers/filters/cel/v3:pkg",
+        "//envoy/extensions/access_loggers/filters/process_ratelimit/v3:pkg",
         "//envoy/extensions/access_loggers/fluentd/v3:pkg",
         "//envoy/extensions/access_loggers/grpc/v3:pkg",
         "//envoy/extensions/access_loggers/open_telemetry/v3:pkg",

changelogs/current.yaml

@@ -90,6 +90,10 @@ removed_config_or_runtime:
     Removed runtime guard ``envoy.reloadable_features.report_load_with_rq_issued`` and legacy code paths.
 
 new_features:
+- area: access_log
+  change: |
+    Support process-level rate limiting on access log emission by
+    :ref:`ProcessRateLimitFilter <envoy_v3_api_msg_extensions.access_loggers.filters.process_ratelimit.v3.ProcessRateLimitFilter>`.
 - area: udp_sink
   change: |
     Enhance UDP sink to handle the tapped message which the size is bigger than 64K.

source/extensions/access_loggers/filters/process_ratelimit/BUILD

@@ -0,0 +1,54 @@
+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 = "provider_singleton_lib",
+    srcs = ["provider_singleton.cc"],
+    hdrs = ["provider_singleton.h"],
+    deps = [
+        "//envoy/event:dispatcher_interface",
+        "//envoy/event:timer_interface",
+        "//envoy/ratelimit:ratelimit_interface",
+        "//source/common/common:thread_synchronizer_lib",
+        "//source/common/common:token_bucket_impl_lib",
+        "//source/common/config:subscription_base_interface",
+        "//source/common/grpc:common_lib",
+        "//source/common/init:target_lib",
+        "//source/extensions/filters/common/local_ratelimit:local_ratelimit_lib",
+        "@envoy_api//envoy/type/v3:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_extension(
+    name = "filter_lib",
+    srcs = ["filter.cc"],
+    hdrs = ["filter.h"],
+    deps = [
+        ":provider_singleton_lib",
+        "//envoy/access_log:access_log_interface",
+        "//source/common/init:target_lib",
+        "@envoy_api//envoy/extensions/access_loggers/filters/process_ratelimit/v3:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_extension(
+    name = "config",
+    srcs = ["config.cc"],
+    hdrs = ["config.h"],
+    deps = [
+        ":filter_lib",
+        "//envoy/access_log:access_log_interface",
+        "//envoy/registry",
+        "//source/common/access_log:access_log_lib",
+        "//source/common/protobuf:utility_lib",
+        "@envoy_api//envoy/extensions/access_loggers/filters/process_ratelimit/v3:pkg_cc_proto",
+    ],
+)

source/extensions/access_loggers/filters/process_ratelimit/config.cc

@@ -0,0 +1,39 @@
+#include "source/extensions/access_loggers/filters/process_ratelimit/config.h"
+
+#include "envoy/extensions/access_loggers/filters/process_ratelimit/v3/process_ratelimit.pb.h"
+
+#include "source/common/protobuf/utility.h"
+#include "source/extensions/access_loggers/filters/process_ratelimit/filter.h"
+#include "source/extensions/filters/common/local_ratelimit/local_ratelimit_impl.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace AccessLoggers {
+namespace Filters {
+namespace ProcessRateLimit {
+
+AccessLog::FilterPtr ProcessRateLimitFilterFactory::createFilter(
+    const envoy::config::accesslog::v3::ExtensionFilter& config,
+    Server::Configuration::GenericFactoryContext& context) {
+  auto factory_config =
+      Config::Utility::translateToFactoryConfig(config, context.messageValidationVisitor(), *this);
+  const auto& process_ratelimit_config =
+      dynamic_cast<const envoy::extensions::access_loggers::filters::process_ratelimit::v3::
+                       ProcessRateLimitFilter&>(*factory_config);
+  auto filter = std::make_unique<ProcessRateLimitFilter>(context.serverFactoryContext(),
+                                                         process_ratelimit_config);
+  return filter;
+}
+
+ProtobufTypes::MessagePtr ProcessRateLimitFilterFactory::createEmptyConfigProto() {
+  return std::make_unique<
+      envoy::extensions::access_loggers::filters::process_ratelimit::v3::ProcessRateLimitFilter>();
+}
+
+REGISTER_FACTORY(ProcessRateLimitFilterFactory, AccessLog::ExtensionFilterFactory);
+
+} // namespace ProcessRateLimit
+} // namespace Filters
+} // namespace AccessLoggers
+} // namespace Extensions
+} // namespace Envoy

source/extensions/access_loggers/filters/process_ratelimit/config.h

@@ -0,0 +1,28 @@
+#pragma once
+
+#include "envoy/access_log/access_log.h"
+#include "envoy/registry/registry.h"
+
+#include "source/common/access_log/access_log_impl.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace AccessLoggers {
+namespace Filters {
+namespace ProcessRateLimit {
+
+class ProcessRateLimitFilterFactory : public AccessLog::ExtensionFilterFactory {
+public:
+  AccessLog::FilterPtr createFilter(const envoy::config::accesslog::v3::ExtensionFilter& config,
+                                    Server::Configuration::GenericFactoryContext& context) override;
+  ProtobufTypes::MessagePtr createEmptyConfigProto() override;
+  std::string name() const override {
+    return "envoy.access_loggers.extension_filters.process_ratelimit";
+  }
+};
+
+} // namespace ProcessRateLimit
+} // namespace Filters
+} // namespace AccessLoggers
+} // namespace Extensions
+} // namespace Envoy

source/extensions/access_loggers/filters/process_ratelimit/filter.cc

@@ -0,0 +1,72 @@
+#include "source/extensions/access_loggers/filters/process_ratelimit/filter.h"
+
+#include "envoy/access_log/access_log.h"
+
+#include "source/common/init/target_impl.h"
+#include "source/extensions/filters/common/local_ratelimit/local_ratelimit_impl.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace AccessLoggers {
+namespace Filters {
+namespace ProcessRateLimit {
+
+ProcessRateLimitFilter::ProcessRateLimitFilter(
+    Server::Configuration::ServerFactoryContext& context,
+    const envoy::extensions::access_loggers::filters::process_ratelimit::v3::ProcessRateLimitFilter&
+        config)
+    : setter_key_(reinterpret_cast<intptr_t>(this)),
+      cancel_cb_(std::make_shared<std::atomic<bool>>(false)), context_(context),
+      stats_({ALL_PROCESS_RATELIMIT_FILTER_STATS(
+          POOL_COUNTER_PREFIX(context.scope(), "access_log.process_ratelimit."))}) {
+  auto setter =
+      [this, cancel_cb = cancel_cb_](
+          Envoy::Extensions::Filters::Common::LocalRateLimit::LocalRateLimiterSharedPtr limiter)
+      -> void {
+    if (!cancel_cb->load()) {
+      ENVOY_BUG(limiter != nullptr, "limiter shouldn't be null if the `limiter` is set from "
+                                    "callback.");
+      rate_limiter_->setLimiter(limiter);
+    }
+  };
+
+  if (!config.has_dynamic_config()) {
+    ExceptionUtil::throwEnvoyException("`dynamic_config` is required.");
+  }
+  rate_limiter_ = Envoy::Extensions::Filters::Common::LocalRateLimit::RateLimiterProviderSingleton::
+      getRateLimiter(context, config.dynamic_config().resource_name(),
+                     config.dynamic_config().config_source(), setter_key_, std::move(setter));
+}
+
+ProcessRateLimitFilter::~ProcessRateLimitFilter() {
+  // The destructor can be called in any thread.
+  // The `cancel_cb_` is set to true to prevent the `limiter` from being set in
+  // the `setter` from the main thread.
+  cancel_cb_->store(true);
+  context_.mainThreadDispatcher().post(
+      [limiter = std::move(rate_limiter_), setter_key = setter_key_] {
+        // remove the setter for this filter.
+        limiter->getSubscription()->removeSetter(setter_key);
+      });
+}
+
+bool ProcessRateLimitFilter::evaluate(const Formatter::Context&,
+                                      const StreamInfo::StreamInfo&) const {
+  ENVOY_BUG(rate_limiter_->getLimiter() != nullptr,
+            "rate_limiter_.limiter_ should be already set in init callback.");
+  Extensions::Filters::Common::LocalRateLimit::LocalRateLimiterSharedPtr limiter =
+      rate_limiter_->getLimiter();
+  auto result = limiter->requestAllowed({});
+  if (!result.allowed) {
+    stats_.denied_.inc();
+  } else {
+    stats_.allowed_.inc();
+  }
+  return result.allowed;
+}
+
+} // namespace ProcessRateLimit
+} // namespace Filters
+} // namespace AccessLoggers
+} // namespace Extensions
+} // namespace Envoy

source/extensions/access_loggers/filters/process_ratelimit/filter.h

@@ -0,0 +1,49 @@
+#pragma once
+
+#include "envoy/access_log/access_log.h"
+#include "envoy/extensions/access_loggers/filters/process_ratelimit/v3/process_ratelimit.pb.h"
+
+#include "source/common/init/target_impl.h"
+#include "source/extensions/access_loggers/filters/process_ratelimit/provider_singleton.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace AccessLoggers {
+namespace Filters {
+namespace ProcessRateLimit {
+
+#define ALL_PROCESS_RATELIMIT_FILTER_STATS(COUNTER)                                                \
+  COUNTER(allowed)                                                                                 \
+  COUNTER(denied)
+
+/**
+ * Struct definition for all process ratelimit filter stats. @see stats_macros.h
+ */
+struct ProcessRateLimitFilterStats {
+  ALL_PROCESS_RATELIMIT_FILTER_STATS(GENERATE_COUNTER_STRUCT)
+};
+class ProcessRateLimitFilter : public AccessLog::Filter {
+public:
+  ProcessRateLimitFilter(Server::Configuration::ServerFactoryContext& context,
+                         const envoy::extensions::access_loggers::filters::process_ratelimit::v3::
+                             ProcessRateLimitFilter& config);
+
+  bool evaluate(const Formatter::Context& log_context,
+                const StreamInfo::StreamInfo& stream_info) const override;
+
+  ~ProcessRateLimitFilter() override;
+
+private:
+  const intptr_t setter_key_;
+  std::shared_ptr<std::atomic<bool>> cancel_cb_;
+  Server::Configuration::ServerFactoryContext& context_;
+  ProcessRateLimitFilterStats stats_;
+  mutable Envoy::Extensions::Filters::Common::LocalRateLimit::RateLimiterProviderSingleton::
+      RateLimiterWrapperPtr rate_limiter_;
+};
+
+} // namespace ProcessRateLimit
+} // namespace Filters
+} // namespace AccessLoggers
+} // namespace Extensions
+} // namespace Envoy