grpc-json: handle google.api.HttpBody when building HTTP response

api/bazel/api_build_system.bzl

@@ -32,6 +32,7 @@ def api_py_proto_library(name, srcs = [], deps = [], has_services = 0):
         protoc = "@com_google_protobuf//:protoc",
         deps = [_LibrarySuffix(d, _PY_SUFFIX) for d in deps] + [
             "@com_lyft_protoc_gen_validate//validate:validate_py",
+            "@googleapis//:api_httpbody_protos_py",
             "@googleapis//:http_api_protos_py",
             "@googleapis//:rpc_status_protos_py",
             "@com_github_gogo_protobuf//:gogo_proto_py",
@@ -108,6 +109,7 @@ def api_proto_library(name, visibility = ["//visibility:private"], srcs = [], de
             "@com_google_protobuf//:struct_proto",
             "@com_google_protobuf//:timestamp_proto",
             "@com_google_protobuf//:wrappers_proto",
+            "@googleapis//:api_httpbody_protos_proto",
             "@googleapis//:http_api_protos_proto",
             "@googleapis//:rpc_status_protos_lib",
             "@com_github_gogo_protobuf//:gogo_proto",

api/bazel/repositories.bzl

@@ -21,14 +21,62 @@ def api_dependencies():
 load("@com_google_protobuf//:protobuf.bzl", "cc_proto_library", "py_proto_library")
 load("@io_bazel_rules_go//proto:def.bzl", "go_proto_library")
 
+filegroup(
+    name = "api_httpbody_protos_src",
+    srcs = [
+        "google/api/httpbody.proto",
+    ],
+    visibility = ["//visibility:public"],
+)
+
+proto_library(
+    name = "api_httpbody_protos_proto",
+    srcs = [":api_httpbody_protos_src"],
+    deps = ["@com_google_protobuf//:descriptor_proto"],
+    visibility = ["//visibility:public"],
+)
+
+cc_proto_library(
+    name = "api_httpbody_protos",
+    srcs = [
+        "google/api/httpbody.proto",
+    ],
+    default_runtime = "@com_google_protobuf//:protobuf",
+    protoc = "@com_google_protobuf//:protoc",
+    deps = ["@com_google_protobuf//:cc_wkt_protos"],
+    visibility = ["//visibility:public"],
+)
+
+py_proto_library(
+    name = "api_httpbody_protos_py",
+    srcs = [
+        "google/api/httpbody.proto",
+    ],
+    include = ".",
+    default_runtime = "@com_google_protobuf//:protobuf_python",
+    protoc = "@com_google_protobuf//:protoc",
+    visibility = ["//visibility:public"],
+    deps = ["@com_google_protobuf//:protobuf_python"],
+)
+
+go_proto_library(
+    name = "api_httpbody_go_proto",
+    importpath = "google.golang.org/genproto/googleapis/api/httpbody",
+    proto = ":api_httpbody_protos_proto",
+    visibility = ["//visibility:public"],
+    deps = [
+      ":descriptor_go_proto",
+    ],
+)
+
 filegroup(
     name = "http_api_protos_src",
     srcs = [
         "google/api/annotations.proto",
         "google/api/http.proto",
     ],
     visibility = ["//visibility:public"],
- )
+)
 
 go_proto_library(
     name = "descriptor_go_proto",
@@ -93,6 +141,7 @@ proto_library(
      deps = ["@com_google_protobuf//:any_proto"],
      visibility = ["//visibility:public"],
 )
+
 cc_proto_library(
      name = "rpc_status_protos",
      srcs = ["google/rpc/status.proto"],

bazel/envoy_build_system.bzl

@@ -366,6 +366,10 @@ def envoy_proto_library(name, srcs = [], deps = [], external_deps = [],
     cc_proto_deps = []
     py_proto_deps = ["@com_google_protobuf//:protobuf_python"]
 
+    if "api_httpbody_protos" in external_deps:
+        cc_proto_deps.append("@googleapis//:api_httpbody_protos")
+        py_proto_deps.append("@googleapis//:api_httpbody_protos_py")
+
     if "http_api_protos" in external_deps:
         cc_proto_deps.append("@googleapis//:http_api_protos")
         py_proto_deps.append("@googleapis//:http_api_protos_py")
@@ -403,6 +407,10 @@ def envoy_proto_descriptor(name, out, srcs = [], external_deps = []):
     input_files = ["$(location " + src + ")" for src in srcs]
     include_paths = [".", PACKAGE_NAME]
 
+    if "api_httpbody_protos" in external_deps:
+        srcs.append("@googleapis//:api_httpbody_protos_src")
+        include_paths.append("external/googleapis")
+
     if "http_api_protos" in external_deps:
         srcs.append("@googleapis//:http_api_protos_src")
         include_paths.append("external/googleapis")

bazel/repositories.bzl

@@ -182,6 +182,10 @@ def _envoy_api_deps():
     if "envoy_api" not in native.existing_rules().keys():
         _default_envoy_api(name="envoy_api")
 
+    native.bind(
+        name = "api_httpbody_protos",
+        actual = "@googleapis//:api_httpbody_protos",
+    )
     native.bind(
         name = "http_api_protos",
         actual = "@googleapis//:http_api_protos",

docs/root/configuration/http_filters/grpc_json_transcoder_filter.rst

@@ -62,3 +62,14 @@ match the incoming request path, set `match_incoming_request_route` to true.
       };
     }
   }
+
+Sending arbitrary content
+-------------------------
+
+By default, when transcoding occurs, gRPC-JSON encodes the message output of a gRPC service method into
+JSON and sets the response `Content-Type` header to `application/json`. To send abritrary content, a gRPC
+service method can use
+`google.api.HttpBody <https://github.com/googleapis/googleapis/blob/master/google/api/httpbody.proto>`_
+as its output message type. The implementation needs to set
+`content_type <https://github.com/googleapis/googleapis/blob/master/google/api/httpbody.proto#L68>`_
+and `data <https://github.com/googleapis/googleapis/blob/master/google/api/httpbody.proto#L71>`_ accordingly.

docs/root/intro/version_history.rst

@@ -7,6 +7,8 @@ Version history
   to filter based on the presence of Envoy response flags.
 * admin: added :http:get:`/hystrix_event_stream` as an endpoint for monitoring envoy's statistics
   through `Hystrix dashboard <https://github.com/Netflix-Skunkworks/hystrix-dashboard/wiki>`_.
+* grpc-json: added support for building HTTP response from
+  `google.api.HttpBody <https://github.com/googleapis/googleapis/blob/master/google/api/httpbody.proto>`_.
 * config: v1 disabled by default. v1 support remains available until October via flipping --v2-config-only=false.
 * health check: added support for :ref:`custom health check <envoy_api_field_core.HealthCheck.custom_health_check>`.
 * health_check: added support for :ref:`health check event logging <arch_overview_health_check_logging>`.

source/extensions/filters/http/grpc_json_transcoder/BUILD

@@ -19,11 +19,11 @@ envoy_cc_library(
         "path_matcher",
         "grpc_transcoding",
         "http_api_protos",
+        "api_httpbody_protos",
     ],
     deps = [
         ":transcoder_input_stream_lib",
         "//include/envoy/http:filter_interface",
-        "//source/common/common:base64_lib",
         "//source/common/grpc:codec_lib",
         "//source/common/grpc:common_lib",
         "//source/common/http:headers_lib",

source/extensions/filters/http/grpc_json_transcoder/json_transcoder_filter.cc

@@ -14,6 +14,7 @@
 
 #include "google/api/annotations.pb.h"
 #include "google/api/http.pb.h"
+#include "google/api/httpbody.pb.h"
 #include "grpc_transcoding/json_request_translator.h"
 #include "grpc_transcoding/path_matcher_utility.h"
 #include "grpc_transcoding/response_to_json_translator.h"
@@ -222,6 +223,7 @@ Http::FilterHeadersStatus JsonTranscoderFilter::decodeHeaders(Http::HeaderMap& h
     // just pass-through the request to upstream.
     return Http::FilterHeadersStatus::Continue;
   }
+  has_http_body_output_ = hasHttpBodyAsOutputType();
 
   headers.removeContentLength();
   headers.insertContentType().value().setReference(Http::Headers::get().ContentTypeValues.Grpc);
@@ -340,6 +342,11 @@ Http::FilterDataStatus JsonTranscoderFilter::encodeData(Buffer::Instance& data,
     return Http::FilterDataStatus::Continue;
   }
 
+  if (has_http_body_output_) {
+    buildResponseFromHttpBodyOutput(*response_headers_, data);
+    return Http::FilterDataStatus::StopIterationAndBuffer;
+  }
+
   response_in_.move(data);
 
   if (end_stream) {
@@ -352,6 +359,7 @@ Http::FilterDataStatus JsonTranscoderFilter::encodeData(Buffer::Instance& data,
     // Buffer until the response is complete.
     return Http::FilterDataStatus::StopIterationAndBuffer;
   }
+  // TODO(dio): Add support for streaming case.
   // TODO(lizan): Check ResponseStatus
 
   return Http::FilterDataStatus::Continue;
@@ -415,6 +423,32 @@ bool JsonTranscoderFilter::readToBuffer(Protobuf::io::ZeroCopyInputStream& strea
   return false;
 }
 
+void JsonTranscoderFilter::buildResponseFromHttpBodyOutput(Http::HeaderMap& response_headers,
+                                                           Buffer::Instance& data) {
+  std::vector<Grpc::Frame> frames;
+  decoder_.decode(data, frames);
+  if (frames.empty()) {
+    return;
+  }
+
+  google::api::HttpBody http_body;
+  for (auto& frame : frames) {
+    if (frame.length_ > 0) {
+      Buffer::ZeroCopyInputStreamImpl stream(std::move(frame.data_));
+      http_body.ParseFromZeroCopyStream(&stream);
+      const auto& body = http_body.data();
+      data.add(body);
+      response_headers.insertContentType().value(http_body.content_type());
+      response_headers.insertContentLength().value(body.length());
+      return;
+    }
+  }
+}
+
+bool JsonTranscoderFilter::hasHttpBodyAsOutputType() {
+  return method_->output_type()->full_name() == google::api::HttpBody::descriptor()->full_name();
+}
+
 } // namespace GrpcJsonTranscoder
 } // namespace HttpFilters
 } // namespace Extensions

source/extensions/filters/http/grpc_json_transcoder/json_transcoder_filter.h

@@ -7,6 +7,7 @@
 #include "envoy/json/json_object.h"
 
 #include "common/common/logger.h"
+#include "common/grpc/codec.h"
 #include "common/protobuf/protobuf.h"
 
 #include "extensions/filters/http/grpc_json_transcoder/transcoder_input_stream_impl.h"
@@ -117,6 +118,8 @@ class JsonTranscoderFilter : public Http::StreamFilter, public Logger::Loggable<
 
 private:
   bool readToBuffer(Protobuf::io::ZeroCopyInputStream& stream, Buffer::Instance& data);
+  void buildResponseFromHttpBodyOutput(Http::HeaderMap& response_headers, Buffer::Instance& data);
+  bool hasHttpBodyAsOutputType();
 
   JsonTranscoderConfig& config_;
   std::unique_ptr<google::grpc::transcoding::Transcoder> transcoder_;
@@ -126,8 +129,10 @@ class JsonTranscoderFilter : public Http::StreamFilter, public Logger::Loggable<
   Http::StreamEncoderFilterCallbacks* encoder_callbacks_{nullptr};
   const Protobuf::MethodDescriptor* method_{nullptr};
   Http::HeaderMap* response_headers_{nullptr};
+  Grpc::Decoder decoder_;
 
   bool error_{false};
+  bool has_http_body_output_{false};
 };
 
 } // namespace GrpcJsonTranscoder

test/extensions/filters/http/grpc_json_transcoder/json_transcoder_filter_test.cc

@@ -550,6 +550,47 @@ TEST_F(GrpcJsonTranscoderFilterTest, TranscodingUnaryNotGrpcResponse) {
   EXPECT_EQ(Http::FilterDataStatus::Continue, filter_.encodeData(request_data, true));
 }
 
+TEST_F(GrpcJsonTranscoderFilterTest, TranscodingUnaryWithHttpBodyAsOutput) {
+  Http::TestHeaderMapImpl request_headers{{":method", "GET"}, {":path", "/index"}};
+
+  EXPECT_CALL(decoder_callbacks_, clearRouteCache());
+
+  EXPECT_EQ(Http::FilterHeadersStatus::Continue, filter_.decodeHeaders(request_headers, false));
+  EXPECT_EQ("application/grpc", request_headers.get_("content-type"));
+  EXPECT_EQ("/index", request_headers.get_("x-envoy-original-path"));
+  EXPECT_EQ("/bookstore.Bookstore/GetIndex", request_headers.get_(":path"));
+  EXPECT_EQ("trailers", request_headers.get_("te"));
+
+  Http::TestHeaderMapImpl continue_headers{{":status", "000"}};
+  EXPECT_EQ(Http::FilterHeadersStatus::Continue,
+            filter_.encode100ContinueHeaders(continue_headers));
+
+  Http::TestHeaderMapImpl response_headers{{"content-type", "application/grpc"},
+                                           {":status", "200"}};
+
+  EXPECT_EQ(Http::FilterHeadersStatus::StopIteration,
+            filter_.encodeHeaders(response_headers, false));
+  EXPECT_EQ("application/json", response_headers.get_("content-type"));
+
+  google::api::HttpBody response;
+  response.set_content_type("text/html");
+  response.set_data("<h1>Hello, world!</h1>");
+
+  auto response_data = Grpc::Common::serializeBody(response);
+
+  EXPECT_EQ(Http::FilterDataStatus::StopIterationAndBuffer,
+            filter_.encodeData(*response_data, false));
+
+  std::string response_html = response_data->toString();
+
+  EXPECT_EQ("text/html", response_headers.get_("content-type"));
+  EXPECT_EQ("<h1>Hello, world!</h1>", response_html);
+
+  Http::TestHeaderMapImpl response_trailers{{"grpc-status", "0"}, {"grpc-message", ""}};
+
+  EXPECT_EQ(Http::FilterTrailersStatus::Continue, filter_.decodeTrailers(response_trailers));
+}
+
 struct GrpcJsonTranscoderFilterPrintTestParam {
   std::string config_json_;
   std::string expected_response_;

test/proto/BUILD

@@ -20,8 +20,9 @@ envoy_proto_library(
     name = "bookstore_proto",
     srcs = [":bookstore.proto"],
     external_deps = [
-        "well_known_protos",
+        "api_httpbody_protos",
         "http_api_protos",
+        "well_known_protos",
     ],
 )
 
@@ -32,6 +33,7 @@ envoy_proto_descriptor(
     ],
     out = "bookstore.descriptor",
     external_deps = [
+        "api_httpbody_protos",
         "http_api_protos",
         "well_known_protos",
     ],

test/proto/bookstore.proto

@@ -3,6 +3,7 @@ syntax = "proto3";
 package bookstore;
 
 import "google/api/annotations.proto";
+import "google/api/httpbody.proto";
 import "google/protobuf/empty.proto";
 
 // A simple Bookstore API.
@@ -84,6 +85,11 @@ service Bookstore {
       get: "/authors/{author}"
     };
   }
+  rpc GetIndex(google.protobuf.Empty) returns (google.api.HttpBody) {
+    option (google.api.http) = {
+      get: "/index"
+    };
+  }
 }
 
 // A shelf resource.