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/grpc/status.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include <map>
+
 namespace Envoy {
 namespace Grpc {
 

include/envoy/http/BUILD

@@ -46,11 +46,13 @@ envoy_cc_library(
 envoy_cc_library(
     name = "filter_interface",
     hdrs = ["filter.h"],
+    external_deps = ["abseil_optional"],
     deps = [
         ":codec_interface",
         ":header_map_interface",
         "//include/envoy/access_log:access_log_interface",
         "//include/envoy/event:dispatcher_interface",
+        "//include/envoy/grpc:status",
         "//include/envoy/router:router_interface",
         "//include/envoy/ssl:connection_interface",
         "//include/envoy/tracing:http_tracer_interface",

include/envoy/http/filter.h

@@ -7,13 +7,16 @@
 
 #include "envoy/access_log/access_log.h"
 #include "envoy/event/dispatcher.h"
+#include "envoy/grpc/status.h"
 #include "envoy/http/codec.h"
 #include "envoy/http/header_map.h"
 #include "envoy/router/router.h"
 #include "envoy/ssl/connection.h"
 #include "envoy/tracing/http_tracer.h"
 #include "envoy/upstream/upstream.h"
 
+#include "absl/types/optional.h"
+
 namespace Envoy {
 namespace Http {
 
@@ -221,9 +224,11 @@ 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 grpc_status the gRPC status code to override the httpToGrpcStatus mapping with.
    */
   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,
+                              const absl::optional<Grpc::Status::GrpcStatus> grpc_status) PURE;
 
   /**
    * Called with 100-Continue headers to be encoded.

source/common/grpc/BUILD

@@ -50,6 +50,7 @@ envoy_cc_library(
     name = "status_lib",
     srcs = ["status.cc"],
     hdrs = ["status.h"],
+    external_deps = ["abseil_optional"],
     deps = [
         "//include/envoy/grpc:status",
     ],

source/common/grpc/status.cc

@@ -3,9 +3,16 @@
 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,
+                                             const absl::optional<Status::GrpcStatus> grpc_status) {
+  // See:
+  // * https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md
+  // * https://cloud.google.com/apis/design/errors#generating_errors
+
+  if (grpc_status) {
+    return grpc_status.value();
+  }
+
   switch (http_response_status) {
   case 400:
     return Status::GrpcStatus::Internal;

source/common/grpc/status.h

@@ -4,6 +4,8 @@
 
 #include "envoy/grpc/status.h"
 
+#include "absl/types/optional.h"
+
 namespace Envoy {
 namespace Grpc {
 
@@ -13,13 +15,22 @@ 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 grpc_status the gRPC status code to override the default mapping with.
    * @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,
+                                             const absl::optional<Status::GrpcStatus> grpc_status);
 
   /**
    * @param grpc_status gRPC status from grpc-status header.

source/common/http/BUILD

@@ -238,6 +238,7 @@ envoy_cc_library(
     name = "utility_lib",
     srcs = ["utility.cc"],
     hdrs = ["utility.h"],
+    external_deps = ["abseil_optional"],
     deps = [
         ":exception_lib",
         ":header_map_lib",

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,
+                      const absl::optional<Grpc::Status::GrpcStatus> grpc_status) 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, grpc_status, is_head_request_);
   }
   // 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

@@ -430,9 +430,9 @@ void ConnectionManagerImpl::ActiveStream::onIdleTimeout() {
     // or gRPC status code, and/or set H2 RST_STREAM error.
     connection_manager_.doEndStream(*this);
   } else {
-    sendLocalReply(request_headers_ != nullptr &&
-                       Grpc::Common::hasGrpcContentType(*request_headers_),
-                   Http::Code::RequestTimeout, "stream timeout", nullptr, is_head_request_);
+    sendLocalReply(
+        request_headers_ != nullptr && Grpc::Common::hasGrpcContentType(*request_headers_),
+        Http::Code::RequestTimeout, "stream timeout", nullptr, is_head_request_, absl::nullopt);
   }
 }
 
@@ -504,7 +504,8 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
       Server::OverloadActionState::Active) {
     connection_manager_.stats_.named_.downstream_rq_overload_close_.inc();
     sendLocalReply(Grpc::Common::hasGrpcContentType(*request_headers_),
-                   Http::Code::ServiceUnavailable, "envoy overloaded", nullptr, is_head_request_);
+                   Http::Code::ServiceUnavailable, "envoy overloaded", nullptr, is_head_request_,
+                   absl::nullopt);
     return;
   }
 
@@ -537,7 +538,7 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
     stream_info_.protocol(protocol);
     if (!connection_manager_.config_.http1Settings().accept_http_10_) {
       // Send "Upgrade Required" if HTTP/1.0 support is not explicitly configured on.
-      sendLocalReply(false, Code::UpgradeRequired, "", nullptr, is_head_request_);
+      sendLocalReply(false, Code::UpgradeRequired, "", nullptr, is_head_request_, absl::nullopt);
       return;
     } else {
       // HTTP/1.0 defaults to single-use connections. Make sure the connection
@@ -561,7 +562,7 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
     } else {
       // Require host header. For HTTP/1.1 Host has already been translated to :authority.
       sendLocalReply(Grpc::Common::hasGrpcContentType(*request_headers_), Code::BadRequest, "",
-                     nullptr, is_head_request_);
+                     nullptr, is_head_request_, absl::nullopt);
       return;
     }
   }
@@ -575,7 +576,7 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
   // envoy users who do not wish to proxy large headers.
   if (request_headers_->byteSize() > (60 * 1024)) {
     sendLocalReply(Grpc::Common::hasGrpcContentType(*request_headers_),
-                   Code::RequestHeaderFieldsTooLarge, "", nullptr, is_head_request_);
+                   Code::RequestHeaderFieldsTooLarge, "", nullptr, is_head_request_, absl::nullopt);
     return;
   }
 
@@ -587,7 +588,7 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
   if (!request_headers_->Path() || request_headers_->Path()->value().c_str()[0] != '/') {
     connection_manager_.stats_.named_.downstream_rq_non_relative_path_.inc();
     sendLocalReply(Grpc::Common::hasGrpcContentType(*request_headers_), Code::NotFound, "", nullptr,
-                   is_head_request_);
+                   is_head_request_, absl::nullopt);
     return;
   }
 
@@ -615,7 +616,7 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
       // Do not allow upgrades if the route does not support it.
       connection_manager_.stats_.named_.downstream_rq_ws_on_non_ws_route_.inc();
       sendLocalReply(Grpc::Common::hasGrpcContentType(*request_headers_), Code::Forbidden, "",
-                     nullptr, is_head_request_);
+                     nullptr, is_head_request_, absl::nullopt);
       return;
     }
     // Allow non websocket requests to go through websocket enabled routes.
@@ -929,7 +930,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,
+    const absl::optional<Grpc::Status::GrpcStatus> grpc_status) {
   Utility::sendLocalReply(is_grpc_request,
                           [this, modify_headers](HeaderMapPtr&& headers, bool end_stream) -> void {
                             if (modify_headers != nullptr) {
@@ -945,7 +947,7 @@ void ConnectionManagerImpl::ActiveStream::sendLocalReply(
                             // request instead.
                             encodeData(nullptr, data, end_stream);
                           },
-                          state_.destroyed_, code, body, is_head_request);
+                          state_.destroyed_, code, body, grpc_status, is_head_request);
 }
 
 void ConnectionManagerImpl::ActiveStream::encode100ContinueHeaders(
@@ -1535,7 +1537,8 @@ void ConnectionManagerImpl::ActiveStreamDecoderFilter::requestDataTooLarge() {
     onDecoderFilterAboveWriteBufferHighWatermark();
   } else {
     parent_.connection_manager_.stats_.named_.downstream_rq_too_large_.inc();
-    sendLocalReply(Code::PayloadTooLarge, CodeUtility::toString(Code::PayloadTooLarge), nullptr);
+    sendLocalReply(Code::PayloadTooLarge, CodeUtility::toString(Code::PayloadTooLarge), nullptr,
+                   absl::nullopt);
   }
 }
 
@@ -1623,7 +1626,8 @@ 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), absl::nullopt,
+          parent_.is_head_request_);
       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,
+                        const absl::optional<Grpc::Status::GrpcStatus> grpc_status) override {
+      parent_.sendLocalReply(is_grpc_request_, code, body, modify_headers, parent_.is_head_request_,
+                             grpc_status);
     }
     void encode100ContinueHeaders(HeaderMapPtr&& headers) override;
     void encodeHeaders(HeaderMapPtr&& headers, bool end_stream) override;
@@ -283,7 +284,8 @@ 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,
+                        const absl::optional<Grpc::Status::GrpcStatus> grpc_status);
     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,6 +243,7 @@ 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,
+                             const absl::optional<Grpc::Status::GrpcStatus> grpc_status,
                              bool is_head_request) {
   sendLocalReply(is_grpc,
                  [&](HeaderMapPtr&& headers, bool end_stream) -> void {
@@ -251,22 +252,23 @@ void Utility::sendLocalReply(bool is_grpc, StreamDecoderFilterCallbacks& callbac
                  [&](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, grpc_status, is_head_request);
 }
 
 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,
+    const absl::optional<Grpc::Status::GrpcStatus> grpc_status, bool is_head_request) {
   // 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))))}}};
+        {Headers::get().GrpcStatus, std::to_string(enumToInt(Grpc::Utility::httpToGrpcStatus(
+                                        enumToInt(response_code), grpc_status)))}}};
     if (!body_text.empty() && !is_head_request) {
       // TODO: GrpcMessage should be percent-encoded
       response_headers->insertGrpcMessage().value(body_text);

source/common/http/utility.h

@@ -15,6 +15,7 @@
 #include "common/json/json_loader.h"
 
 #include "absl/strings/string_view.h"
+#include "absl/types/optional.h"
 
 namespace Envoy {
 namespace Http {
@@ -136,10 +137,13 @@ Http1Settings parseHttp1Settings(const envoy::api::v2::core::Http1ProtocolOption
  * @param response_code supplies the HTTP response code.
  * @param body_text supplies the optional body text which is sent using the text/plain content
  *                  type.
+ * @param grpc_status the gRPC status code to override the httpToGrpcStatus mapping with.
  * @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,
+                    const absl::optional<Grpc::Status::GrpcStatus> grpc_status,
+                    bool is_head_request);
 
 /**
  * Create a locally generated response using the provided lambdas.
@@ -152,11 +156,13 @@ void sendLocalReply(bool is_grpc, StreamDecoderFilterCallbacks& callbacks, const
  * @param response_code supplies the HTTP response code.
  * @param body_text supplies the optional body text which is sent using the text/plain content
  *                  type.
+ * @param grpc_status the gRPC status code to override the httpToGrpcStatus mapping with.
  */
 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,
+                    const absl::optional<Grpc::Status::GrpcStatus> grpc_status,
                     bool is_head_request = false);
 
 struct GetLastAddressFromXffInfo {