[bp/1.37] dym: fix a bug where part of body maynot be sent to upstream (#44081) (#44083)

Commit Message: dym: fix a bug where part of body maynot be sent to
upstream
Additional Description:

I have provided two different ABIs (buffered body and received body) to
access buffered or latest received body. The the previous legacy
addDecodedData and addEncodedData have been unnecessary.

Risk Level: low.
Testing: unit.
Docs Changes: n/a.
Release Notes: added.
Platform Specific Features: n/a.
[Optional Fixes #Issue] To close
#43063

---------

Signed-off-by: wbpcode/wangbaiping <wbphub@gmail.com>

Author: wbpcode

changelogs/current.yaml

@@ -12,6 +12,11 @@ bug_fixes:
   change: |
     Fixed segfault from timer thread-safety violation, a ring buffer overflow and alpha calculation in
     peak_ewma loadbalancer.
+- area: dynamic_modules
+  change: |
+    Fixed a bug where dynamic module filter may result in a incomplete body being sent to upstream
+    or downstream when some filters before or after the dynamic module filter in the chain
+    buffered the body and the dynamic module filter did not.
 
 removed_config_or_runtime:
 # *Normally occurs at the end of the* :ref:`deprecation period <deprecated>`

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

@@ -81,11 +81,6 @@ FilterHeadersStatus DynamicModuleHttpFilter::decodeHeaders(RequestHeaderMap&, bo
 };
 
 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);
@@ -129,11 +124,6 @@ FilterDataStatus DynamicModuleHttpFilter::encodeData(Buffer::Instance& chunk, bo
   if (sent_local_reply_) { // See the comment on the flag.
     return FilterDataStatus::Continue;
   }
-  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);

test/extensions/dynamic_modules/http/filter_test.cc

@@ -440,6 +440,62 @@ TEST_P(DynamicModuleHttpLanguageTests, FilterStateCallbacks) {
   EXPECT_EQ(stream_complete_value->serializeAsString(), "stream_complete_value");
 }
 
+TEST_P(DynamicModuleTestLanguages, WillNotMoveDataAutomatically) {
+  const std::string filter_name = "foo";
+  const std::string filter_config = "bar";
+
+  const auto language = GetParam();
+  auto dynamic_module = newDynamicModule(testSharedObjectPath("no_op", language), false);
+  EXPECT_TRUE(dynamic_module.ok());
+
+  NiceMock<Server::Configuration::MockServerFactoryContext> context;
+  Stats::IsolatedStoreImpl stats_store;
+  auto filter_config_or_status =
+      Envoy::Extensions::DynamicModules::HttpFilters::newDynamicModuleHttpFilterConfig(
+          filter_name, filter_config, false, std::move(dynamic_module.value()),
+          *stats_store.createScope(""), context);
+  EXPECT_TRUE(filter_config_or_status.ok());
+
+  auto filter = std::make_shared<DynamicModuleHttpFilter>(filter_config_or_status.value(),
+                                                          stats_store.symbolTable());
+  filter->initializeInModuleFilter();
+
+  NiceMock<Http::MockStreamDecoderFilterCallbacks> decoder_callbacks;
+  filter->setDecoderFilterCallbacks(decoder_callbacks);
+  NiceMock<Http::MockStreamEncoderFilterCallbacks> encoder_callbacks;
+  filter->setEncoderFilterCallbacks(encoder_callbacks);
+
+  TestRequestHeaderMapImpl headers{{}};
+  EXPECT_EQ(FilterHeadersStatus::Continue, filter->decodeHeaders(headers, false));
+
+  Buffer::OwnedImpl buffered_request_data("buffered data");
+  EXPECT_CALL(decoder_callbacks, addDecodedData(_, _)).Times(0); // should not be called.
+
+  Buffer::OwnedImpl new_request_data1("new data 1"); // 10 bytes
+  EXPECT_EQ(FilterDataStatus::Continue, filter->decodeData(new_request_data1, false));
+  EXPECT_EQ(10U, new_request_data1.length());
+  Buffer::OwnedImpl new_request_data2("new data 2"); // 10 bytes
+  EXPECT_EQ(FilterDataStatus::Continue, filter->decodeData(new_request_data2, true));
+  EXPECT_EQ(10U, new_request_data2.length());
+
+  // Complete response lifecycle (needed for Rust no_op module lifecycle assertions).
+  TestResponseHeaderMapImpl response_headers{{}};
+  EXPECT_EQ(FilterHeadersStatus::Continue, filter->encodeHeaders(response_headers, false));
+
+  Buffer::OwnedImpl buffered_response_data("buffered data");
+  EXPECT_CALL(encoder_callbacks, addEncodedData(_, _)).Times(0); // should not be called.
+
+  Buffer::OwnedImpl new_response_data1("new data 1"); // 10 bytes
+  EXPECT_EQ(FilterDataStatus::Continue, filter->encodeData(new_response_data1, false));
+  EXPECT_EQ(10U, new_response_data1.length());
+  Buffer::OwnedImpl new_response_data2("new data 2"); // 10 bytes
+  EXPECT_EQ(FilterDataStatus::Continue, filter->encodeData(new_response_data2, true));
+  EXPECT_EQ(10U, new_response_data2.length());
+
+  filter->onStreamComplete();
+  filter->onDestroy();
+}
+
 TEST_P(DynamicModuleHttpLanguageTests, BodyCallbacks) {
   const std::string filter_name = "body_callbacks";
   const std::string filter_config = "";
@@ -469,12 +525,8 @@ TEST_P(DynamicModuleHttpLanguageTests, 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/test_data/rust/no_op.rs

@@ -94,11 +94,7 @@ struct NopHttpFilter {
 impl Drop for NopHttpFilter {
   fn drop(&mut self) {
     assert!(self.on_request_headers_called);
-    assert!(self.on_request_body_called);
-    assert!(self.on_request_trailers_called);
     assert!(self.on_response_headers_called);
-    assert!(self.on_response_body_called);
-    assert!(self.on_response_trailers_called);
     assert!(self.on_stream_complete_called);
   }
 }