dynamic_modules: adds integration tests

source/extensions/filters/http/dynamic_modules/abi_impl.cc

@@ -375,7 +375,11 @@ bool envoy_dynamic_module_callback_http_get_request_body_vector(
   auto filter = static_cast<DynamicModuleHttpFilter*>(filter_envoy_ptr);
   auto buffer = filter->decoder_callbacks_->decodingBuffer();
   if (!buffer) {
-    return false;
+    buffer = filter->current_request_body_;
+    if (!buffer) {
+      return false;
+    }
+    // See the comment on current_request_body_ for when we reach this.
   }
   auto raw_slices = buffer->getRawSlices(std::nullopt);
   auto counter = 0;
@@ -392,7 +396,11 @@ bool envoy_dynamic_module_callback_http_get_request_body_vector_size(
   auto filter = static_cast<DynamicModuleHttpFilter*>(filter_envoy_ptr);
   auto buffer = filter->decoder_callbacks_->decodingBuffer();
   if (!buffer) {
-    return false;
+    buffer = filter->current_request_body_;
+    if (!buffer) {
+      return false;
+    }
+    // See the comment on current_request_body_ for when we reach this line.
   }
   *size = buffer->getRawSlices(std::nullopt).size();
   return true;
@@ -403,6 +411,11 @@ bool envoy_dynamic_module_callback_http_append_request_body(
     envoy_dynamic_module_type_buffer_module_ptr data, size_t length) {
   auto filter = static_cast<DynamicModuleHttpFilter*>(filter_envoy_ptr);
   if (!filter->decoder_callbacks_->decodingBuffer()) {
+    if (filter->current_request_body_) { // See the comment on current_request_body_ for when we
+                                         // enter this block.
+      filter->current_request_body_->add(absl::string_view(static_cast<const char*>(data), length));
+      return true;
+    }
     return false;
   }
   filter->decoder_callbacks_->modifyDecodingBuffer([data, length](Buffer::Instance& buffer) {
@@ -415,6 +428,12 @@ bool envoy_dynamic_module_callback_http_drain_request_body(
     envoy_dynamic_module_type_http_filter_envoy_ptr filter_envoy_ptr, size_t number_of_bytes) {
   auto filter = static_cast<DynamicModuleHttpFilter*>(filter_envoy_ptr);
   if (!filter->decoder_callbacks_->decodingBuffer()) {
+    if (filter->current_request_body_) { // See the comment on current_request_body_ for when we
+                                         // enter this block.
+      auto size = std::min(filter->current_request_body_->length(), number_of_bytes);
+      filter->current_request_body_->drain(size);
+      return true;
+    }
     return false;
   }
 
@@ -431,7 +450,11 @@ bool envoy_dynamic_module_callback_http_get_response_body_vector(
   auto filter = static_cast<DynamicModuleHttpFilter*>(filter_envoy_ptr);
   auto buffer = filter->encoder_callbacks_->encodingBuffer();
   if (!buffer) {
-    return false;
+    buffer = filter->current_response_body_;
+    if (!buffer) {
+      return false;
+    }
+    // See the comment on current_response_body_ for when we reach this line.
   }
   auto raw_slices = buffer->getRawSlices(std::nullopt);
   auto counter = 0;
@@ -448,7 +471,11 @@ bool envoy_dynamic_module_callback_http_get_response_body_vector_size(
   auto filter = static_cast<DynamicModuleHttpFilter*>(filter_envoy_ptr);
   auto buffer = filter->encoder_callbacks_->encodingBuffer();
   if (!buffer) {
-    return false;
+    buffer = filter->current_response_body_;
+    if (!buffer) {
+      return false;
+    }
+    // See the comment on current_response_body_ for when we reach this line.
   }
   *size = buffer->getRawSlices(std::nullopt).size();
   return true;
@@ -459,6 +486,12 @@ bool envoy_dynamic_module_callback_http_append_response_body(
     envoy_dynamic_module_type_buffer_module_ptr data, size_t length) {
   auto filter = static_cast<DynamicModuleHttpFilter*>(filter_envoy_ptr);
   if (!filter->encoder_callbacks_->encodingBuffer()) {
+    if (filter->current_response_body_) { // See the comment on current_response_body_ for when we
+                                          // enter this block.
+      filter->current_response_body_->add(
+          absl::string_view(static_cast<const char*>(data), length));
+      return true;
+    }
     return false;
   }
   filter->encoder_callbacks_->modifyEncodingBuffer([data, length](Buffer::Instance& buffer) {
@@ -471,6 +504,12 @@ bool envoy_dynamic_module_callback_http_drain_response_body(
     envoy_dynamic_module_type_http_filter_envoy_ptr filter_envoy_ptr, size_t number_of_bytes) {
   auto filter = static_cast<DynamicModuleHttpFilter*>(filter_envoy_ptr);
   if (!filter->encoder_callbacks_->encodingBuffer()) {
+    if (filter->current_response_body_) { // See the comment on current_response_body_ for when we
+                                          // enter this block.
+      auto size = std::min(filter->current_response_body_->length(), number_of_bytes);
+      filter->current_response_body_->drain(size);
+      return true;
+    }
     return false;
   }
 

source/extensions/filters/http/dynamic_modules/filter.cc

@@ -31,9 +31,16 @@ FilterHeadersStatus DynamicModuleHttpFilter::decodeHeaders(RequestHeaderMap& hea
   return static_cast<FilterHeadersStatus>(status);
 };
 
-FilterDataStatus DynamicModuleHttpFilter::decodeData(Buffer::Instance&, bool end_of_stream) {
+FilterDataStatus DynamicModuleHttpFilter::decodeData(Buffer::Instance& chunk, bool end_of_stream) {
+  if (end_of_stream && decoder_callbacks_->decodingBuffer()) {
+    // To make the very last chunk of the body available to the filter when buffering is enabled,
+    // we need to call addDecodedData. See the code comment there for more details.
+    decoder_callbacks_->addDecodedData(chunk, false);
+  }
+  current_request_body_ = &chunk;
   const envoy_dynamic_module_type_on_http_filter_request_body_status status =
       config_->on_http_filter_request_body_(thisAsVoidPtr(), in_module_filter_, end_of_stream);
+  current_request_body_ = nullptr;
   return static_cast<FilterDataStatus>(status);
 };
 
@@ -62,9 +69,16 @@ FilterHeadersStatus DynamicModuleHttpFilter::encodeHeaders(ResponseHeaderMap& he
   return static_cast<FilterHeadersStatus>(status);
 };
 
-FilterDataStatus DynamicModuleHttpFilter::encodeData(Buffer::Instance&, bool end_of_stream) {
+FilterDataStatus DynamicModuleHttpFilter::encodeData(Buffer::Instance& chunk, bool end_of_stream) {
+  if (end_of_stream && encoder_callbacks_->encodingBuffer()) {
+    // To make the very last chunk of the body available to the filter when buffering is enabled,
+    // we need to call addEncodedData. See the code comment there for more details.
+    encoder_callbacks_->addEncodedData(chunk, false);
+  }
+  current_response_body_ = &chunk;
   const envoy_dynamic_module_type_on_http_filter_response_body_status status =
       config_->on_http_filter_response_body_(thisAsVoidPtr(), in_module_filter_, end_of_stream);
+  current_response_body_ = nullptr;
   return static_cast<FilterDataStatus>(status);
 };
 

source/extensions/filters/http/dynamic_modules/filter.h

@@ -64,6 +64,12 @@ class DynamicModuleHttpFilter : public Http::StreamFilter,
   ResponseHeaderMap* response_headers_ = nullptr;
   ResponseTrailerMap* response_trailers_ = nullptr;
 
+  // These are used to hold the current chunk of the request/response body during the decodeData and
+  // encodeData callbacks. It is only valid during the call and should not be used outside of the
+  // call.
+  Buffer::Instance* current_request_body_ = nullptr;
+  Buffer::Instance* current_response_body_ = nullptr;
+
   /**
    * Helper to get the downstream information of the stream.
    */

test/extensions/dynamic_modules/http/BUILD

@@ -66,7 +66,7 @@ envoy_cc_test(
     name = "integration_test",
     srcs = ["integration_test.cc"],
     data = [
-        "//test/extensions/dynamic_modules/test_data/rust:http",
+        "//test/extensions/dynamic_modules/test_data/rust:http_integration_test",
     ],
     rbe_pool = "6gig",
     deps = [

test/extensions/dynamic_modules/http/filter_test.cc

@@ -168,8 +168,12 @@ TEST(DynamiModulesTest, BodyCallbacks) {
   filter->setEncoderFilterCallbacks(encoder_callbacks);
   Buffer::OwnedImpl request_body;
   EXPECT_CALL(decoder_callbacks, decodingBuffer()).WillRepeatedly(testing::Return(&request_body));
+  EXPECT_CALL(decoder_callbacks, addDecodedData(_, _))
+      .WillOnce(Invoke([&](Buffer::Instance&, bool) -> void {}));
   Buffer::OwnedImpl response_body;
   EXPECT_CALL(encoder_callbacks, encodingBuffer()).WillRepeatedly(testing::Return(&response_body));
+  EXPECT_CALL(encoder_callbacks, addEncodedData(_, _))
+      .WillOnce(Invoke([&](Buffer::Instance&, bool) -> void {}));
   EXPECT_CALL(decoder_callbacks, modifyDecodingBuffer(_))
       .WillRepeatedly(Invoke([&](std::function<void(Buffer::Instance&)> callback) -> void {
         callback(request_body);

test/extensions/dynamic_modules/http/integration_test.cc

@@ -6,19 +6,26 @@ class DynamicModulesIntegrationTest : public testing::TestWithParam<Network::Add
 public:
   DynamicModulesIntegrationTest() : HttpIntegrationTest(Http::CodecType::HTTP2, GetParam()){};
 
-  void initializeFilter(const std::string& module_name, const std::string& filter_name,
-                        const std::string& config = "") {
+  void initializeFilter(const std::string& filter_name, const std::string& config = "") {
+    TestEnvironment::setEnvVar(
+        "ENVOY_DYNAMIC_MODULES_SEARCH_PATH",
+        TestEnvironment::substitute(
+            "{{ test_rundir }}/test/extensions/dynamic_modules/test_data/rust"),
+        1);
+
     constexpr auto filter_config = R"EOF(
 name: envoy.extensions.filters.http.dynamic_modules
 typed_config:
   "@type": type.googleapis.com/envoy.extensions.filters.http.dynamic_modules.v3.DynamicModuleFilter
   dynamic_module_config:
-    name: {}
+    name: http_integration_test
   filter_name: {}
   filter_config: {}
 )EOF";
 
-    config_helper_.prependFilter(fmt::format(filter_config, module_name, filter_name, config));
+    config_helper_.addConfigModifier(setEnableDownstreamTrailersHttp1());
+    config_helper_.addConfigModifier(setEnableUpstreamTrailersHttp1());
+    config_helper_.prependFilter(fmt::format(filter_config, filter_name, config));
     initialize();
   }
 };
@@ -27,14 +34,9 @@ INSTANTIATE_TEST_SUITE_P(IpVersions, DynamicModulesIntegrationTest,
                          testing::ValuesIn(TestEnvironment::getIpVersionsForTest()),
                          TestUtility::ipTestParamsToString);
 
-TEST_P(DynamicModulesIntegrationTest, Nop) {
-  TestEnvironment::setEnvVar(
-      "ENVOY_DYNAMIC_MODULES_SEARCH_PATH",
-      TestEnvironment::substitute(
-          "{{ test_rundir }}/test/extensions/dynamic_modules/test_data/rust"),
-      1);
+TEST_P(DynamicModulesIntegrationTest, PassThrough) {
+  initializeFilter("passthrough");
 
-  initializeFilter("http", "passthrough");
   // Create a client aimed at Envoy’s default HTTP port.
   codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));
 
@@ -56,4 +58,142 @@ TEST_P(DynamicModulesIntegrationTest, Nop) {
   EXPECT_EQ(10U, response->body().size());
 }
 
+TEST_P(DynamicModulesIntegrationTest, HeaderCallbacks) {
+  initializeFilter("header_callbacks", "dog:cat");
+  codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));
+
+  Http::TestRequestHeaderMapImpl request_headers{{"foo", "bar"},
+                                                 {":method", "POST"},
+                                                 {":path", "/test/long/url"},
+                                                 {":scheme", "http"},
+                                                 {":authority", "host"}};
+  Http::TestRequestTrailerMapImpl request_trailers{{"foo", "bar"}};
+  Http::TestResponseHeaderMapImpl response_headers{{":status", "200"}, {"foo", "bar"}};
+  Http::TestResponseTrailerMapImpl response_trailers{{"foo", "bar"}};
+
+  auto encoder_decoder = codec_client_->startRequest(request_headers);
+  auto response = std::move(encoder_decoder.second);
+  codec_client_->sendData(encoder_decoder.first, 10, false);
+  codec_client_->sendTrailers(encoder_decoder.first, request_trailers);
+
+  waitForNextUpstreamRequest();
+  upstream_request_->encodeHeaders(response_headers, false);
+  upstream_request_->encodeData(10, false);
+  upstream_request_->encodeTrailers(response_trailers);
+
+  ASSERT_TRUE(response->waitForEndStream());
+
+  // Verify the proxied request was received upstream, as expected.
+  EXPECT_TRUE(upstream_request_->complete());
+  EXPECT_EQ(10U, upstream_request_->bodyLength());
+  // Verify that the headers/trailers are added as expected.
+  EXPECT_EQ(
+      "cat",
+      upstream_request_->headers().get(Http::LowerCaseString("dog"))[0]->value().getStringView());
+  EXPECT_EQ("cat", upstream_request_->trailers()
+                       .get()
+                       ->get(Http::LowerCaseString("dog"))[0]
+                       ->value()
+                       .getStringView());
+  // Verify the proxied response was received downstream, as expected.
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().Status()->value().getStringView());
+  EXPECT_EQ(10U, response->body().size());
+  // Verify that the headers/trailers are added as expected.
+  EXPECT_EQ("cat",
+            response->headers().get(Http::LowerCaseString("dog"))[0]->value().getStringView());
+  EXPECT_EQ(
+      "cat",
+      response->trailers().get()->get(Http::LowerCaseString("dog"))[0]->value().getStringView());
+}
+
+TEST_P(DynamicModulesIntegrationTest, BodyCallbacks) {
+  initializeFilter("body_callbacks");
+  codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));
+
+  Http::TestRequestHeaderMapImpl request_headers{{":method", "POST"},
+                                                 {":path", "/test/long/url"},
+                                                 {":scheme", "http"},
+                                                 {":authority", "test.com"}};
+  auto encoder_decoder = codec_client_->startRequest(request_headers, false);
+  auto response = std::move(encoder_decoder.second);
+  codec_client_->sendData(encoder_decoder.first, "request", false);
+  codec_client_->sendData(encoder_decoder.first, "_b", false);
+  codec_client_->sendData(encoder_decoder.first, "ody", true);
+
+  waitForNextUpstreamRequest();
+  upstream_request_->encodeHeaders(default_response_headers_, false);
+  upstream_request_->encodeData("res", false);
+  upstream_request_->encodeData("ponse", false);
+  upstream_request_->encodeData("_body", true);
+
+  ASSERT_TRUE(response->waitForEndStream());
+
+  // Verify the proxied request was received upstream, as expected.
+  EXPECT_TRUE(upstream_request_->complete());
+  EXPECT_EQ("new_request_body", upstream_request_->body().toString());
+  // Verify the proxied response was received downstream, as expected.
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().Status()->value().getStringView());
+  EXPECT_EQ("new_response_body", response->body());
+}
+
+TEST_P(DynamicModulesIntegrationTest, BodyCallbacks_WithoutBuffering) {
+  initializeFilter("body_callbacks", "immediate_end_of_stream");
+  codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));
+
+  auto response = codec_client_->makeRequestWithBody(default_request_headers_, "request_body");
+
+  waitForNextUpstreamRequest();
+  upstream_request_->encodeHeaders(default_response_headers_, false);
+  upstream_request_->encodeData("response_body", true);
+
+  ASSERT_TRUE(response->waitForEndStream());
+
+  // Verify the proxied request was received upstream, as expected.
+  EXPECT_TRUE(upstream_request_->complete());
+  EXPECT_EQ("new_request_body", upstream_request_->body().toString());
+  // Verify the proxied response was received downstream, as expected.
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().Status()->value().getStringView());
+  EXPECT_EQ("new_response_body", response->body());
+}
+
+TEST_P(DynamicModulesIntegrationTest, SendResponseFromOnRequestHeaders) {
+  initializeFilter("send_response", "on_request_headers");
+  codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));
+
+  auto encoder_decoder = codec_client_->startRequest(default_request_headers_);
+  auto response = std::move(encoder_decoder.second);
+
+  ASSERT_TRUE(response->waitForEndStream());
+
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().Status()->value().getStringView());
+  auto body = response->body();
+  EXPECT_EQ("local_response_body_from_on_request_headers", body);
+  EXPECT_EQ(
+      "some_value",
+      response->headers().get(Http::LowerCaseString("some_header"))[0]->value().getStringView());
+}
+
+TEST_P(DynamicModulesIntegrationTest, SendResponseFromOnRequestBody) {
+  initializeFilter("send_response", "on_request_body");
+  codec_client_ = makeHttpConnection(makeClientConnection((lookupPort("http"))));
+
+  auto encoder_decoder = codec_client_->startRequest(default_request_headers_);
+  auto response = std::move(encoder_decoder.second);
+  codec_client_->sendData(encoder_decoder.first, 10, true);
+
+  ASSERT_TRUE(response->waitForEndStream());
+
+  EXPECT_TRUE(response->complete());
+  EXPECT_EQ("200", response->headers().Status()->value().getStringView());
+  auto body = response->body();
+  EXPECT_EQ("local_response_body_from_on_request_body", body);
+  EXPECT_EQ(
+      "some_value",
+      response->headers().get(Http::LowerCaseString("some_header"))[0]->value().getStringView());
+}
+
 } // namespace Envoy

test/extensions/dynamic_modules/test_data/rust/BUILD

@@ -16,3 +16,5 @@ test_program(name = "program_init_fail")
 test_program(name = "abi_version_mismatch")
 
 test_program(name = "http")
+
+test_program(name = "http_integration_test")

test/extensions/dynamic_modules/test_data/rust/Cargo.toml

@@ -37,3 +37,9 @@ name = "http"
 path = "http.rs"
 crate-type = ["cdylib"]
 test = true
+
+[[example]]
+name = "http_integration_test"
+path = "http_integration_test.rs"
+crate-type = ["cdylib"]
+test = true