Make filter fuzz tester support filters that process off-thread (envo…

…yproxy#22470)

Signed-off-by: Raven Black <[email protected]>

test/extensions/filters/http/common/fuzz/http_filter_fuzzer.h

@@ -44,13 +44,21 @@ class HttpFilterFuzzer {
   // Fuzzed headers and trailers are needed for access logging, reset the data and destroy filters.
   void reset() {
     enabled_ = true;
+    encoding_finished_ = false;
+    decoding_finished_ = false;
     request_headers_.clear();
     response_headers_.clear();
     request_trailers_.clear();
     response_trailers_.clear();
     encoded_trailers_.clear();
   }
 
+  // Returns true if decoder and encoder are both finished or nonexistent.
+  bool isFilterFinished() { return decoding_finished_ && encoding_finished_; }
+
+  // Records that the filter type has finished processing.
+  template <class FilterType> void finishFilter(FilterType* filter) = delete;
+
 protected:
   // Templated functions to validate and send headers/data/trailers for decoders/encoders.
   // General functions are deleted, but templated specializations for encoders/decoders are defined
@@ -71,6 +79,9 @@ class HttpFilterFuzzer {
   // sendLocalReply to set enabled_ to false.
   bool enabled_ = true;
 
+  bool decoding_finished_ = false;
+  bool encoding_finished_ = false;
+
   // Headers/trailers need to be saved for the lifetime of the filter,
   // so save them as member variables.
   Http::TestRequestHeaderMapImpl request_headers_;
@@ -89,9 +100,8 @@ void HttpFilterFuzzer::runData(FilterType* filter, const test::fuzz::HttpData& d
   }
   const auto& headersStatus = sendHeaders(filter, data, end_stream);
   ENVOY_LOG_MISC(debug, "Finished with FilterHeadersStatus: {}", static_cast<int>(headersStatus));
-  if ((headersStatus != Http::FilterHeadersStatus::Continue &&
-       headersStatus != Http::FilterHeadersStatus::StopIteration) ||
-      !enabled_) {
+  if ((end_stream && headersStatus == Http::FilterHeadersStatus::Continue) || !enabled_) {
+    finishFilter(filter);
     return;
   }
 
@@ -103,13 +113,15 @@ void HttpFilterFuzzer::runData(FilterType* filter, const test::fuzz::HttpData& d
     Buffer::OwnedImpl buffer(data_chunks[i]);
     const auto& dataStatus = sendData(filter, buffer, end_stream);
     ENVOY_LOG_MISC(debug, "Finished with FilterDataStatus: {}", static_cast<int>(dataStatus));
-    if (dataStatus != Http::FilterDataStatus::Continue || !enabled_) {
+    if ((end_stream && dataStatus == Http::FilterDataStatus::Continue) || !enabled_) {
+      finishFilter(filter);
       return;
     }
   }
 
   if (data.has_trailers() && enabled_) {
     sendTrailers(filter, data);
+    finishFilter(filter);
   }
 }
 
@@ -157,6 +169,14 @@ inline Http::FilterHeadersStatus HttpFilterFuzzer::sendHeaders(Http::StreamEncod
   return status;
 }
 
+template <> inline void HttpFilterFuzzer::finishFilter(Http::StreamDecoderFilter*) {
+  decoding_finished_ = true;
+}
+
+template <> inline void HttpFilterFuzzer::finishFilter(Http::StreamEncoderFilter*) {
+  encoding_finished_ = true;
+}
+
 template <>
 inline Http::FilterDataStatus HttpFilterFuzzer::sendData(Http::StreamDecoderFilter* filter,
                                                          Buffer::Instance& buffer,

test/extensions/filters/http/common/fuzz/uber_filter.cc

@@ -1,6 +1,8 @@
 #include "test/extensions/filters/http/common/fuzz/uber_filter.h"
 
+#include "source/common/common/thread_impl.h"
 #include "source/common/config/utility.h"
+#include "source/common/event/dispatcher_impl.h"
 #include "source/common/http/message_impl.h"
 #include "source/common/http/utility.h"
 #include "source/common/protobuf/protobuf.h"
@@ -13,7 +15,11 @@ namespace Extensions {
 namespace HttpFilters {
 
 UberFilterFuzzer::UberFilterFuzzer()
-    : async_request_{&cluster_manager_.thread_local_cluster_.async_client_} {
+    : async_request_{&cluster_manager_.thread_local_cluster_.async_client_},
+      thread_factory_(Thread::threadFactoryForTest()) {
+  ON_CALL(api_, threadFactory()).WillByDefault(testing::ReturnRef(thread_factory_));
+  worker_thread_dispatcher_ =
+      std::make_unique<Event::DispatcherImpl>("filter_fuzz_test", api_, api_.time_system_);
   // This is a decoder filter.
   ON_CALL(filter_callback_, addStreamDecoderFilter(_))
       .WillByDefault(Invoke([&](Http::StreamDecoderFilterSharedPtr filter) -> void {
@@ -77,14 +83,38 @@ void UberFilterFuzzer::fuzz(
   // Data path should not throw exceptions.
   if (decoder_filter_ != nullptr) {
     HttpFilterFuzzer::runData(decoder_filter_.get(), downstream_data);
+  } else {
+    decoding_finished_ = true;
   }
   if (encoder_filter_ != nullptr) {
     HttpFilterFuzzer::runData(encoder_filter_.get(), upstream_data);
+  } else {
+    encoding_finished_ = true;
   }
   if (access_logger_ != nullptr) {
     HttpFilterFuzzer::accessLog(access_logger_.get(), stream_info_);
   }
 
+  // Most filters should have finished processing during runData, but filters that
+  // rely on an additional thread (e.g. for file system interaction) may need to wait
+  // for the worker thread to complete the filter's task.
+  // We can't use a time-based timeout for this, as lint forbids use of clock time,
+  // and fake time isn't useful for allowing other threads to complete work.
+  int loop_cycles = 5000;
+  while (!isFilterFinished() && --loop_cycles > 0) {
+    worker_thread_dispatcher_->run(Event::DispatcherImpl::RunType::NonBlock);
+    // Apparently RunType::Block doesn't actually block if there's only a timer
+    // event not ready to fire, so we use NonBlock for clarity, and this loop
+    // spins. We yield to keep it from spinning too wildly. For almost all cases
+    // this shouldn't matter as even entering the loop at all is rare, and for
+    // the cases where the loop is useful, a few cycles should be enough to
+    // complete unless the test is genuinely failing.
+    std::this_thread::yield();
+  }
+  if (!isFilterFinished()) {
+    throw EnvoyException("filter did not finish processing");
+  }
+
   reset();
 }
 

test/extensions/filters/http/common/fuzz/uber_filter.h

@@ -4,6 +4,7 @@
 
 #include "test/extensions/filters/http/common/fuzz/http_filter_fuzzer.h"
 #include "test/fuzz/utility.h"
+#include "test/mocks/api/mocks.h"
 #include "test/mocks/buffer/mocks.h"
 #include "test/mocks/http/mocks.h"
 #include "test/mocks/server/factory_context.h"
@@ -58,6 +59,10 @@ class UberFilterFuzzer : public HttpFilterFuzzer {
   NiceMock<Http::MockStreamDecoderFilterCallbacks> decoder_callbacks_;
   NiceMock<Http::MockStreamEncoderFilterCallbacks> encoder_callbacks_;
 
+  Api::MockApi api_{};
+  Thread::ThreadFactory& thread_factory_;
+  Event::DispatcherPtr worker_thread_dispatcher_;
+
   const Buffer::Instance* decoding_buffer_{};
 };
 

test/extensions/filters/http/common/fuzz/uber_per_filter.cc

@@ -179,6 +179,28 @@ void UberFilterFuzzer::perFilterSetup() {
   ON_CALL(decoder_callbacks_, decodingBuffer()).WillByDefault([this]() -> const Buffer::Instance* {
     return decoding_buffer_;
   });
+  ON_CALL(encoder_callbacks_, dispatcher()).WillByDefault([this]() -> Event::Dispatcher& {
+    return *worker_thread_dispatcher_;
+  });
+  ON_CALL(decoder_callbacks_, dispatcher()).WillByDefault([this]() -> Event::Dispatcher& {
+    return *worker_thread_dispatcher_;
+  });
+  ON_CALL(encoder_callbacks_, injectEncodedDataToFilterChain(_, true))
+      .WillByDefault([this]() -> void { finishFilter(encoder_filter_.get()); });
+  ON_CALL(encoder_callbacks_, addEncodedData(_, true)).WillByDefault([this]() -> void {
+    finishFilter(encoder_filter_.get());
+  });
+  ON_CALL(encoder_callbacks_, continueEncoding()).WillByDefault([this]() -> void {
+    finishFilter(encoder_filter_.get());
+  });
+  ON_CALL(decoder_callbacks_, injectDecodedDataToFilterChain(_, true))
+      .WillByDefault([this]() -> void { finishFilter(decoder_filter_.get()); });
+  ON_CALL(decoder_callbacks_, addDecodedData(_, true)).WillByDefault([this]() -> void {
+    finishFilter(decoder_filter_.get());
+  });
+  ON_CALL(decoder_callbacks_, continueDecoding()).WillByDefault([this]() -> void {
+    finishFilter(decoder_filter_.get());
+  });
 }
 
 } // namespace HttpFilters