transcoding: Handle RequestStatus error (istio#305)

* Handle RequestStatus error

* Add test for RequestStatus error

Author: lizan

src/envoy/transcoding/filter.cc

@@ -14,6 +14,7 @@
  */
 #include "src/envoy/transcoding/filter.h"
 
+#include "common/http/utility.h"
 #include "google/protobuf/descriptor.h"
 #include "google/protobuf/descriptor.pb.h"
 #include "google/protobuf/message.h"
@@ -59,6 +60,17 @@ Http::FilterHeadersStatus Instance::decodeHeaders(Http::HeaderMap& headers,
 
       request_in_.Finish();
 
+      const auto& request_status = transcoder_->RequestStatus();
+      if (!request_status.ok()) {
+        log().debug("Transcoding request error " + request_status.ToString());
+        error_ = true;
+        Http::Utility::sendLocalReply(
+            *decoder_callbacks_, Http::Code::BadRequest,
+            request_status.error_message().ToString());
+
+        return Http::FilterHeadersStatus::StopIteration;
+      }
+
       Buffer::OwnedImpl data;
       ReadToBuffer(transcoder_->RequestOutput(), data);
 
@@ -67,7 +79,7 @@ Http::FilterHeadersStatus Instance::decodeHeaders(Http::HeaderMap& headers,
       }
     }
   } else {
-    log().debug("No transcoding" + status.ToString());
+    log().debug("No transcoding: " + status.ToString());
   }
 
   return Http::FilterHeadersStatus::Continue;
@@ -76,12 +88,30 @@ Http::FilterHeadersStatus Instance::decodeHeaders(Http::HeaderMap& headers,
 Http::FilterDataStatus Instance::decodeData(Buffer::Instance& data,
                                             bool end_stream) {
   log().debug("Transcoding::Instance::decodeData");
+
+  if (error_) {
+    return Http::FilterDataStatus::StopIterationNoBuffer;
+  }
+
   if (transcoder_) {
     request_in_.Move(data);
 
+    if (end_stream) {
+      request_in_.Finish();
+    }
+
     ReadToBuffer(transcoder_->RequestOutput(), data);
 
-    // TODO: Check RequesStatus
+    const auto& request_status = transcoder_->RequestStatus();
+
+    if (!request_status.ok()) {
+      log().debug("Transcoding request error " + request_status.ToString());
+      error_ = true;
+      Http::Utility::sendLocalReply(*decoder_callbacks_, Http::Code::BadRequest,
+                                    request_status.error_message().ToString());
+
+      return Http::FilterDataStatus::StopIterationNoBuffer;
+    }
   }
 
   return Http::FilterDataStatus::Continue;
@@ -111,6 +141,10 @@ void Instance::setDecoderFilterCallbacks(
 Http::FilterHeadersStatus Instance::encodeHeaders(Http::HeaderMap& headers,
                                                   bool end_stream) {
   log().debug("Transcoding::Instance::encodeHeaders");
+  if (error_) {
+    return Http::FilterHeadersStatus::Continue;
+  }
+
   if (transcoder_) {
     headers.insertContentType().value(kJsonContentType);
   }
@@ -120,6 +154,10 @@ Http::FilterHeadersStatus Instance::encodeHeaders(Http::HeaderMap& headers,
 Http::FilterDataStatus Instance::encodeData(Buffer::Instance& data,
                                             bool end_stream) {
   log().debug("Transcoding::Instance::encodeData");
+  if (error_) {
+    return Http::FilterDataStatus::Continue;
+  }
+
   if (transcoder_) {
     response_in_.Move(data);
 

src/envoy/transcoding/filter.h

@@ -64,6 +64,8 @@ class Instance : public Http::StreamFilter,
   EnvoyInputStream response_in_;
   Http::StreamDecoderFilterCallbacks* decoder_callbacks_{nullptr};
   Http::StreamEncoderFilterCallbacks* encoder_callbacks_{nullptr};
+
+  bool error_{false};
 };
 
 }  // namespace Transcoding

src/envoy/transcoding/filter_test.cc

@@ -31,6 +31,7 @@
 #include "gtest/gtest.h"
 
 using testing::_;
+using testing::Invoke;
 using testing::NiceMock;
 using testing::Return;
 using testing::ReturnPointee;
@@ -145,6 +146,30 @@ TEST_F(GrpcHttpJsonTranscodingFilterTest, TranscodingUnaryPost) {
             filter_.decodeTrailers(response_trailers));
 }
 
+TEST_F(GrpcHttpJsonTranscodingFilterTest, TranscodingUnaryError) {
+  Http::TestHeaderMapImpl request_headers{{"content-type", "application/json"},
+                                          {":method", "POST"},
+                                          {":path", "/shelf"}};
+
+  EXPECT_EQ(Http::FilterHeadersStatus::Continue,
+            filter_.decodeHeaders(request_headers, false));
+  EXPECT_EQ("application/grpc", request_headers.get_("content-type"));
+  EXPECT_EQ("/bookstore.Bookstore/CreateShelf", request_headers.get_(":path"));
+  EXPECT_EQ("trailers", request_headers.get_("te"));
+
+  Buffer::OwnedImpl request_data{"{\"theme\": \"Children\""};
+
+  EXPECT_CALL(decoder_callbacks_, encodeHeaders_(_, false))
+      .WillOnce(Invoke([](Http::HeaderMap& headers, bool end_stream) {
+        EXPECT_STREQ("400", headers.Status()->value().c_str());
+      }));
+  EXPECT_CALL(decoder_callbacks_, encodeData(_, true));
+
+  EXPECT_EQ(Http::FilterDataStatus::StopIterationNoBuffer,
+            filter_.decodeData(request_data, true));
+  EXPECT_EQ(0, request_data.length());
+}
+
 }  // namespace Transcoding
 }  // namespace Grpc
 }  // namespace Envoy