http/filter: Add a callback for sendLocalReply() and encode gRPC message for local responses when the request is gRPC

docs/root/intro/version_history.rst

@@ -48,6 +48,9 @@ Version history
   local configuration.
 * http: added the ability to pass DNS type Subject Alternative Names of the client certificate in the
   :ref:`config_http_conn_man_headers_x-forwarded-client-cert` header.
+* http: local responses to gRPC requests are now sent as trailers-only gRPC responses instead of plain HTTP responses.
+  Notably the HTTP response code is always "200" in this case, and the gRPC error code is carried in "grpc-status"
+  header, optionally accompanied with a text message in "grpc-message" header.
 * listeners: added :ref:`tcp_fast_open_queue_length <envoy_api_field_Listener.tcp_fast_open_queue_length>` option.
 * load balancing: added :ref:`weighted round robin
   <arch_overview_load_balancing_types_round_robin>` support. The round robin

include/envoy/http/filter.h

@@ -190,6 +190,17 @@ class StreamDecoderFilterCallbacks : public virtual StreamFilterCallbacks {
    */
   virtual void addDecodedData(Buffer::Instance& data, bool streaming_filter) PURE;
 
+  /**
+   * Create a locally generated response using the provided lambdas.
+   * @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 modify_headers supplies an optional callback function that can modify the
+   *                       response headers.
+   */
+  virtual void sendLocalReply(Code response_code, const std::string& body_text,
+                              std::function<void(HeaderMap& headers)> modify_headers) PURE;
+
   /**
    * Called with 100-Continue headers to be encoded.
    *

source/common/grpc/async_client_impl.cc

@@ -106,7 +106,7 @@ void AsyncStreamImpl::onHeaders(Http::HeaderMapPtr&& headers, bool end_stream) {
     }
     // Technically this should be
     // https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md
-    // as given by Common::httpToGrpcStatus(), but the Google gRPC client treats
+    // as given by Http::Utility::httpToGrpcStatus(), but the Google gRPC client treats
     // this as GrpcStatus::Canceled.
     streamError(Status::GrpcStatus::Canceled);
     return;

source/common/grpc/common.cc

@@ -124,88 +124,6 @@ bool Common::resolveServiceAndMethod(const Http::HeaderEntry* path, std::string*
   return true;
 }
 
-Status::GrpcStatus Common::httpToGrpcStatus(uint64_t http_response_status) {
-  // From
-  // https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md.
-  switch (http_response_status) {
-  case 400:
-    return Status::GrpcStatus::Internal;
-  case 401:
-    return Status::GrpcStatus::Unauthenticated;
-  case 403:
-    return Status::GrpcStatus::PermissionDenied;
-  case 404:
-    return Status::GrpcStatus::Unimplemented;
-  case 429:
-  case 502:
-  case 503:
-  case 504:
-    return Status::GrpcStatus::Unavailable;
-  default:
-    return Status::GrpcStatus::Unknown;
-  }
-}
-
-uint64_t Common::grpcToHttpStatus(Status::GrpcStatus grpc_status) {
-  // From https://cloud.google.com/apis/design/errors#handling_errors.
-  switch (grpc_status) {
-  case Status::GrpcStatus::Ok:
-    return 200;
-  case Status::GrpcStatus::Canceled:
-    // Client closed request.
-    return 499;
-  case Status::GrpcStatus::Unknown:
-    // Internal server error.
-    return 500;
-  case Status::GrpcStatus::InvalidArgument:
-    // Bad request.
-    return 400;
-  case Status::GrpcStatus::DeadlineExceeded:
-    // Gateway Time-out.
-    return 504;
-  case Status::GrpcStatus::NotFound:
-    // Not found.
-    return 404;
-  case Status::GrpcStatus::AlreadyExists:
-    // Conflict.
-    return 409;
-  case Status::GrpcStatus::PermissionDenied:
-    // Forbidden.
-    return 403;
-  case Status::GrpcStatus::ResourceExhausted:
-    //  Too many requests.
-    return 429;
-  case Status::GrpcStatus::FailedPrecondition:
-    // Bad request.
-    return 400;
-  case Status::GrpcStatus::Aborted:
-    // Conflict.
-    return 409;
-  case Status::GrpcStatus::OutOfRange:
-    // Bad request.
-    return 400;
-  case Status::GrpcStatus::Unimplemented:
-    // Not implemented.
-    return 501;
-  case Status::GrpcStatus::Internal:
-    // Internal server error.
-    return 500;
-  case Status::GrpcStatus::Unavailable:
-    // Service unavailable.
-    return 503;
-  case Status::GrpcStatus::DataLoss:
-    // Internal server error.
-    return 500;
-  case Status::GrpcStatus::Unauthenticated:
-    // Unauthorized.
-    return 401;
-  case Status::GrpcStatus::InvalidCode:
-  default:
-    // Internal server error.
-    return 500;
-  }
-}
-
 Buffer::InstancePtr Common::serializeBody(const Protobuf::Message& message) {
   // http://www.grpc.io/docs/guides/wire.html
   // Reserve enough space for the entire message and the 5 byte header.

source/common/grpc/common.h

@@ -56,21 +56,6 @@ class Common {
    */
   static std::string getGrpcMessage(const Http::HeaderMap& trailers);
 
-  /**
-   * 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.
-   * @param http_response_status HTTP status code.
-   * @return Status::GrpcStatus corresponding gRPC status code.
-   */
-  static Status::GrpcStatus httpToGrpcStatus(uint64_t http_response_status);
-
-  /**
-   * @param grpc_status gRPC status from grpc-status header.
-   * @return uint64_t the canonical HTTP status code corresponding to a gRPC status code.
-   */
-  static uint64_t grpcToHttpStatus(Status::GrpcStatus grpc_status);
-
   /**
    * Charge a success/failure stat to a cluster/service/method.
    * @param cluster supplies the target cluster.

source/common/http/BUILD

@@ -239,6 +239,7 @@ envoy_cc_library(
         ":exception_lib",
         ":header_map_lib",
         ":headers_lib",
+        "//include/envoy/grpc:status",
         "//include/envoy/http:codes_interface",
         "//include/envoy/http:filter_interface",
         "//include/envoy/http:header_map_interface",

source/common/http/async_client_impl.cc

@@ -6,6 +6,7 @@
 #include <string>
 #include <vector>
 
+#include "common/grpc/common.h"
 #include "common/http/utility.h"
 
 namespace Envoy {
@@ -110,6 +111,7 @@ void AsyncStreamImpl::encodeTrailers(HeaderMapPtr&& trailers) {
 }
 
 void AsyncStreamImpl::sendHeaders(HeaderMap& headers, bool end_stream) {
+  is_grpc_request_ = Grpc::Common::hasGrpcContentType(headers);
   headers.insertEnvoyInternalRequest().value().setReference(
       Headers::get().EnvoyInternalRequestValues.True);
   Utility::appendXff(headers, *parent_.config_.local_info_.address());

source/common/http/async_client_impl.h

@@ -261,6 +261,19 @@ class AsyncStreamImpl : public AsyncClient::Stream,
   void continueDecoding() override { NOT_IMPLEMENTED; }
   void addDecodedData(Buffer::Instance&, bool) override { NOT_IMPLEMENTED; }
   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 {
+    Utility::sendLocalReply(
+        is_grpc_request_,
+        [this, modify_headers](HeaderMapPtr&& headers, bool end_stream) -> void {
+          if (headers != nullptr && modify_headers != nullptr) {
+            modify_headers(*headers);
+          }
+          encodeHeaders(std::move(headers), end_stream);
+        },
+        [this](Buffer::Instance& data, bool end_stream) -> void { encodeData(data, end_stream); },
+        remote_closed_, code, body);
+  }
   // The async client won't pause if sending an Expect: 100-Continue so simply
   // swallows any incoming encode100Continue.
   void encode100ContinueHeaders(HeaderMapPtr&&) override {}
@@ -284,6 +297,7 @@ class AsyncStreamImpl : public AsyncClient::Stream,
   bool local_closed_{};
   bool remote_closed_{};
   Buffer::InstancePtr buffered_body_;
+  bool is_grpc_request_{};
   friend class AsyncClientImpl;
 };
 

source/common/http/conn_manager_impl.cc

@@ -473,10 +473,9 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
     // The protocol may have shifted in the HTTP/1.0 case so reset it.
     request_info_.protocol(protocol);
     if (!connection_manager_.config_.http1Settings().accept_http_10_) {
-      // Send "Upgrade Required" if HTTP/1.0 support is not expliictly configured on.
-      HeaderMapImpl headers{
-          {Headers::get().Status, std::to_string(enumToInt(Code::UpgradeRequired))}};
-      encodeHeaders(nullptr, headers, true);
+      // Send "Upgrade Required" if HTTP/1.0 support is not explictly configured on.
+      sendLocalReply(nullptr, Grpc::Common::hasGrpcContentType(*request_headers_),
+                     Code::UpgradeRequired, "", nullptr);
       return;
     } else {
       // HTTP/1.0 defaults to single-use connections. Make sure the connection
@@ -499,8 +498,8 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
           connection_manager_.config_.http1Settings().default_host_for_http_10_);
     } else {
       // Require host header. For HTTP/1.1 Host has already been translated to :authority.
-      HeaderMapImpl headers{{Headers::get().Status, std::to_string(enumToInt(Code::BadRequest))}};
-      encodeHeaders(nullptr, headers, true);
+      sendLocalReply(nullptr, Grpc::Common::hasGrpcContentType(*request_headers_), Code::BadRequest,
+                     "", nullptr);
       return;
     }
   }
@@ -513,9 +512,8 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
   // header size http_parser and nghttp2 will allow, down to 16k or 8k for
   // envoy users who do not wish to proxy large headers.
   if (request_headers_->byteSize() > (60 * 1024)) {
-    HeaderMapImpl headers{
-        {Headers::get().Status, std::to_string(enumToInt(Code::RequestHeaderFieldsTooLarge))}};
-    encodeHeaders(nullptr, headers, true);
+    sendLocalReply(nullptr, Grpc::Common::hasGrpcContentType(*request_headers_),
+                   Code::RequestHeaderFieldsTooLarge, "", nullptr);
     return;
   }
 
@@ -526,8 +524,8 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
   // don't support that currently.
   if (!request_headers_->Path() || request_headers_->Path()->value().c_str()[0] != '/') {
     connection_manager_.stats_.named_.downstream_rq_non_relative_path_.inc();
-    HeaderMapImpl headers{{Headers::get().Status, std::to_string(enumToInt(Code::NotFound))}};
-    encodeHeaders(nullptr, headers, true);
+    sendLocalReply(nullptr, Grpc::Common::hasGrpcContentType(*request_headers_), Code::NotFound, "",
+                   nullptr);
     return;
   }
 
@@ -570,8 +568,8 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(HeaderMapPtr&& headers,
     } else if (websocket_requested) {
       // Do not allow WebSocket upgrades if the route does not support it.
       connection_manager_.stats_.named_.downstream_rq_ws_on_non_ws_route_.inc();
-      HeaderMapImpl headers{{Headers::get().Status, std::to_string(enumToInt(Code::Forbidden))}};
-      encodeHeaders(nullptr, headers, true);
+      sendLocalReply(nullptr, Grpc::Common::hasGrpcContentType(*request_headers_), Code::Forbidden,
+                     "", nullptr);
       return;
     }
     // Allow non websocket requests to go through websocket enabled routes.
@@ -653,7 +651,7 @@ void ConnectionManagerImpl::ActiveStream::decodeHeaders(ActiveStreamDecoderFilte
   for (; entry != decoder_filters_.end(); entry++) {
     ASSERT(!(state_.filter_call_state_ & FilterCallState::DecodeHeaders));
     state_.filter_call_state_ |= FilterCallState::DecodeHeaders;
-    FilterHeadersStatus status = (*entry)->handle_->decodeHeaders(
+    FilterHeadersStatus status = (*entry)->decodeHeaders(
         headers, end_stream && continue_data_entry == decoder_filters_.end());
     state_.filter_call_state_ &= ~FilterCallState::DecodeHeaders;
     ENVOY_STREAM_LOG(trace, "decode headers called: filter={} status={}", *this,
@@ -821,6 +819,24 @@ void ConnectionManagerImpl::ActiveStream::refreshCachedRoute() {
   cached_route_ = std::move(route);
 }
 
+void ConnectionManagerImpl::ActiveStream::sendLocalReply(
+    ActiveStreamEncoderFilter* filter, bool is_grpc_request, Code code, const std::string& body,
+    std::function<void(HeaderMap& headers)> modify_headers) {
+  Utility::sendLocalReply(
+      is_grpc_request,
+      [this, filter, modify_headers](HeaderMapPtr&& headers, bool end_stream) -> void {
+        if (headers != nullptr && modify_headers != nullptr) {
+          modify_headers(*headers);
+        }
+        response_headers_ = std::move(headers);
+        encodeHeaders(filter, *response_headers_, end_stream);
+      },
+      [this, filter](Buffer::Instance& data, bool end_stream) -> void {
+        encodeData(filter, data, end_stream);
+      },
+      state_.destroyed_, code, body);
+}
+
 void ConnectionManagerImpl::ActiveStream::encode100ContinueHeaders(
     ActiveStreamEncoderFilter* filter, HeaderMap& headers) {
   ASSERT(connection_manager_.config_.proxy100Continue());
@@ -1316,8 +1332,7 @@ void ConnectionManagerImpl::ActiveStreamDecoderFilter::requestDataTooLarge() {
     onDecoderFilterAboveWriteBufferHighWatermark();
   } else {
     parent_.connection_manager_.stats_.named_.downstream_rq_too_large_.inc();
-    Http::Utility::sendLocalReply(*this, parent_.state_.destroyed_, Http::Code::PayloadTooLarge,
-                                  CodeUtility::toString(Http::Code::PayloadTooLarge));
+    sendLocalReply(Code::PayloadTooLarge, CodeUtility::toString(Code::PayloadTooLarge), nullptr);
   }
 }
 
@@ -1389,18 +1404,20 @@ void ConnectionManagerImpl::ActiveStreamEncoderFilter::responseDataTooLarge() {
       parent_.state_.encoder_filters_streaming_ = true;
       stopped_ = false;
 
-      Http::Utility::sendLocalReply(
-          [&](HeaderMapPtr&& response_headers, bool end_stream) -> void {
-            parent_.response_headers_ = std::move(response_headers);
-            parent_.response_encoder_->encodeHeaders(*parent_.response_headers_, end_stream);
-          },
-          [&](Buffer::Instance& data, bool end_stream) -> void {
-            parent_.response_encoder_->encodeData(data, end_stream);
-            parent_.state_.local_complete_ = end_stream;
-            parent_.maybeEndEncode(end_stream);
-          },
-          parent_.state_.destroyed_, Http::Code::InternalServerError,
-          CodeUtility::toString(Http::Code::InternalServerError));
+      Http::Utility::sendLocalReply(Grpc::Common::hasGrpcContentType(*parent_.request_headers_),
+                                    [&](HeaderMapPtr&& response_headers, bool end_stream) -> void {
+                                      parent_.response_headers_ = std::move(response_headers);
+                                      parent_.response_encoder_->encodeHeaders(
+                                          *parent_.response_headers_, end_stream);
+                                      parent_.state_.local_complete_ = end_stream;
+                                    },
+                                    [&](Buffer::Instance& data, bool end_stream) -> void {
+                                      parent_.response_encoder_->encodeData(data, end_stream);
+                                      parent_.state_.local_complete_ = end_stream;
+                                    },
+                                    parent_.state_.destroyed_, Http::Code::InternalServerError,
+                                    CodeUtility::toString(Http::Code::InternalServerError));
+      parent_.maybeEndEncode(parent_.state_.local_complete_);
     } else {
       resetStream();
     }

source/common/http/conn_manager_impl.h

@@ -25,8 +25,10 @@
 
 #include "common/buffer/watermark_buffer.h"
 #include "common/common/linked_object.h"
+#include "common/grpc/common.h"
 #include "common/http/conn_manager_config.h"
 #include "common/http/user_agent.h"
+#include "common/http/utility.h"
 #include "common/request_info/request_info_impl.h"
 #include "common/tracing/http_tracer_impl.h"
 
@@ -164,6 +166,10 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     const Buffer::Instance* decodingBuffer() override {
       return parent_.buffered_request_data_.get();
     }
+    void sendLocalReply(Code code, const std::string& body,
+                        std::function<void(HeaderMap& headers)> modify_headers) override {
+      parent_.sendLocalReply(nullptr, is_grpc_request_, code, body, modify_headers);
+    }
     void encode100ContinueHeaders(HeaderMapPtr&& headers) override;
     void encodeHeaders(HeaderMapPtr&& headers, bool end_stream) override;
     void encodeData(Buffer::Instance& data, bool end_stream) override;
@@ -177,10 +183,16 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     void setDecoderBufferLimit(uint32_t limit) override { parent_.setBufferLimit(limit); }
     uint32_t decoderBufferLimit() override { return parent_.buffer_limit_; }
 
+    FilterHeadersStatus decodeHeaders(HeaderMap& headers, bool end_stream) {
+      is_grpc_request_ = Grpc::Common::hasGrpcContentType(headers);
+      return handle_->decodeHeaders(headers, end_stream);
+    }
+
     void requestDataTooLarge();
     void requestDataDrained();
 
     StreamDecoderFilterSharedPtr handle_;
+    bool is_grpc_request_{};
   };
 
   typedef std::unique_ptr<ActiveStreamDecoderFilter> ActiveStreamDecoderFilterPtr;
@@ -257,6 +269,9 @@ class ConnectionManagerImpl : Logger::Loggable<Logger::Id::http>,
     void decodeTrailers(ActiveStreamDecoderFilter* filter, HeaderMap& trailers);
     void maybeEndDecode(bool end_stream);
     void addEncodedData(ActiveStreamEncoderFilter& filter, Buffer::Instance& data, bool streaming);
+    void sendLocalReply(ActiveStreamEncoderFilter* filter, bool is_grpc_request, Code code,
+                        const std::string& body,
+                        std::function<void(HeaderMap& headers)> modify_headers);
     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);