Merge dee70d1 into d5834c1

changelogs/current.yaml

@@ -219,6 +219,10 @@ bug_fixes:
 - area: tracing
   change: |
     Fixed a bug where the OpenTelemetry tracer exports the OTLP request even when no spans are present.
+- area: wasm
+  change: |
+    Fixed a bug where a body received in chunks is not correctly dumped, resulting in
+    an incomplete dump and loss of the last chunk. This issue is manifested in HTTP/2.
 - area: dynamic_forward_proxy
   change: |
     Fixed a bug where DFP sub-cluster gets removed due to CDS update and doesn't gets recreated. This behavior can be reverted by

source/extensions/common/wasm/context.cc

@@ -1723,6 +1723,9 @@ Http::FilterDataStatus Context::decodeData(::Envoy::Buffer::Instance& data, bool
   if (!in_vm_context_created_) {
     return Http::FilterDataStatus::Continue;
   }
+  if (buffering_request_body_) {
+    decoder_callbacks_->addDecodedData(data, false);
+  }
   request_body_buffer_ = &data;
   end_of_stream_ = end_stream;
   const auto buffer = getBuffer(WasmBufferType::HttpRequestBody);
@@ -1793,6 +1796,9 @@ Http::FilterDataStatus Context::encodeData(::Envoy::Buffer::Instance& data, bool
   if (!in_vm_context_created_) {
     return Http::FilterDataStatus::Continue;
   }
+  if (buffering_response_body_) {
+    encoder_callbacks_->addEncodedData(data, false);
+  }
   response_body_buffer_ = &data;
   end_of_stream_ = end_stream;
   const auto buffer = getBuffer(WasmBufferType::HttpResponseBody);
@@ -1801,7 +1807,7 @@ Http::FilterDataStatus Context::encodeData(::Envoy::Buffer::Instance& data, bool
   buffering_response_body_ = false;
   switch (result) {
   case Http::FilterDataStatus::Continue:
-    request_body_buffer_ = nullptr;
+    response_body_buffer_ = nullptr;
     break;
   case Http::FilterDataStatus::StopIterationAndBuffer:
     buffering_response_body_ = true;

test/extensions/common/wasm/BUILD

@@ -98,6 +98,7 @@ envoy_cc_test_library(
     hdrs = ["wasm_runtime.h"],
     tags = ["skip_on_windows"],
     deps = [
+        "//envoy/http:codec_interface",
         "//source/extensions/wasm_runtime/null:config",
         "//source/extensions/wasm_runtime/v8:config",
         "//source/extensions/wasm_runtime/wamr:config",

test/extensions/common/wasm/wasm_runtime.cc

@@ -55,15 +55,51 @@ wasmDualFilterTestMatrix(bool include_nullvm, bool cpp_only) {
   return values;
 }
 
+std::vector<std::tuple<std::string, std::string, bool, Http::CodecType>>
+wasmDualFilterWithCodecsTestMatrix(bool include_nullvm, bool cpp_only,
+                                   std::vector<Http::CodecType> codecs_type) {
+  std::vector<std::tuple<std::string, std::string, bool, Http::CodecType>> values;
+  for (const Http::CodecType codec_type : codecs_type) {
+    for (const auto& [runtime, language] :
+         Envoy::Extensions::Common::Wasm::wasmTestMatrix(include_nullvm, cpp_only)) {
+      values.push_back(std::make_tuple(runtime, language, true, codec_type));
+      values.push_back(std::make_tuple(runtime, language, false, codec_type));
+    }
+  }
+  return values;
+}
+
 std::string
 wasmTestParamsToString(const ::testing::TestParamInfo<std::tuple<std::string, std::string>>& p) {
   return std::get<0>(p.param) + "_" + std::get<1>(p.param);
 }
 
 std::string wasmDualFilterTestParamsToString(
     const ::testing::TestParamInfo<std::tuple<std::string, std::string, bool>>& p) {
-  return (std::get<2>(p.param) ? "downstream_" : "upstream_") + std::get<0>(p.param) + "_" +
-         std::get<1>(p.param);
+  auto [runtime, language, direction] = p.param;
+  return fmt::format("{}_{}_{}", direction ? "downstream" : "upstream", runtime, language);
+}
+
+static std::string codecToString(const Http::CodecType& e) {
+  switch (e) {
+  case Http::CodecType::HTTP1:
+    return "http1";
+  case Http::CodecType::HTTP2:
+    return "http2";
+  case Http::CodecType::HTTP3:
+    return "http3";
+  default:
+    break;
+  }
+  return "Unknown";
+}
+
+std::string wasmDualFilterWithCodecsTestParamsToString(
+    const ::testing::TestParamInfo<std::tuple<std::string, std::string, bool, Http::CodecType>>&
+        p) {
+  auto [runtime, language, direction, codec_type] = p.param;
+  return fmt::format("{}_{}_{}_{}", direction ? "downstream" : "upstream", runtime, language,
+                     codecToString(codec_type));
 }
 
 } // namespace Wasm

test/extensions/common/wasm/wasm_runtime.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include "envoy/http/codec.h"
+
 #include "gtest/gtest.h"
 
 namespace Envoy {
@@ -14,17 +16,26 @@ std::vector<std::tuple<std::string, std::string>> wasmTestMatrix(bool include_nu
 std::vector<std::tuple<std::string, std::string, bool>>
 wasmDualFilterTestMatrix(bool include_nullvm, bool cpp_only);
 
+std::vector<std::tuple<std::string, std::string, bool, Http::CodecType>>
+wasmDualFilterWithCodecsTestMatrix(bool include_nullvm, bool cpp_only,
+                                   std::vector<Http::CodecType> codecs_type);
+
 inline auto runtime_and_cpp_values = testing::ValuesIn(wasmTestMatrix(true, true));
 inline auto sandbox_runtime_and_cpp_values = testing::ValuesIn(wasmTestMatrix(false, true));
 inline auto dual_filter_sandbox_runtime_and_cpp_values =
     testing::ValuesIn(wasmDualFilterTestMatrix(false, true));
+inline auto dual_filter_with_codecs_sandbox_runtime_and_cpp_values =
+    testing::ValuesIn(wasmDualFilterWithCodecsTestMatrix(
+        false, true, {Envoy::Http::CodecType::HTTP1, Envoy::Http::CodecType::HTTP2}));
 inline auto runtime_and_language_values = testing::ValuesIn(wasmTestMatrix(true, false));
 inline auto sandbox_runtime_and_language_values = testing::ValuesIn(wasmTestMatrix(false, false));
 
 std::string
 wasmTestParamsToString(const ::testing::TestParamInfo<std::tuple<std::string, std::string>>& p);
 std::string wasmDualFilterTestParamsToString(
     const ::testing::TestParamInfo<std::tuple<std::string, std::string, bool>>& p);
+std::string wasmDualFilterWithCodecsTestParamsToString(
+    const ::testing::TestParamInfo<std::tuple<std::string, std::string, bool, Http::CodecType>>& p);
 
 } // namespace Wasm
 } // namespace Common

test/extensions/filters/http/wasm/test_data/test_body_cpp.cc

@@ -118,6 +118,15 @@ FilterDataStatus BodyContext::onBody(WasmBufferType type, size_t buffer_length,
     }
     return FilterDataStatus::StopIterationAndBuffer;
 
+  } else if (body_op_ == "SetEndOfBodies") {
+    logBody(type);
+    if (end_of_stream) {
+      getBufferStatus(type, &size, &flags);
+      setBuffer(type, size, 0, ".end");
+      return FilterDataStatus::Continue;
+    }
+    return FilterDataStatus::StopIterationAndBuffer;
+
   } else {
     // This is a test and the test was configured incorrectly.
     logError("Invalid body test op " + body_op_);

test/extensions/filters/http/wasm/wasm_filter_integration_test.cc

@@ -12,13 +12,20 @@ namespace {
 
 class WasmFilterIntegrationTest
     : public HttpIntegrationTest,
-      public testing::TestWithParam<std::tuple<std::string, std::string, bool>> {
+      public testing::TestWithParam<std::tuple<std::string, std::string, bool, Http::CodecType>> {
 public:
   WasmFilterIntegrationTest()
-      : HttpIntegrationTest(Http::CodecType::HTTP1, Network::Address::IpVersion::v4) {}
+      : HttpIntegrationTest(std::get<3>(GetParam()), Network::Address::IpVersion::v4) {}
 
   void SetUp() override {
-    setUpstreamProtocol(Http::CodecType::HTTP1);
+    setUpstreamProtocol(std::get<3>(GetParam()));
+    if (std::get<3>(GetParam()) == Http::CodecType::HTTP2) {
+      config_helper_.setClientCodec(envoy::extensions::filters::network::http_connection_manager::
+                                        v3::HttpConnectionManager::HTTP2);
+    } else {
+      config_helper_.setClientCodec(envoy::extensions::filters::network::http_connection_manager::
+                                        v3::HttpConnectionManager::HTTP1);
+    }
     // Wasm filters are expensive to setup and sometime default is not enough,
     // It needs to increase timeout to avoid flaky tests
     setListenersBoundTimeout(10 * TestUtility::DefaultTimeout);
@@ -68,11 +75,12 @@ class WasmFilterIntegrationTest
     });
   }
 
-  void runTest(const Http::RequestHeaderMap& request_headers, const std::string& request_body,
+  void runTest(const Http::RequestHeaderMap& request_headers,
+               const std::vector<std::string>& request_body,
                const Http::RequestHeaderMap& expected_request_headers,
                const std::string& expected_request_body,
                const Http::ResponseHeaderMap& upstream_response_headers,
-               const std::string& upstream_response_body,
+               const std::vector<std::string>& upstream_response_body,
                const Http::ResponseHeaderMap& expected_response_headers,
                const std::string& expected_response_body) {
 
@@ -84,8 +92,11 @@ class WasmFilterIntegrationTest
       auto encoder_decoder = codec_client_->startRequest(request_headers);
       request_encoder_ = &encoder_decoder.first;
       response = std::move(encoder_decoder.second);
-      Buffer::OwnedImpl buffer(request_body);
-      codec_client_->sendData(*request_encoder_, buffer, true);
+      const auto request_body_size = request_body.size();
+      for (size_t n = 0; n < request_body_size; n++) {
+        Buffer::OwnedImpl buffer(request_body[n]);
+        codec_client_->sendData(*request_encoder_, buffer, n == request_body_size - 1);
+      }
     }
 
     ASSERT_TRUE(fake_upstreams_[0]->waitForHttpConnection(*dispatcher_, fake_upstream_connection_));
@@ -99,8 +110,11 @@ class WasmFilterIntegrationTest
       upstream_request_->encodeHeaders(upstream_response_headers, true /*end_stream*/);
     } else {
       upstream_request_->encodeHeaders(upstream_response_headers, false /*end_stream*/);
-      Buffer::OwnedImpl buffer(upstream_response_body);
-      upstream_request_->encodeData(buffer, true);
+      const auto upstream_response_body_size = upstream_response_body.size();
+      for (size_t n = 0; n < upstream_response_body_size; n++) {
+        Buffer::OwnedImpl buffer(upstream_response_body[n]);
+        upstream_request_->encodeData(buffer, n == upstream_response_body_size - 1);
+      }
     }
 
     ASSERT_TRUE(response->waitForEndStream());
@@ -116,10 +130,9 @@ class WasmFilterIntegrationTest
   }
 };
 
-INSTANTIATE_TEST_SUITE_P(
-    Runtimes, WasmFilterIntegrationTest,
-    Envoy::Extensions::Common::Wasm::dual_filter_sandbox_runtime_and_cpp_values,
-    Envoy::Extensions::Common::Wasm::wasmDualFilterTestParamsToString);
+INSTANTIATE_TEST_SUITE_P(Runtimes, WasmFilterIntegrationTest,
+                         Common::Wasm::dual_filter_with_codecs_sandbox_runtime_and_cpp_values,
+                         Common::Wasm::wasmDualFilterWithCodecsTestParamsToString);
 GTEST_ALLOW_UNINSTANTIATED_PARAMETERIZED_TEST(WasmFilterIntegrationTest);
 
 TEST_P(WasmFilterIntegrationTest, HeadersManipulation) {
@@ -143,8 +156,10 @@ TEST_P(WasmFilterIntegrationTest, HeadersManipulation) {
   Http::TestResponseHeaderMapImpl expected_response_headers{
       {":status", "200"}, {"content-type", "application/json"}, {"test-status", "OK"}};
 
-  runTest(request_headers, "", expected_request_headers, "", upstream_response_headers, "",
-          expected_response_headers, "");
+  auto request_body = std::vector<std::string>{};
+  auto upstream_response_body = std::vector<std::string>{};
+  runTest(request_headers, request_body, expected_request_headers, "", upstream_response_headers,
+          upstream_response_body, expected_response_headers, "");
 }
 
 TEST_P(WasmFilterIntegrationTest, BodyManipulation) {
@@ -164,8 +179,59 @@ TEST_P(WasmFilterIntegrationTest, BodyManipulation) {
 
   Http::TestResponseHeaderMapImpl expected_response_headers{{":status", "200"}};
 
-  runTest(request_headers, "request_body", expected_request_headers, "replace",
-          upstream_response_headers, "response_body", expected_response_headers, "replace");
+  auto request_body = std::vector<std::string>{{"request_body"}};
+  auto upstream_response_body = std::vector<std::string>{{"upstream_body"}};
+  runTest(request_headers, request_body, expected_request_headers, "replace",
+          upstream_response_headers, upstream_response_body, expected_response_headers, "replace");
+}
+
+TEST_P(WasmFilterIntegrationTest, BodyBufferedManipulation) {
+  setupWasmFilter("", "body");
+  HttpIntegrationTest::initialize();
+
+  Http::TestRequestHeaderMapImpl request_headers{{":scheme", "http"},
+                                                 {":method", "GET"},
+                                                 {":path", "/"},
+                                                 {":authority", "host"},
+                                                 {"x-test-operation", "SetEndOfBodies"}};
+
+  Http::TestRequestHeaderMapImpl expected_request_headers{{":path", "/"}};
+
+  Http::TestResponseHeaderMapImpl upstream_response_headers{{":status", "200"},
+                                                            {"x-test-operation", "SetEndOfBodies"}};
+
+  Http::TestResponseHeaderMapImpl expected_response_headers{{":status", "200"}};
+
+  auto request_body = std::vector<std::string>{{"request_"}, {"body"}};
+  auto upstream_response_body = std::vector<std::string>{{"upstream_"}, {"body"}};
+  runTest(request_headers, request_body, expected_request_headers, "request_body.end",
+          upstream_response_headers, upstream_response_body, expected_response_headers,
+          "upstream_body.end");
+}
+
+TEST_P(WasmFilterIntegrationTest, BodyBufferedMultipleChunksManipulation) {
+  setupWasmFilter("", "body");
+  HttpIntegrationTest::initialize();
+
+  Http::TestRequestHeaderMapImpl request_headers{{":scheme", "http"},
+                                                 {":method", "GET"},
+                                                 {":path", "/"},
+                                                 {":authority", "host"},
+                                                 {"x-test-operation", "SetEndOfBodies"}};
+
+  Http::TestRequestHeaderMapImpl expected_request_headers{{":path", "/"}};
+
+  Http::TestResponseHeaderMapImpl upstream_response_headers{{":status", "200"},
+                                                            {"x-test-operation", "SetEndOfBodies"}};
+
+  Http::TestResponseHeaderMapImpl expected_response_headers{{":status", "200"}};
+
+  auto request_body = std::vector<std::string>{{"request_"}, {"very_"}, {"long_"}, {"body"}};
+  auto upstream_response_body =
+      std::vector<std::string>{{"upstream_"}, {"very_"}, {"long_"}, {"body"}};
+  runTest(request_headers, request_body, expected_request_headers, "request_very_long_body.end",
+          upstream_response_headers, upstream_response_body, expected_response_headers,
+          "upstream_very_long_body.end");
 }
 
 } // namespace