From 2eebf0dc57c2b5393b2deaeb0166cb8a9a6bffa3 Mon Sep 17 00:00:00 2001
From: Tianyu <72890320+tyxia@users.noreply.github.com>
Date: Thu, 10 Oct 2024 11:55:06 -0400
Subject: [PATCH] ext_proc: skip timeout timer on trailer in async mode.
 (#36524)

Commit Message: Like header and body, timer/timeout should be skipped on
trailer as well in async. It is because there is no response/ or
response is ignored.
Risk Level: Low
Testing: Integration test
Docs Changes: N/A
Release Notes: N/A
Platform Specific Features: N/A

---------

Signed-off-by: tyxia <tyxia@google.com>
---
 .../filters/http/ext_proc/ext_proc.cc         | 13 +++++--
 .../ext_proc/ext_proc_integration_test.cc     | 37 +++++++++++++++++++
 2 files changed, 47 insertions(+), 3 deletions(-)

diff --git a/source/extensions/filters/http/ext_proc/ext_proc.cc b/source/extensions/filters/http/ext_proc/ext_proc.cc
index 2f3e39e5eb06..0d35d0542581 100644
--- a/source/extensions/filters/http/ext_proc/ext_proc.cc
+++ b/source/extensions/filters/http/ext_proc/ext_proc.cc
@@ -976,9 +976,16 @@ void Filter::sendTrailers(ProcessorState& state, const Http::HeaderMap& trailers
   auto* trailers_req = state.mutableTrailers(req);
   MutationUtils::headersToProto(trailers, config_->allowedHeaders(), config_->disallowedHeaders(),
                                 *trailers_req->mutable_trailers());
-  state.onStartProcessorCall(std::bind(&Filter::onMessageTimeout, this), config_->messageTimeout(),
-                             ProcessorState::CallbackState::TrailersCallback);
-  ENVOY_LOG(debug, "Sending trailers message");
+
+  if (observability_mode) {
+    ENVOY_LOG(debug, "Sending trailers message in observability mode");
+  } else {
+    state.onStartProcessorCall(std::bind(&Filter::onMessageTimeout, this),
+                               config_->messageTimeout(),
+                               ProcessorState::CallbackState::TrailersCallback);
+    ENVOY_LOG(debug, "Sending trailers message");
+  }
+
   sendRequest(std::move(req), false);
   stats_.stream_msgs_sent_.inc();
 }
diff --git a/test/extensions/filters/http/ext_proc/ext_proc_integration_test.cc b/test/extensions/filters/http/ext_proc/ext_proc_integration_test.cc
index e229fc88b850..7762d38d5b27 100644
--- a/test/extensions/filters/http/ext_proc/ext_proc_integration_test.cc
+++ b/test/extensions/filters/http/ext_proc/ext_proc_integration_test.cc
@@ -4334,6 +4334,43 @@ TEST_P(ExtProcIntegrationTest, ObservabilityModeWithFullResponse) {
   timeSystem().advanceTimeWaitImpl(std::chrono::milliseconds(deferred_close_timeout_ms));
 }
 
+TEST_P(ExtProcIntegrationTest, ObservabilityModeWithFullRequestAndTimeout) {
+  proto_config_.set_observability_mode(true);
+  uint32_t deferred_close_timeout_ms = 2000;
+  proto_config_.mutable_deferred_close_timeout()->set_seconds(deferred_close_timeout_ms / 1000);
+
+  proto_config_.mutable_processing_mode()->set_request_body_mode(ProcessingMode::STREAMED);
+  proto_config_.mutable_processing_mode()->set_request_trailer_mode(ProcessingMode::SEND);
+  proto_config_.mutable_processing_mode()->set_response_header_mode(ProcessingMode::SKIP);
+
+  initializeConfig();
+  HttpIntegrationTest::initialize();
+  auto response = sendDownstreamRequestWithBodyAndTrailer("Hello");
+
+  processRequestHeadersMessage(*grpc_upstreams_[0], true,
+                               [this](const HttpHeaders&, HeadersResponse&) {
+                                 // Advance 400 ms. Default timeout is 200ms
+                                 timeSystem().advanceTimeWaitImpl(400ms);
+                                 return false;
+                               });
+  processRequestBodyMessage(*grpc_upstreams_[0], false, [this](const HttpBody&, BodyResponse&) {
+    // Advance 400 ms. Default timeout is 200ms
+    timeSystem().advanceTimeWaitImpl(400ms);
+    return false;
+  });
+  processRequestTrailersMessage(*grpc_upstreams_[0], false,
+                                [this](const HttpTrailers&, TrailersResponse&) {
+                                  // Advance 400 ms. Default timeout is 200ms
+                                  timeSystem().advanceTimeWaitImpl(400ms);
+                                  return false;
+                                });
+
+  handleUpstreamRequest();
+  verifyDownstreamResponse(*response, 200);
+
+  timeSystem().advanceTimeWaitImpl(std::chrono::milliseconds(deferred_close_timeout_ms - 1200));
+}
+
 TEST_P(ExtProcIntegrationTest, ObservabilityModeWithLogging) {
   proto_config_.set_observability_mode(true);