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