feat: impl apply_on_stream_done for local rate limit

api/envoy/config/route/v3/route_components.proto

@@ -2807,7 +2807,11 @@ message RateLimit {
   // requests based on the action with this field set to true. Users should ensure that the rate limit is enforced by the actions
   // applied on the request path, i.e. the ones with this field set to false.
   //
-  // Currently, this is only supported by the HTTP global rate filter.
+  // Supported by HTTP global rate limiting. HTTP local rate limiting supports this only for the
+  // filter's embedded ``rate_limits``, including per-route ``typed_per_filter_config``, and ignores it
+  // for route and virtual-host ``rate_limits``. Local completion charges matching descriptor buckets
+  // in full, overdrawing them by up to one bucket of tokens when the usage exceeds the remaining
+  // budget; the default bucket is consumed only on requests.
   bool apply_on_stream_done = 6;
 
   // Descriptor level X-RateLimit headers options which may override the filter level setting.

api/envoy/extensions/filters/http/local_ratelimit/v3/local_rate_limit.proto

@@ -180,6 +180,11 @@ message LocalRateLimit {
   //   2. :ref:`dynamic metadata <envoy_v3_api_field_config.route.v3.RateLimit.Action.dynamic_metadata>`.
   //   3. :ref:`disable_key <envoy_v3_api_field_config.route.v3.RateLimit.disable_key>`.
   //   4. :ref:`override limit <envoy_v3_api_field_config.route.v3.RateLimit.limit>`.
+  //
+  //   :ref:`apply_on_stream_done <envoy_v3_api_field_config.route.v3.RateLimit.apply_on_stream_done>`
+  //   updates descriptor buckets from response context at completion without enforcing the current
+  //   request, in both the filter-level and per-route configuration. Use request-path actions for the
+  //   same descriptor to enforce subsequent requests.
   repeated config.route.v3.RateLimit rate_limits = 17;
 
   // Specifies the max dynamic descriptors kept in the cache for a particular wildcard descriptor

changelogs/current/new_features/local_ratelimit__apply-on-stream-done.rst

@@ -0,0 +1,6 @@
+Added HTTP local rate limit completion accounting for embedded ``rate_limits`` with
+:ref:`apply_on_stream_done <envoy_v3_api_field_config.route.v3.RateLimit.apply_on_stream_done>`,
+in both the filter-level and per-route configuration. These actions update matching descriptor
+token buckets when the stream completes and remain skipped on the request path. The default local
+token bucket continues to apply only on the request path. Route and virtual-host rate limit entries
+are unchanged and continue to apply on the request path.

docs/root/configuration/http/http_filters/local_rate_limit_filter.rst

@@ -151,6 +151,67 @@ path, then 10 req/min are allowed. But if they come from a downstream service
 cluster "foo" for "/foo/bar2" path, then 100 req/min are allowed. Otherwise,
 1000 req/min are allowed.
 
+Accounting for usage when a stream completes
+--------------------------------------------
+
+Set :ref:`apply_on_stream_done <envoy_v3_api_field_config.route.v3.RateLimit.apply_on_stream_done>`
+on an embedded ``rate_limits`` action to update local descriptor token buckets when the stream completes.
+This allows :ref:`hits_addend <envoy_v3_api_field_config.route.v3.RateLimit.hits_addend>` in the
+filter's ``rate_limits`` to use response headers or final stream information, such as
+``%RESP(x-usage)%`` or ``%BYTES_SENT%``.
+
+This works in both the filter-level configuration and the per-route configuration in
+:ref:`typed_per_filter_config <envoy_v3_api_field_config.route.v3.Route.typed_per_filter_config>`.
+The local rate limit filter ignores ``apply_on_stream_done`` on route and virtual-host
+``rate_limits``; those entries continue to apply on the request path.
+
+Example using embedded ``rate_limits``:
+
+.. code-block:: yaml
+
+  rate_limits:
+  - actions:
+    - generic_key:
+        descriptor_key: usage
+        descriptor_value: service
+  - actions:
+    - generic_key:
+        descriptor_key: usage
+        descriptor_value: service
+    apply_on_stream_done: true
+    hits_addend:
+      format: "%RESP(x-usage)%"
+
+The first action enforces the budget and consumes one token on the request path. The second
+charges additional usage reported in the ``x-usage`` response header when the stream completes.
+Both actions must match a configured local descriptor. Completion actions update every matching
+descriptor bucket without blocking the completed request. Usage exceeding the remaining budget
+overdraws the bucket: it borrows future tokens, and later requests are rejected until the refill
+repays the debt. The debt is capped at the bucket's ``max_tokens``, and usage beyond the cap is
+dropped. The default token bucket is consumed only on the request path.
+Missing, empty or invalid formatted usage does not generate a descriptor and logs a warning.
+Negative hits addends refund tokens using the existing local descriptor behavior.
+
+With ``local_cluster_rate_limit``, positive charges are divided by the local cluster's token share
+factor, while refunds add the specified token count without scaling. This matches request-path
+refund behavior. A charge and refund with the same addend do not cancel when the share factor is
+less than one.
+
+Request and completion actions share the same bounded cache for wildcard descriptors. A value
+first seen at completion creates a bucket and can evict another bucket, including one used for
+request enforcement. Revisiting an evicted value creates a fresh bucket with its initial budget.
+Size ``max_dynamic_descriptors`` for the combined cardinality of request and completion values.
+
+Completion accounting works with process-wide and per-connection buckets. Request counters and
+rate limit response headers reflect the request-path decision; completion accounting does not
+change the response or increment request counters again.
+
+As with the global rate limit filter, completion actions also run when the request was rejected
+by the rate limit filter. Stream information such as ``%BYTES_SENT%`` includes usage from a local
+rate limit response. To account for upstream-reported usage, use a response header such as
+``%RESP(x-usage)%`` that is present only when the upstream reports a cost. A missing header
+drops the descriptor and logs a warning.
+
 Statistics
 ----------
 

source/extensions/filters/common/local_ratelimit/local_ratelimit_impl.cc

@@ -88,7 +88,7 @@ RateLimitTokenBucket::RateLimitTokenBucket(uint64_t max_tokens, uint64_t tokens_
       fill_interval_(fill_interval), shadow_mode_(shadow_mode),
       preserve_one_request_(Runtime::runtimeFeatureEnabled(
           "envoy.reloadable_features.local_ratelimit_local_cluster_preserve_one_request")) {}
-bool RateLimitTokenBucket::consume(double factor, uint64_t to_consume) {
+bool RateLimitTokenBucket::consume(double factor, uint64_t to_consume, bool overdraft) {
   ASSERT(!(factor <= 0.0 || factor > 1.0));
   const double max_tokens = token_bucket_.maxTokens();
   const double tokens = to_consume / factor;
@@ -99,7 +99,14 @@ bool RateLimitTokenBucket::consume(double factor, uint64_t to_consume) {
   const double threshold = preserve_one_request_ && max_tokens > 0 && to_consume <= max_tokens
                                ? std::min(tokens, max_tokens)
                                : tokens;
-  auto cb = [tokens, threshold](double total) { return total < threshold ? 0.0 : tokens; };
+  auto cb = [tokens, threshold, overdraft, max_tokens](double total) {
+    if (overdraft) {
+      // Borrow future tokens when the bucket is short, but by at most max_tokens, so a single
+      // large cost can't block the bucket indefinitely.
+      return std::max(0.0, std::min(tokens, total + max_tokens));
+    }
+    return total >= threshold ? tokens : 0.0;
+  };
   return token_bucket_.consume(cb) != 0.0;
 }
 
@@ -193,6 +200,29 @@ LocalRateLimiterImpl::LocalRateLimiterImpl(
 
 LocalRateLimiterImpl::~LocalRateLimiterImpl() = default;
 
+RateLimitTokenBucketSharedPtr
+LocalRateLimiterImpl::getBucket(const RateLimit::Descriptor& descriptor) {
+  auto iter = descriptors_.find(descriptor);
+  return iter != descriptors_.end() ? iter->second : dynamic_descriptors_.getBucket(descriptor);
+}
+
+void LocalRateLimiterImpl::consumeDescriptors(
+    absl::Span<const RateLimit::Descriptor> request_descriptors) {
+  const double share_factor =
+      share_provider_ != nullptr ? share_provider_->getTokensShareFactor() : 1.0;
+  for (const auto& descriptor : request_descriptors) {
+    auto bucket = getBucket(descriptor);
+    if (bucket == nullptr) {
+      continue;
+    }
+    if (descriptor.is_negative_hits_ && descriptor.hits_addend_.has_value()) {
+      bucket->refill(descriptor.hits_addend_.value());
+    } else {
+      bucket->consume(share_factor, descriptor.hits_addend_.value_or(1), /*overdraft=*/true);
+    }
+  }
+}
+
 struct MatchResult {
   RateLimitTokenBucketSharedPtr token_bucket;
   std::reference_wrapper<const RateLimit::Descriptor> request_descriptor;
@@ -207,14 +237,9 @@ LocalRateLimiterImpl::requestAllowed(absl::Span<const RateLimit::Descriptor> req
 
   // Find all matched descriptors.
   for (const auto& request_descriptor : request_descriptors) {
-    auto iter = descriptors_.find(request_descriptor);
-    if (iter != descriptors_.end()) {
-      matched_results.push_back(MatchResult{iter->second, request_descriptor});
-    } else {
-      auto token_bucket = dynamic_descriptors_.getBucket(request_descriptor);
-      if (token_bucket != nullptr) {
-        matched_results.push_back(MatchResult{token_bucket, request_descriptor});
-      }
+    auto token_bucket = getBucket(request_descriptor);
+    if (token_bucket != nullptr) {
+      matched_results.push_back(MatchResult{std::move(token_bucket), request_descriptor});
     }
   }
 
@@ -345,9 +370,9 @@ void DynamicDescriptorMap::addDescriptor(const RateLimit::LocalDescriptor& confi
 }
 
 RateLimitTokenBucketSharedPtr
-DynamicDescriptorMap::getBucket(const RateLimit::Descriptor request_descriptor) {
+DynamicDescriptorMap::getBucket(const RateLimit::Descriptor& request_descriptor) {
   for (const auto& pair : config_descriptors_) {
-    auto config_descriptor = pair.first;
+    const auto& config_descriptor = pair.first;
     if (!matchDescriptorEntries(request_descriptor.entries_, config_descriptor.entries_)) {
       continue;
     }

source/extensions/filters/common/local_ratelimit/local_ratelimit_impl.h

@@ -59,7 +59,7 @@ class DynamicDescriptorMap : public Logger::Loggable<Logger::Id::rate_limit_quot
   void addDescriptor(const RateLimit::LocalDescriptor& descriptor,
                      DynamicDescriptorSharedPtr dynamic_descriptor);
   // pass request_descriptors to the dynamic descriptor set to get the token bucket.
-  RateLimitTokenBucketSharedPtr getBucket(const RateLimit::Descriptor);
+  RateLimitTokenBucketSharedPtr getBucket(const RateLimit::Descriptor&);
 
 private:
   bool matchDescriptorEntries(const std::vector<RateLimit::DescriptorEntry>& request_entries,
@@ -111,7 +111,10 @@ class RateLimitTokenBucket : public TokenBucketContext,
                        bool shadow_mode);
 
   // RateLimitTokenBucket
-  bool consume(double factor = 1.0, uint64_t tokens = 1);
+  // With overdraft, the cost is consumed even when the bucket is short, and the bucket can go
+  // below zero by at most max_tokens. Later requests wait until the refill repays the debt.
+  // Returns false only when the bucket is already at the debt cap.
+  bool consume(double factor = 1.0, uint64_t tokens = 1, bool overdraft = false);
   // Refill tokens back to the bucket, capped at max_tokens.
   void refill(uint64_t tokens);
   double fillRate() const { return token_bucket_.fillRate(); }
@@ -151,7 +154,12 @@ class LocalRateLimiterImpl : public Logger::Loggable<Logger::Id::local_rate_limi
   LocalRateLimiter::Result
   requestAllowed(absl::Span<const RateLimit::Descriptor> request_descriptors) override;
 
+  // Account for completed descriptor usage without enforcement. Usage beyond the remaining budget
+  // overdraws the bucket. The default bucket is consumed only on the request path.
+  void consumeDescriptors(absl::Span<const RateLimit::Descriptor> request_descriptors);
+
 private:
+  RateLimitTokenBucketSharedPtr getBucket(const RateLimit::Descriptor& descriptor);
   RateLimitTokenBucketSharedPtr default_token_bucket_;
 
   TimeSource& time_source_;

source/extensions/filters/http/local_ratelimit/local_ratelimit.cc

@@ -60,6 +60,9 @@ FilterConfig::FilterConfig(
               : std::nullopt),
       stage_(static_cast<uint64_t>(config.stage())),
       has_descriptors_(!config.descriptors().empty()),
+      has_stream_done_actions_(
+          std::any_of(config.rate_limits().begin(), config.rate_limits().end(),
+                      [](const auto& rate_limit) { return rate_limit.apply_on_stream_done(); })),
       enable_x_rate_limit_headers_(config.enable_x_ratelimit_headers() ==
                                    envoy::extensions::common::ratelimit::v3::DRAFT_VERSION_03),
       enable_retry_after_header_(config.enable_retry_after_header() &&
@@ -164,6 +167,9 @@ bool FilterConfig::enforced() const {
   return filter_enforced_.has_value() ? filter_enforced_->enabled() : false;
 }
 
+Filter::Filter(FilterConfigSharedPtr config)
+    : config_(std::move(config)), used_config_(config_.get()) {}
+
 Http::FilterHeadersStatus Filter::decodeHeaders(Http::RequestHeaderMap& headers, bool) {
   const auto* route_config =
       Http::Utility::resolveMostSpecificPerFilterConfig<FilterConfig>(decoder_callbacks_);
@@ -176,7 +182,8 @@ Http::FilterHeadersStatus Filter::decodeHeaders(Http::RequestHeaderMap& headers,
     used_config_ = route_config; // Overwrite the used configuration.
   }
 
-  if (!used_config_->enabled()) {
+  rate_limit_enabled_ = used_config_->enabled();
+  if (!rate_limit_enabled_) {
     return Http::FilterHeadersStatus::Continue;
   }
 
@@ -185,7 +192,9 @@ Http::FilterHeadersStatus Filter::decodeHeaders(Http::RequestHeaderMap& headers,
   std::vector<RateLimit::Descriptor> descriptors;
   if (used_config_->hasDescriptors()) {
     if (used_config_->hasRateLimitConfigs()) {
-      used_config_->populateDescriptors(headers, decoder_callbacks_->streamInfo(), descriptors);
+      used_config_->populateDescriptors(headers, /*response_headers=*/nullptr,
+                                        decoder_callbacks_->streamInfo(), descriptors,
+                                        /*on_stream_done=*/false);
     } else {
       populateDescriptors(descriptors, headers);
     }
@@ -265,6 +274,32 @@ Http::FilterHeadersStatus Filter::encodeHeaders(Http::ResponseHeaderMap& headers
   return Http::FilterHeadersStatus::Continue;
 }
 
+void Filter::onDestroy() {
+  // Completion accounting needs local descriptors and embedded apply_on_stream_done rate limits.
+  if (!rate_limit_enabled_ || !used_config_->hasDescriptors() ||
+      !used_config_->hasStreamDoneActions()) {
+    return;
+  }
+  // Read the headers from the stream rather than keeping references to them.
+  const auto request_headers = decoder_callbacks_->requestHeaders();
+  if (!request_headers.has_value()) {
+    return;
+  }
+  const auto response_headers = decoder_callbacks_->responseHeaders();
+  std::vector<RateLimit::Descriptor> descriptors;
+  used_config_->populateDescriptors(*request_headers, response_headers.ptr(),
+                                    decoder_callbacks_->streamInfo(), descriptors,
+                                    /*on_stream_done=*/true);
+  if (descriptors.empty()) {
+    return;
+  }
+  if (used_config_->rateLimitPerConnection()) {
+    getPerConnectionRateLimiter().consumeDescriptors(descriptors);
+  } else {
+    used_config_->consumeDescriptors(descriptors);
+  }
+}
+
 Filters::Common::LocalRateLimit::LocalRateLimiterImpl::Result
 Filter::requestAllowed(absl::Span<const RateLimit::Descriptor> request_descriptors) {
   return used_config_->rateLimitPerConnection()

source/extensions/filters/http/local_ratelimit/local_ratelimit.h

@@ -94,6 +94,9 @@ class FilterConfig : public Router::RouteSpecificFilterConfig,
   Runtime::Loader& runtime() { return runtime_; }
   Filters::Common::LocalRateLimit::LocalRateLimiter::Result
   requestAllowed(absl::Span<const RateLimit::Descriptor> request_descriptors) const;
+  void consumeDescriptors(absl::Span<const RateLimit::Descriptor> descriptors) const {
+    rate_limiter_->consumeDescriptors(descriptors);
+  }
   bool enabled() const;
   bool enforced() const;
   LocalRateLimitStats& stats() const { return stats_; }
@@ -102,6 +105,7 @@ class FilterConfig : public Router::RouteSpecificFilterConfig,
   Http::Code status() const { return status_; }
   uint64_t stage() const { return stage_; }
   bool hasDescriptors() const { return has_descriptors_; }
+  bool hasStreamDoneActions() const { return has_stream_done_actions_; }
   const std::chrono::milliseconds& fillInterval() const { return fill_interval_; }
   uint32_t maxTokens() const { return max_tokens_; }
   uint32_t tokensPerFill() const { return tokens_per_fill_; }
@@ -128,13 +132,13 @@ class FilterConfig : public Router::RouteSpecificFilterConfig,
   }
 
   void populateDescriptors(const Http::RequestHeaderMap& headers,
+                           const Http::ResponseHeaderMap* response_headers,
                            const StreamInfo::StreamInfo& info,
-                           Filters::Common::RateLimit::RateLimitDescriptors& descriptors) const {
+                           Filters::Common::RateLimit::RateLimitDescriptors& descriptors,
+                           bool on_stream_done) const {
     ASSERT(rate_limit_config_ != nullptr);
-    // Local rate limiting runs on the decode path only; no response headers are available to the
-    // formatter (and %RESP()% in hits_addend.format resolves empty, as before).
-    rate_limit_config_->populateDescriptors(headers, /*response_headers=*/nullptr, info,
-                                            local_info_.clusterName(), descriptors);
+    rate_limit_config_->populateDescriptors(headers, response_headers, info,
+                                            local_info_.clusterName(), descriptors, on_stream_done);
   }
 
 private:
@@ -172,6 +176,7 @@ class FilterConfig : public Router::RouteSpecificFilterConfig,
   Router::HeaderParserPtr request_headers_parser_;
   const uint64_t stage_;
   const bool has_descriptors_;
+  const bool has_stream_done_actions_;
   const bool enable_x_rate_limit_headers_;
   const bool enable_retry_after_header_ = false;
   const envoy::extensions::common::ratelimit::v3::VhRateLimitsOptions vh_rate_limits_;
@@ -187,7 +192,10 @@ using FilterConfigSharedPtr = std::shared_ptr<FilterConfig>;
  */
 class Filter : public Http::PassThroughFilter, Logger::Loggable<Logger::Id::filter> {
 public:
-  Filter(FilterConfigSharedPtr config) : config_(config), used_config_(config_.get()) {}
+  Filter(FilterConfigSharedPtr config);
+
+  // Http::StreamFilterBase
+  void onDestroy() override;
 
   // Http::StreamDecoderFilter
   Http::FilterHeadersStatus decodeHeaders(Http::RequestHeaderMap& headers,
@@ -224,6 +232,8 @@ class Filter : public Http::PassThroughFilter, Logger::Loggable<Logger::Id::filt
   const FilterConfig* used_config_{};
   std::shared_ptr<const Filters::Common::LocalRateLimit::TokenBucketContext> token_bucket_context_;
   RateLimit::XRateLimitOption x_ratelimit_option_{};
+  // Whether decodeHeaders() ran with the filter enabled. Completion accounting respects it.
+  bool rate_limit_enabled_{false};
 
   VhRateLimitOptions vh_rate_limits_;
 };

test/extensions/filters/common/local_ratelimit/BUILD

@@ -13,11 +13,13 @@ envoy_cc_test(
     srcs = ["local_ratelimit_test.cc"],
     rbe_pool = "linux_x64_small",
     deps = [
+        "//envoy/common:time_interface",
         "//source/common/singleton:manager_impl_lib",
         "//source/extensions/filters/common/local_ratelimit:local_ratelimit_lib",
         "//test/mocks/event:event_mocks",
         "//test/mocks/upstream:cluster_manager_mocks",
         "//test/mocks/upstream:cluster_priority_set_mocks",
+        "//test/test_common:simulated_time_system_lib",
         "//test/test_common:test_runtime_lib",
         "//test/test_common:utility_lib",
     ],