rate-limit: make 429 response mapping configurable

api/envoy/config/filter/http/rate_limit/v2/rate_limit.proto

@@ -40,4 +40,8 @@ message RateLimit {
   // communication failure between rate limiting service and the proxy.
   // Defaults to false.
   bool failure_mode_deny = 5;
+
+  // Specifies whether a `RESOURCE_EXHAUSTED` code must be returned instead of
+  // the default `UNAVAILABLE` code for a rate limited gRPC call.
+  bool rate_limited_as_resource_exhausted = 6;
 }

docs/root/intro/version_history.rst

@@ -17,6 +17,9 @@ Version history
   See `#3611 <https://github.com/envoyproxy/envoy/issues/3611>`_ for details.
 * network: removed the reference to `FilterState` in `Connection` in favor of `StreamInfo`.
 * logging: added missing [ in log prefix.
+* rate-limit: added :ref:`configuration <envoy_api_field_config.filter.http.rate_limit.v2.RateLimit.rate_limited_as_resource_exhausted>`
+  to specify whether the `GrpcStatus` status returned should be `RESOURCE_EXHAUSTED` or
+  `UNAVAILABLE` when a gRPC call is rate limited.
 * rbac: added support for permission matching by :ref:`requested server name <envoy_api_field_config.rbac.v2alpha.Permission.requested_server_name>`.
 * router: added ability to configure arbitrary :ref:`retriable status codes. <envoy_api_field_route.RouteAction.RetryPolicy.retriable_status_codes>`
 * router: added ability to set attempt count in upstream requests, see :ref:`virtual host's include request

include/envoy/http/filter.h

@@ -221,9 +221,18 @@ class StreamDecoderFilterCallbacks : public virtual StreamFilterCallbacks {
    *                  type, or encoded in the grpc-message header.
    * @param modify_headers supplies an optional callback function that can modify the
    *                       response headers.
+   * @param rate_limited_as_resource_exhausted specifies whether a RESOURCE_EXHAUSTED code
+   *                                           should be returned instead of the default
+   *                                           UNAVAILABLE code for rate limited gRPC calls.
    */
   virtual void sendLocalReply(Code response_code, const std::string& body_text,
-                              std::function<void(HeaderMap& headers)> modify_headers) PURE;
+                              std::function<void(HeaderMap& headers)> modify_headers,
+                              bool rate_limited_as_resource_exhausted) PURE;
+
+  void sendLocalReply(Code response_code, const std::string& body_text,
+                      std::function<void(HeaderMap& headers)> modify_headers) {
+    sendLocalReply(response_code, body_text, modify_headers, false);
+  }
 
   /**
    * Called with 100-Continue headers to be encoded.

source/common/grpc/status.cc

@@ -3,9 +3,11 @@
 namespace Envoy {
 namespace Grpc {
 
-Status::GrpcStatus Utility::httpToGrpcStatus(uint64_t http_response_status) {
-  // From
-  // https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md.
+Status::GrpcStatus Utility::httpToGrpcStatus(uint64_t http_response_status,
+                                             bool rate_limited_as_resource_exhausted) {
+  // See:
+  // * https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md
+  // * https://cloud.google.com/apis/design/errors#generating_errors
   switch (http_response_status) {
   case 400:
     return Status::GrpcStatus::Internal;
@@ -16,6 +18,10 @@ Status::GrpcStatus Utility::httpToGrpcStatus(uint64_t http_response_status) {
   case 404:
     return Status::GrpcStatus::Unimplemented;
   case 429:
+    if (rate_limited_as_resource_exhausted) {
+      return Status::GrpcStatus::ResourceExhausted;
+    }
+    return Status::GrpcStatus::Unavailable;
   case 502:
   case 503:
   case 504:

source/common/grpc/status.h

@@ -13,13 +13,23 @@ namespace Grpc {
 class Utility {
 public:
   /**
-   * Returns the gRPC status code from a given HTTP response status code. Ordinarily, it is expected
-   * that a 200 response is provided, but gRPC defines a mapping for intermediaries that are not
-   * gRPC aware, see https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md.
+   * Returns the gRPC status code from a given HTTP response status code.
+   * Ordinarily, it is expected that a 200 response is provided, but gRPC
+   * defines a mapping for intermediaries that are not gRPC aware,
+   * see https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md.
+   *
+   * Google defines a mapping where a code of 429 (rate limited) is mapped to
+   * RESOURCE_EXHAUSTED instead of UNAVAILABLE as defined by gRPC. This function
+   * allows the user to specify the GrpcStatus that should map to a 429 response.
+   * See https://cloud.google.com/apis/design/errors#generating_errors.
+   *
    * @param http_response_status HTTP status code.
+   * @param rate_limited_as_resource_exhausted whether a 429 response code
+   * should be mapped to RESOURCE_EXHAUSTED instead of UNAVAILABLE.
    * @return Status::GrpcStatus corresponding gRPC status code.
    */
-  static Status::GrpcStatus httpToGrpcStatus(uint64_t http_response_status);
+  static Status::GrpcStatus httpToGrpcStatus(uint64_t http_response_status,
+                                             bool rate_limited_as_resource_exhausted);
 
   /**
    * @param grpc_status gRPC status from grpc-status header.

source/common/http/async_client_impl.h

@@ -286,7 +286,8 @@ class AsyncStreamImpl : public AsyncClient::Stream,
   void addDecodedData(Buffer::Instance&, bool) override { NOT_IMPLEMENTED_GCOVR_EXCL_LINE; }
   const Buffer::Instance* decodingBuffer() override { return buffered_body_.get(); }
   void sendLocalReply(Code code, const std::string& body,
-                      std::function<void(HeaderMap& headers)> modify_headers) override {
+                      std::function<void(HeaderMap& headers)> modify_headers,
+                      bool rate_limited_as_resource_exhausted = false) override {
     Utility::sendLocalReply(
         is_grpc_request_,
         [this, modify_headers](HeaderMapPtr&& headers, bool end_stream) -> void {
@@ -296,7 +297,7 @@ class AsyncStreamImpl : public AsyncClient::Stream,
           encodeHeaders(std::move(headers), end_stream);
         },
         [this](Buffer::Instance& data, bool end_stream) -> void { encodeData(data, end_stream); },
-        remote_closed_, code, body, is_head_request_);
+        remote_closed_, code, body, is_head_request_, rate_limited_as_resource_exhausted);
   }
   // The async client won't pause if sending an Expect: 100-Continue so simply
   // swallows any incoming encode100Continue.

source/common/http/conn_manager_impl.cc

@@ -929,7 +929,8 @@ void ConnectionManagerImpl::ActiveStream::refreshCachedRoute() {
 
 void ConnectionManagerImpl::ActiveStream::sendLocalReply(
     bool is_grpc_request, Code code, const std::string& body,
-    std::function<void(HeaderMap& headers)> modify_headers, bool is_head_request) {
+    std::function<void(HeaderMap& headers)> modify_headers, bool is_head_request,
+    bool rate_limited_as_resource_exhausted) {
   Utility::sendLocalReply(is_grpc_request,
                           [this, modify_headers](HeaderMapPtr&& headers, bool end_stream) -> void {
                             if (modify_headers != nullptr) {
@@ -945,7 +946,8 @@ void ConnectionManagerImpl::ActiveStream::sendLocalReply(
                             // request instead.
                             encodeData(nullptr, data, end_stream);
                           },
-                          state_.destroyed_, code, body, is_head_request);
+                          state_.destroyed_, code, body, is_head_request,
+                          rate_limited_as_resource_exhausted);
 }
 
 void ConnectionManagerImpl::ActiveStream::encode100ContinueHeaders(
@@ -1623,7 +1625,7 @@ void ConnectionManagerImpl::ActiveStreamEncoderFilter::responseDataTooLarge() {
             parent_.state_.local_complete_ = end_stream;
           },
           parent_.state_.destroyed_, Http::Code::InternalServerError,
-          CodeUtility::toString(Http::Code::InternalServerError), parent_.is_head_request_);
+          CodeUtility::toString(Http::Code::InternalServerError), parent_.is_head_request_, false);
       parent_.maybeEndEncode(parent_.state_.local_complete_);
     } else {
       resetStream();

source/common/http/conn_manager_impl.h

@@ -173,9 +173,10 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
       return parent_.buffered_request_data_.get();
     }
     void sendLocalReply(Code code, const std::string& body,
-                        std::function<void(HeaderMap& headers)> modify_headers) override {
-      parent_.sendLocalReply(is_grpc_request_, code, body, modify_headers,
-                             parent_.is_head_request_);
+                        std::function<void(HeaderMap& headers)> modify_headers,
+                        bool rate_limited_as_resource_exhausted = false) override {
+      parent_.sendLocalReply(is_grpc_request_, code, body, modify_headers, parent_.is_head_request_,
+                             rate_limited_as_resource_exhausted);
     }
     void encode100ContinueHeaders(HeaderMapPtr&& headers) override;
     void encodeHeaders(HeaderMapPtr&& headers, bool end_stream) override;
@@ -283,7 +284,7 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     HeaderMap& addEncodedTrailers();
     void sendLocalReply(bool is_grpc_request, Code code, const std::string& body,
                         std::function<void(HeaderMap& headers)> modify_headers,
-                        bool is_head_request);
+                        bool is_head_request, bool rate_limited_as_resource_exhausted = false);
     void encode100ContinueHeaders(ActiveStreamEncoderFilter* filter, HeaderMap& headers);
     void encodeHeaders(ActiveStreamEncoderFilter* filter, HeaderMap& headers, bool end_stream);
     void encodeData(ActiveStreamEncoderFilter* filter, Buffer::Instance& data, bool end_stream);

source/common/http/utility.cc

@@ -243,30 +243,33 @@ Utility::parseHttp1Settings(const envoy::api::v2::core::Http1ProtocolOptions& co
 
 void Utility::sendLocalReply(bool is_grpc, StreamDecoderFilterCallbacks& callbacks,
                              const bool& is_reset, Code response_code, const std::string& body_text,
-                             bool is_head_request) {
+                             bool is_head_request, bool rate_limited_as_resource_exhausted) {
   sendLocalReply(is_grpc,
                  [&](HeaderMapPtr&& headers, bool end_stream) -> void {
                    callbacks.encodeHeaders(std::move(headers), end_stream);
                  },
                  [&](Buffer::Instance& data, bool end_stream) -> void {
                    callbacks.encodeData(data, end_stream);
                  },
-                 is_reset, response_code, body_text, is_head_request);
+                 is_reset, response_code, body_text, is_head_request,
+                 rate_limited_as_resource_exhausted);
 }
 
 void Utility::sendLocalReply(
     bool is_grpc, std::function<void(HeaderMapPtr&& headers, bool end_stream)> encode_headers,
     std::function<void(Buffer::Instance& data, bool end_stream)> encode_data, const bool& is_reset,
-    Code response_code, const std::string& body_text, bool is_head_request) {
+    Code response_code, const std::string& body_text, bool is_head_request,
+    bool rate_limited_as_resource_exhausted) {
   // encode_headers() may reset the stream, so the stream must not be reset before calling it.
   ASSERT(!is_reset);
   // Respond with a gRPC trailers-only response if the request is gRPC
   if (is_grpc) {
-    HeaderMapPtr response_headers{new HeaderMapImpl{
-        {Headers::get().Status, std::to_string(enumToInt(Code::OK))},
-        {Headers::get().ContentType, Headers::get().ContentTypeValues.Grpc},
-        {Headers::get().GrpcStatus,
-         std::to_string(enumToInt(Grpc::Utility::httpToGrpcStatus(enumToInt(response_code))))}}};
+    HeaderMapPtr response_headers{
+        new HeaderMapImpl{{Headers::get().Status, std::to_string(enumToInt(Code::OK))},
+                          {Headers::get().ContentType, Headers::get().ContentTypeValues.Grpc},
+                          {Headers::get().GrpcStatus,
+                           std::to_string(enumToInt(Grpc::Utility::httpToGrpcStatus(
+                               enumToInt(response_code), rate_limited_as_resource_exhausted)))}}};
     if (!body_text.empty() && !is_head_request) {
       // TODO: GrpcMessage should be percent-encoded
       response_headers->insertGrpcMessage().value(body_text);

source/common/http/utility.h

@@ -139,7 +139,8 @@ Http1Settings parseHttp1Settings(const envoy::api::v2::core::Http1ProtocolOption
  * @param is_head_request tells if this is a response to a HEAD request
  */
 void sendLocalReply(bool is_grpc, StreamDecoderFilterCallbacks& callbacks, const bool& is_reset,
-                    Code response_code, const std::string& body_text, bool is_head_request);
+                    Code response_code, const std::string& body_text, bool is_head_request,
+                    bool rate_limited_as_resource_exhausted);
 
 /**
  * Create a locally generated response using the provided lambdas.
@@ -157,7 +158,7 @@ void sendLocalReply(bool is_grpc,
                     std::function<void(HeaderMapPtr&& headers, bool end_stream)> encode_headers,
                     std::function<void(Buffer::Instance& data, bool end_stream)> encode_data,
                     const bool& is_reset, Code response_code, const std::string& body_text,
-                    bool is_head_request = false);
+                    bool is_head_request = false, bool rate_limited_as_resource_exhausted = false);
 
 struct GetLastAddressFromXffInfo {
   // Last valid address pulled from the XFF header.

source/common/upstream/health_checker_impl.cc

@@ -402,8 +402,8 @@ void GrpcHealthCheckerImpl::GrpcActiveHealthCheckSession::decodeHeaders(
         return;
       }
     }
-    onRpcComplete(Grpc::Utility::httpToGrpcStatus(http_response_status), "non-200 HTTP response",
-                  end_stream);
+    onRpcComplete(Grpc::Utility::httpToGrpcStatus(http_response_status, false),
+                  "non-200 HTTP response", end_stream);
     return;
   }
   if (!Grpc::Common::hasGrpcContentType(*headers)) {

source/extensions/filters/http/ratelimit/ratelimit.cc

@@ -144,7 +144,8 @@ void Filter::complete(RateLimit::LimitStatus status, Http::HeaderMapPtr&& header
       config_->runtime().snapshot().featureEnabled("ratelimit.http_filter_enforcing", 100)) {
     state_ = State::Responded;
     callbacks_->sendLocalReply(Http::Code::TooManyRequests, "",
-                               [this](Http::HeaderMap& headers) { addHeaders(headers); });
+                               [this](Http::HeaderMap& headers) { addHeaders(headers); },
+                               config_->rateLimitedAsResourceExhausted());
     callbacks_->streamInfo().setResponseFlag(StreamInfo::ResponseFlag::RateLimited);
   } else if (status == RateLimit::LimitStatus::Error) {
     if (config_->failureModeAllow()) {
@@ -154,7 +155,8 @@ void Filter::complete(RateLimit::LimitStatus status, Http::HeaderMapPtr&& header
       }
     } else {
       state_ = State::Responded;
-      callbacks_->sendLocalReply(Http::Code::InternalServerError, "", nullptr);
+      callbacks_->sendLocalReply(Http::Code::InternalServerError, "", nullptr,
+                                 config_->rateLimitedAsResourceExhausted());
       callbacks_->streamInfo().setResponseFlag(StreamInfo::ResponseFlag::RateLimitServiceError);
     }
   } else if (!initiating_call_) {

source/extensions/filters/http/ratelimit/ratelimit.h

@@ -38,15 +38,16 @@ class FilterConfig {
         request_type_(config.request_type().empty() ? stringToType("both")
                                                     : stringToType(config.request_type())),
         local_info_(local_info), scope_(scope), runtime_(runtime),
-        failure_mode_deny_(config.failure_mode_deny()) {}
+        failure_mode_deny_(config.failure_mode_deny()),
+        rate_limited_as_resource_exhausted_(config.rate_limited_as_resource_exhausted()) {}
   const std::string& domain() const { return domain_; }
   const LocalInfo::LocalInfo& localInfo() const { return local_info_; }
   uint64_t stage() const { return stage_; }
   Runtime::Loader& runtime() { return runtime_; }
   Stats::Scope& scope() { return scope_; }
   FilterRequestType requestType() const { return request_type_; }
-
   bool failureModeAllow() const { return !failure_mode_deny_; }
+  bool rateLimitedAsResourceExhausted() const { return rate_limited_as_resource_exhausted_; }
 
 private:
   static FilterRequestType stringToType(const std::string& request_type) {
@@ -67,6 +68,7 @@ class FilterConfig {
   Stats::Scope& scope_;
   Runtime::Loader& runtime_;
   const bool failure_mode_deny_;
+  const bool rate_limited_as_resource_exhausted_;
 };
 
 typedef std::shared_ptr<FilterConfig> FilterConfigSharedPtr;

test/common/grpc/common_test.cc

@@ -259,10 +259,15 @@ TEST(GrpcCommonTest, HttpToGrpcStatus) {
       {500, Status::GrpcStatus::Unknown},
   };
   for (const auto& test_case : test_set) {
-    EXPECT_EQ(test_case.second, Grpc::Utility::httpToGrpcStatus(test_case.first));
+    EXPECT_EQ(test_case.second, Grpc::Utility::httpToGrpcStatus(test_case.first, false));
   }
 }
 
+TEST(GrpcCommonTest, HttpToGrpcStatusRateLimited) {
+  EXPECT_EQ(Status::GrpcStatus::Unavailable, Grpc::Utility::httpToGrpcStatus(429, false));
+  EXPECT_EQ(Status::GrpcStatus::ResourceExhausted, Grpc::Utility::httpToGrpcStatus(429, true));
+}
+
 TEST(GrpcCommonTest, HasGrpcContentType) {
   {
     Http::TestHeaderMapImpl headers{};

test/common/http/utility_test.cc

@@ -456,7 +456,8 @@ TEST(HttpUtility, SendLocalReply) {
 
   EXPECT_CALL(callbacks, encodeHeaders_(_, false));
   EXPECT_CALL(callbacks, encodeData(_, true));
-  Utility::sendLocalReply(false, callbacks, is_reset, Http::Code::PayloadTooLarge, "large", false);
+  Utility::sendLocalReply(false, callbacks, is_reset, Http::Code::PayloadTooLarge, "large", false,
+                          false);
 }
 
 TEST(HttpUtility, SendLocalGrpcReply) {
@@ -471,7 +472,26 @@ TEST(HttpUtility, SendLocalGrpcReply) {
         EXPECT_NE(headers.GrpcMessage(), nullptr);
         EXPECT_STREQ(headers.GrpcMessage()->value().c_str(), "large");
       }));
-  Utility::sendLocalReply(true, callbacks, is_reset, Http::Code::PayloadTooLarge, "large", false);
+  Utility::sendLocalReply(true, callbacks, is_reset, Http::Code::PayloadTooLarge, "large", false,
+                          false);
+}
+
+TEST(HttpUtility, RateLimitedGrpcStatus) {
+  MockStreamDecoderFilterCallbacks callbacks;
+
+  EXPECT_CALL(callbacks, encodeHeaders_(_, true))
+      .WillOnce(Invoke([&](const HeaderMap& headers, bool) -> void {
+        EXPECT_NE(headers.GrpcStatus(), nullptr);
+        EXPECT_STREQ(headers.GrpcStatus()->value().c_str(), "14"); // Unavailable
+      }));
+  Utility::sendLocalReply(true, callbacks, false, Http::Code::TooManyRequests, "", false, false);
+
+  EXPECT_CALL(callbacks, encodeHeaders_(_, true))
+      .WillOnce(Invoke([&](const HeaderMap& headers, bool) -> void {
+        EXPECT_NE(headers.GrpcStatus(), nullptr);
+        EXPECT_STREQ(headers.GrpcStatus()->value().c_str(), "8"); // ResourceExhausted
+      }));
+  Utility::sendLocalReply(true, callbacks, false, Http::Code::TooManyRequests, "", false, true);
 }
 
 TEST(HttpUtility, SendLocalReplyDestroyedEarly) {
@@ -482,7 +502,8 @@ TEST(HttpUtility, SendLocalReplyDestroyedEarly) {
     is_reset = true;
   }));
   EXPECT_CALL(callbacks, encodeData(_, true)).Times(0);
-  Utility::sendLocalReply(false, callbacks, is_reset, Http::Code::PayloadTooLarge, "large", false);
+  Utility::sendLocalReply(false, callbacks, is_reset, Http::Code::PayloadTooLarge, "large", false,
+                          false);
 }
 
 TEST(HttpUtility, SendLocalReplyHeadRequest) {
@@ -493,7 +514,8 @@ TEST(HttpUtility, SendLocalReplyHeadRequest) {
         EXPECT_STREQ(headers.ContentLength()->value().c_str(),
                      fmt::format("{}", strlen("large")).c_str());
       }));
-  Utility::sendLocalReply(false, callbacks, is_reset, Http::Code::PayloadTooLarge, "large", true);
+  Utility::sendLocalReply(false, callbacks, is_reset, Http::Code::PayloadTooLarge, "large", true,
+                          false);
 }
 
 TEST(HttpUtility, TestExtractHostPathFromUri) {