http/filter: Add a callback for sendLocalReply() and encode gRPC mess…

…age for local responses when the request is gRPC (#3299)

Some gRPC clients get confused when they get non-gRPC responses to
their gRPC requests. While such clients should be fixed, we can play
nice and format the local responses returned by Envoy router filter as
gRPC responses when the request is gRPC.

Fixes: #1930 

Signed-off-by: Jarno Rajahalme <[email protected]>

docs/root/intro/version_history.rst

@@ -49,6 +49,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,21 @@ class StreamDecoderFilterCallbacks : public virtual StreamFilterCallbacks {
    */
   virtual void addDecodedData(Buffer::Instance& data, bool streaming_filter) PURE;
 
+  /**
+   * Create a locally generated response using the provided response_code and body_text parameters.
+   * If the request was a gRPC request the local reply will be encoded as a gRPC response with a 200
+   * HTTP response code and grpc-status and grpc-message headers mapped from the provided
+   * parameters.
+   *
+   * @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, or encoded in the grpc-message header.
+   * @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/BUILD

@@ -46,13 +46,21 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "status_lib",
+    srcs = ["status.cc"],
+    hdrs = ["status.h"],
+    deps = [
+        "//include/envoy/grpc:status",
+    ],
+)
+
 envoy_cc_library(
     name = "common_lib",
     srcs = ["common.cc"],
     hdrs = ["common.h"],
     external_deps = ["abseil_optional"],
     deps = [
-        "//include/envoy/grpc:status",
         "//include/envoy/http:header_map_interface",
         "//include/envoy/http:message_interface",
         "//include/envoy/stats:stats_interface",
@@ -64,6 +72,7 @@ envoy_cc_library(
         "//source/common/common:enum_to_int",
         "//source/common/common:macros",
         "//source/common/common:utility_lib",
+        "//source/common/grpc:status_lib",
         "//source/common/http:filter_utility_lib",
         "//source/common/http:headers_lib",
         "//source/common/http:message_lib",

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 Grpc::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

@@ -10,6 +10,7 @@
 #include "envoy/http/message.h"
 #include "envoy/stats/stats.h"
 
+#include "common/grpc/status.h"
 #include "common/protobuf/protobuf.h"
 
 #include "absl/types/optional.h"
@@ -56,21 +57,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/grpc/status.cc

@@ -0,0 +1,89 @@
+#include "common/grpc/status.h"
+
+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.
+  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 Utility::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;
+  }
+}
+
+} // namespace Grpc
+} // namespace Envoy

source/common/grpc/status.h

@@ -0,0 +1,32 @@
+#pragma once
+
+#include <cstdint>
+
+#include "envoy/grpc/status.h"
+
+namespace Envoy {
+namespace Grpc {
+
+/**
+ * Grpc::Status utilities.
+ */
+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.
+   * @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);
+};
+
+} // namespace Grpc
+} // namespace Envoy

source/common/http/BUILD

@@ -247,6 +247,7 @@ envoy_cc_library(
         "//source/common/common:empty_string",
         "//source/common/common:enum_to_int",
         "//source/common/common:utility_lib",
+        "//source/common/grpc:status_lib",
         "//source/common/json:json_loader_lib",
         "//source/common/network:utility_lib",
         "//source/common/protobuf:utility_lib",

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 (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;
 };