ext_proc: remove unnecessary watermark (#36468)

Commit Message: `StopIterationAndWatermark` will raise the watermark
when buffered data exceeds the limits, the `requestWatermark` here is
redundant and it will also introduce unnecessary stall of Envoy
processing and overhead of raise and clear watermark possibly for small
bodies

Risk Level: LOW
Testing: 
1.Passed all functional tests (unit test and integration test) 

2.This PR performs slightly better in load test.
```
4KB  request body:  
Without this PR:  Memory 183.29MB; Latency P99 : 2716
With this PR:  Memory: 160.17MB;    Latency P99: 2624

64KB request body:
Without this PR:  Memory 178.49MB; Latency P99 : 3512
With this PR:  Memory: 172.70MB;    Latency P99: 3505
```

Signed-off-by: tyxia <[email protected]>

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

@@ -579,13 +579,11 @@ FilterDataStatus Filter::onData(ProcessorState& state, Buffer::Instance& data, b
     } else {
       ENVOY_LOG(trace, "Header processing still in progress -- holding body data");
       // We don't know what to do with the body until the response comes back.
-      // We must buffer it in case we need it when that happens.
-      // Raise a watermark to prevent a buffer overflow until the response comes back.
-      // When end_stream is true, we need to StopIterationAndWatermark as well to stop the
-      // ActiveStream from returning error when the last chunk added to stream buffer exceeds the
-      // buffer limit.
+      // We must buffer it in case we need it when that happens. Watermark will be raised when the
+      // buffered data reaches the buffer's watermark limit. When end_stream is true, we need to
+      // StopIterationAndWatermark as well to stop the ActiveStream from returning error when the
+      // last chunk added to stream buffer exceeds the buffer limit.
       state.setPaused(true);
-      state.requestWatermark();
       return FilterDataStatus::StopIterationAndWatermark;
     }
   }

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

@@ -1068,8 +1068,6 @@ TEST_F(HttpFilterTest, PostAndChangeRequestBodyBufferedComesFast) {
   Buffer::OwnedImpl buffered_data;
   setUpDecodingBuffering(buffered_data);
 
-  // Buffering and callback isn't complete so we should watermark
-  EXPECT_CALL(decoder_callbacks_, onDecoderFilterAboveWriteBufferHighWatermark());
   EXPECT_EQ(FilterDataStatus::StopIterationAndWatermark, filter_->decodeData(req_data_1, false));
   buffered_data.add(req_data_1);
 
@@ -1082,7 +1080,6 @@ TEST_F(HttpFilterTest, PostAndChangeRequestBodyBufferedComesFast) {
   EXPECT_EQ(FilterDataStatus::StopIterationAndWatermark, filter_->decodeData(req_data_4, true));
   buffered_data.add(req_data_4);
 
-  EXPECT_CALL(decoder_callbacks_, onDecoderFilterBelowWriteBufferLowWatermark());
   processRequestHeaders(true, absl::nullopt);
 
   processRequestBody([](const HttpBody& req_body, ProcessingResponse&, BodyResponse&) {
@@ -1136,15 +1133,11 @@ TEST_F(HttpFilterTest, PostAndChangeRequestBodyBufferedComesALittleFast) {
   Buffer::OwnedImpl buffered_data;
   setUpDecodingBuffering(buffered_data);
 
-  // Buffering and callback isn't complete so we should watermark
-  EXPECT_CALL(decoder_callbacks_, onDecoderFilterAboveWriteBufferHighWatermark());
   EXPECT_EQ(FilterDataStatus::StopIterationAndWatermark, filter_->decodeData(req_data_1, false));
   buffered_data.add(req_data_1);
   EXPECT_EQ(FilterDataStatus::StopIterationAndWatermark, filter_->decodeData(req_data_2, false));
   buffered_data.add(req_data_2);
 
-  // Now the headers response comes in before we get all the data
-  EXPECT_CALL(decoder_callbacks_, onDecoderFilterBelowWriteBufferLowWatermark());
   processRequestHeaders(true, absl::nullopt);
 
   EXPECT_EQ(FilterDataStatus::StopIterationAndBuffer, filter_->decodeData(req_data_3, false));
@@ -1454,15 +1447,11 @@ TEST_F(HttpFilterTest, PostFastRequestPartialBuffering) {
   Buffer::OwnedImpl buffered_data;
   setUpDecodingBuffering(buffered_data);
 
-  // Buffering and callback isn't complete so we should watermark
-  EXPECT_CALL(decoder_callbacks_, onDecoderFilterAboveWriteBufferHighWatermark());
   EXPECT_EQ(FilterDataStatus::StopIterationAndWatermark, filter_->decodeData(req_data_1, false));
   buffered_data.add(req_data_1);
   EXPECT_EQ(FilterDataStatus::StopIterationAndWatermark, filter_->decodeData(req_data_2, true));
   buffered_data.add(req_data_2);
 
-  // Now the headers response comes in and we are all done
-  EXPECT_CALL(decoder_callbacks_, onDecoderFilterBelowWriteBufferLowWatermark());
   processRequestHeaders(true, absl::nullopt);
 
   processRequestBody([](const HttpBody& req_body, ProcessingResponse&, BodyResponse&) {
@@ -1518,8 +1507,6 @@ TEST_F(HttpFilterTest, PostFastAndBigRequestPartialBuffering) {
   expected_request_data.add(req_data_1);
   expected_request_data.add(req_data_2);
 
-  // Buffering and callback isn't complete so we should watermark
-  EXPECT_CALL(decoder_callbacks_, onDecoderFilterAboveWriteBufferHighWatermark());
   EXPECT_EQ(FilterDataStatus::StopIterationAndWatermark, filter_->decodeData(req_data_1, false));
   buffered_data.add(req_data_1);
   EXPECT_EQ(FilterDataStatus::StopIterationAndWatermark, filter_->decodeData(req_data_2, false));

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

@@ -492,13 +492,7 @@ TEST_F(OrderingTest, ResponseSomeDataComesFast) {
 
   EXPECT_CALL(stream_delegate_, send(_, false));
   sendResponseHeaders(true);
-  // Some of the data might come back but we should watermark so that we
-  // don't fill the buffer.
-  EXPECT_CALL(encoder_callbacks_, onEncoderFilterAboveWriteBufferHighWatermark());
   EXPECT_EQ(FilterDataStatus::StopIterationAndWatermark, filter_->encodeData(resp_body_1, false));
-
-  // When the response does comes back, we should lift the watermark
-  EXPECT_CALL(encoder_callbacks_, onEncoderFilterBelowWriteBufferLowWatermark());
   sendResponseHeadersReply();
 
   EXPECT_CALL(stream_delegate_, send(_, false));