ext-proc: Support "immediate_response" options for request headers

source/extensions/filters/http/ext_proc/BUILD

@@ -22,6 +22,7 @@ envoy_cc_library(
         "//source/extensions/filters/http/common:pass_through_filter_lib",
         "@com_google_absl//absl/strings:str_format",
         "@envoy_api//envoy/extensions/filters/http/ext_proc/v3alpha:pkg_cc_proto",
+        "@envoy_api//envoy/service/ext_proc/v3alpha:pkg_cc_proto",
     ],
 )
 

source/extensions/filters/http/ext_proc/ext_proc.cc

@@ -9,6 +9,7 @@ namespace Extensions {
 namespace HttpFilters {
 namespace ExternalProcessing {
 
+using envoy::service::ext_proc::v3alpha::ImmediateResponse;
 using envoy::service::ext_proc::v3alpha::ProcessingRequest;
 using envoy::service::ext_proc::v3alpha::ProcessingResponse;
 
@@ -66,8 +67,8 @@ void Filter::onReceiveMessage(
         }
       }
     } else if (response->has_immediate_response()) {
-      // To be implemented later. Leave stream open to allow people to implement
-      // correct servers that don't break us.
+      ENVOY_LOG(debug, "Returning immediate response from processor");
+      sendImmediateResponse(response->immediate_response());
       message_valid = true;
     }
     request_state_ = FilterState::IDLE;
@@ -129,6 +130,23 @@ void Filter::onGrpcClose() {
   }
 }
 
+void Filter::sendImmediateResponse(const ImmediateResponse& response) {
+  const auto status_code = response.has_status() ? response.status().code() : 200;
+  const auto grpc_status =
+      response.has_grpc_status()
+          ? absl::optional<Grpc::Status::GrpcStatus>(response.grpc_status().status())
+          : absl::nullopt;
+
+  decoder_callbacks_->sendLocalReply(
+      static_cast<Http::Code>(status_code), response.body(),
+      [&response](Http::ResponseHeaderMap& headers) {
+        if (response.has_headers()) {
+          MutationUtils::applyHeaderMutations(response.headers(), headers);
+        }
+      },
+      grpc_status, response.details());
+}
+
 } // namespace ExternalProcessing
 } // namespace HttpFilters
 } // namespace Extensions

source/extensions/filters/http/ext_proc/ext_proc.h

@@ -6,6 +6,7 @@
 #include "envoy/extensions/filters/http/ext_proc/v3alpha/ext_proc.pb.h"
 #include "envoy/grpc/async_client.h"
 #include "envoy/http/filter.h"
+#include "envoy/service/ext_proc/v3alpha/external_processor.pb.h"
 #include "envoy/stats/scope.h"
 #include "envoy/stats/stats_macros.h"
 
@@ -100,6 +101,7 @@ class Filter : public Logger::Loggable<Logger::Id::filter>,
 
 private:
   void closeStream();
+  void sendImmediateResponse(const envoy::service::ext_proc::v3alpha::ImmediateResponse& response);
 
   const FilterConfigSharedPtr config_;
   const ExternalProcessorClientPtr client_;

test/extensions/filters/http/ext_proc/ext_proc_integration_test.cc

@@ -59,6 +59,7 @@ class ExtProcIntegrationTest : public HttpIntegrationTest,
       ext_proc_filter.mutable_typed_config()->PackFrom(proto_config_);
       config_helper_.addFilter(MessageUtil::getJsonStringFromMessage(ext_proc_filter));
     });
+    setDownstreamProtocol(Http::CodecClient::Type::HTTP2);
   }
 
   void waitForFirstMessage(ProcessingRequest& request) {
@@ -77,7 +78,6 @@ INSTANTIATE_TEST_SUITE_P(IpVersionsClientType, ExtProcIntegrationTest,
 
 TEST_P(ExtProcIntegrationTest, GetAndCloseStream) {
   initializeConfig();
-  setDownstreamProtocol(Http::CodecClient::Type::HTTP2);
   HttpIntegrationTest::initialize();
 
   auto conn = makeClientConnection(lookupPort("http"));
@@ -111,7 +111,6 @@ TEST_P(ExtProcIntegrationTest, GetAndCloseStream) {
 
 TEST_P(ExtProcIntegrationTest, GetAndFailStream) {
   initializeConfig();
-  setDownstreamProtocol(Http::CodecClient::Type::HTTP2);
   HttpIntegrationTest::initialize();
 
   auto conn = makeClientConnection(lookupPort("http"));
@@ -132,7 +131,6 @@ TEST_P(ExtProcIntegrationTest, GetAndFailStream) {
 
 TEST_P(ExtProcIntegrationTest, GetAndSetHeaders) {
   initializeConfig();
-  setDownstreamProtocol(Http::CodecClient::Type::HTTP2);
   HttpIntegrationTest::initialize();
 
   auto conn = makeClientConnection(lookupPort("http"));
@@ -188,4 +186,44 @@ TEST_P(ExtProcIntegrationTest, GetAndSetHeaders) {
   EXPECT_EQ("200", response->headers().getStatusValue());
 }
 
+TEST_P(ExtProcIntegrationTest, GetAndRespondImmediately) {
+  initializeConfig();
+  HttpIntegrationTest::initialize();
+
+  auto conn = makeClientConnection(lookupPort("http"));
+  codec_client_ = makeHttpConnection(std::move(conn));
+  Http::TestRequestHeaderMapImpl headers;
+  HttpTestUtility::addDefaultHeaders(headers);
+  auto response = codec_client_->makeHeaderOnlyRequest(headers);
+
+  ProcessingRequest request_headers_msg;
+  waitForFirstMessage(request_headers_msg);
+
+  EXPECT_TRUE(request_headers_msg.has_request_headers());
+  processor_stream_->encodeHeaders(Http::TestResponseHeaderMapImpl{{":status", "200"}}, false);
+
+  // Produce an immediate response
+  ProcessingResponse response_msg;
+  auto* immediate_response = response_msg.mutable_immediate_response();
+  immediate_response->mutable_status()->set_code(envoy::type::v3::StatusCode::Unauthorized);
+  immediate_response->set_body("{\"reason\": \"Not authorized\"}");
+  immediate_response->set_details("Failed because you are not authorized");
+  auto* hdr1 = immediate_response->mutable_headers()->add_set_headers();
+  hdr1->mutable_header()->set_key("x-failure-reason");
+  hdr1->mutable_header()->set_value("testing");
+  auto* hdr2 = immediate_response->mutable_headers()->add_set_headers();
+  hdr2->mutable_header()->set_key("content-type");
+  hdr2->mutable_header()->set_value("application/json");
+  processor_stream_->sendGrpcMessage(response_msg);
+
+  response->waitForEndStream();
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("401", response->headers().getStatusValue());
+  EXPECT_EQ(
+      "testing",
+      response->headers().get(LowerCaseString("x-failure-reason"))[0]->value().getStringView());
+  EXPECT_EQ("application/json", response->headers().ContentType()->value().getStringView());
+  EXPECT_EQ("{\"reason\": \"Not authorized\"}", response->body());
+}
+
 } // namespace Envoy

test/extensions/filters/http/ext_proc/filter_test.cc

@@ -30,6 +30,7 @@ using Http::FilterTrailersStatus;
 using Http::LowerCaseString;
 
 using testing::Invoke;
+using testing::Unused;
 
 using namespace std::chrono_literals;
 
@@ -81,8 +82,8 @@ class HttpFilterTest : public testing::Test {
   NiceMock<Stats::MockIsolatedStatsStore> stats_store_;
   FilterConfigSharedPtr config_;
   std::unique_ptr<Filter> filter_;
-  NiceMock<Http::MockStreamDecoderFilterCallbacks> decoder_callbacks_;
-  NiceMock<Http::MockStreamEncoderFilterCallbacks> encoder_callbacks_;
+  Http::MockStreamDecoderFilterCallbacks decoder_callbacks_;
+  Http::MockStreamEncoderFilterCallbacks encoder_callbacks_;
   Http::TestRequestHeaderMapImpl request_headers_;
   Http::TestResponseHeaderMapImpl response_headers_;
   Http::TestRequestTrailerMapImpl request_trailers_;
@@ -185,22 +186,14 @@ TEST_F(HttpFilterTest, PostAndChangeHeaders) {
   stream_callbacks_->onReceiveMessage(std::move(resp1));
 
   // We should now have changed the original header a bit
-  request_headers_.iterate([](const Http::HeaderEntry& e) -> Http::HeaderMap::Iterate {
-    std::cerr << e.key().getStringView() << ": " << e.value().getStringView() << '\n';
-    return Http::HeaderMap::Iterate::Continue;
-  });
-  auto get1 = request_headers_.get(LowerCaseString("x-new-header"));
-  EXPECT_EQ(get1.size(), 1);
-  EXPECT_EQ(get1[0]->key(), "x-new-header");
-  EXPECT_EQ(get1[0]->value(), "new");
-  auto get2 = request_headers_.get(LowerCaseString("x-some-other-header"));
-  EXPECT_EQ(get2.size(), 2);
-  EXPECT_EQ(get2[0]->key(), "x-some-other-header");
-  EXPECT_EQ(get2[0]->value(), "yes");
-  EXPECT_EQ(get2[1]->key(), "x-some-other-header");
-  EXPECT_EQ(get2[1]->value(), "no");
-  auto get3 = request_headers_.get(LowerCaseString("x-do-we-want-this"));
-  EXPECT_TRUE(get3.empty());
+  Http::TestRequestHeaderMapImpl expected{{":path", "/"},
+                                          {":method", "POST"},
+                                          {":scheme", "http"},
+                                          {"host", "host"},
+                                          {"x-new-header", "new"},
+                                          {"x-some-other-header", "yes"},
+                                          {"x-some-other-header", "no"}};
+  EXPECT_TRUE(TestUtility::headerMapEqualIgnoreOrder(expected, request_headers_));
 
   data_.add("foo");
   EXPECT_EQ(FilterDataStatus::Continue, filter_->decodeData(data_, true));
@@ -233,19 +226,39 @@ TEST_F(HttpFilterTest, PostAndRespondImmediately) {
   EXPECT_EQ(FilterHeadersStatus::StopAllIterationAndWatermark,
             filter_->decodeHeaders(request_headers_, false));
 
+  Http::TestResponseHeaderMapImpl immediate_response_headers;
   EXPECT_CALL(decoder_callbacks_, continueDecoding());
+  EXPECT_CALL(decoder_callbacks_,
+              sendLocalReply(Http::Code::BadRequest, "Bad request", _, _, "Got a bad request"))
+      .WillOnce(Invoke([&immediate_response_headers](
+                           Unused, Unused,
+                           std::function<void(Http::ResponseHeaderMap & headers)> modify_headers,
+                           Unused, Unused) { modify_headers(immediate_response_headers); }));
   std::unique_ptr<ProcessingResponse> resp1 = std::make_unique<ProcessingResponse>();
   auto* immediate_response = resp1->mutable_immediate_response();
   immediate_response->mutable_status()->set_code(envoy::type::v3::StatusCode::BadRequest);
   immediate_response->set_body("Bad request");
   immediate_response->set_details("Got a bad request");
+  auto* immediate_headers = immediate_response->mutable_headers();
+  auto* hdr1 = immediate_headers->add_set_headers();
+  hdr1->mutable_header()->set_key("content-type");
+  hdr1->mutable_header()->set_value("text/plain");
+  auto* hdr2 = immediate_headers->add_set_headers();
+  hdr2->mutable_append()->set_value(true);
+  hdr2->mutable_header()->set_key("x-another-thing");
+  hdr2->mutable_header()->set_value("1");
+  auto* hdr3 = immediate_headers->add_set_headers();
+  hdr3->mutable_append()->set_value(true);
+  hdr3->mutable_header()->set_key("x-another-thing");
+  hdr3->mutable_header()->set_value("2");
   stream_callbacks_->onReceiveMessage(std::move(resp1));
-
-  // Immediate response processing not yet implemented -- all we can expect
-  // at this point is that continueDecoding is called and that the
-  // stream is not yet closed.
   EXPECT_FALSE(stream_close_sent_);
 
+  Http::TestResponseHeaderMapImpl expected_response_headers{
+      {"content-type", "text/plain"}, {"x-another-thing", "1"}, {"x-another-thing", "2"}};
+  EXPECT_TRUE(TestUtility::headerMapEqualIgnoreOrder(expected_response_headers,
+                                                     immediate_response_headers));
+
   data_.add("foo");
   EXPECT_EQ(FilterDataStatus::Continue, filter_->decodeData(data_, true));
   EXPECT_EQ(FilterTrailersStatus::Continue, filter_->decodeTrailers(request_trailers_));
@@ -282,6 +295,8 @@ TEST_F(HttpFilterTest, PostAndFail) {
 
   // Oh no! The remote server had a failure!
   EXPECT_CALL(decoder_callbacks_, sendLocalReply(Http::Code::InternalServerError, _, _, _, _));
+  // In this case, this call includes header encoding
+  EXPECT_CALL(decoder_callbacks_, encodeHeaders_(_, _));
   stream_callbacks_->onGrpcError(Grpc::Status::Internal);
 
   data_.add("foo");