Tests: Reduce flakiness

WORKSPACE

@@ -42,6 +42,7 @@ git_repository(
         "@//patches:0003-original_dst_cluster-Avoid-multiple-hosts-for-the-sa.patch",
         "@//patches:0004-thread_local-reset-slot-in-worker-threads-first.patch",
         "@//patches:0005-http-header-expose-attribute.patch",
+        "@//patches:0006-test-integration-Defer-fake-upstream-read-enable-un.patch",
     ],
     # // clang-format off: Envoy's format check: Only repository_locations.bzl may contains URL references
     remote = "https://github.com/envoyproxy/envoy.git",

patches/0006-test-integration-Defer-fake-upstream-read-enable-un.patch

@@ -0,0 +1,95 @@
+diff --git a/test/integration/fake_upstream.cc b/test/integration/fake_upstream.cc
+--- a/test/integration/fake_upstream.cc
++++ b/test/integration/fake_upstream.cc
+@@ -436,6 +436,16 @@
+       Network::ReadFilterSharedPtr{new ReadFilter(*this)});
+ }
+
++void FakeHttpConnection::initialize() {
++  FakeConnectionBase::initialize();
++  if (shared_connection_.connected() && !shared_connection_.connection().readEnabled()) {
++    // FakeUpstream::consumeConnection() may hand HTTP connections off before the codec/filter
++    // stack is initialized. Re-enable reads only after initialize() has attached the HTTP read
++    // filter, or early request bytes can be consumed without ever creating a FakeStream.
++    shared_connection_.connection().readDisable(false);
++  }
++}
++
+ AssertionResult FakeConnectionBase::close(std::chrono::milliseconds timeout) {
+   ENVOY_LOG(trace, "FakeConnectionBase close");
+   if (!shared_connection_.connected()) {
+@@ -749,7 +756,8 @@
+   // not lazily create for HTTP/3
+   if (http_type_ == Http::CodecType::HTTP3) {
+     quic_connections_.push_back(std::make_unique<FakeHttpConnection>(
+-        *this, consumeConnection(), http_type_, time_system_, config_.max_request_headers_kb_,
++        *this, consumeConnection(/*defer_read_enable=*/true), http_type_, time_system_,
++        config_.max_request_headers_kb_,
+         config_.max_request_headers_count_, config_.headers_with_underscores_action_));
+     quic_connections_.back()->initialize();
+   }
+@@ -820,7 +828,8 @@
+   return runOnDispatcherThreadAndWait([&]() {
+     absl::MutexLock lock(lock_);
+     connection = std::make_unique<FakeHttpConnection>(
+-        *this, consumeConnection(), http_type_, time_system_, config_.max_request_headers_kb_,
++        *this, consumeConnection(/*defer_read_enable=*/true), http_type_, time_system_,
++        config_.max_request_headers_kb_,
+         config_.max_request_headers_count_, config_.headers_with_underscores_action_);
+     connection->initialize();
+     return AssertionSuccess();
+@@ -857,7 +866,8 @@
+       EXPECT_TRUE(upstream.runOnDispatcherThreadAndWait([&]() {
+         absl::MutexLock lock(&upstream.lock_);
+         connection = std::make_unique<FakeHttpConnection>(
+-            upstream, upstream.consumeConnection(), upstream.http_type_, upstream.timeSystem(),
++            upstream, upstream.consumeConnection(/*defer_read_enable=*/true),
++            upstream.http_type_, upstream.timeSystem(),
+             Http::DEFAULT_MAX_REQUEST_HEADERS_KB, Http::DEFAULT_MAX_HEADERS_COUNT,
+             envoy::config::core::v3::HttpProtocolOptions::ALLOW);
+         connection->initialize();
+@@ -920,7 +930,7 @@
+   raw_connection.release();
+ }
+
+-SharedConnectionWrapper& FakeUpstream::consumeConnection() {
++SharedConnectionWrapper& FakeUpstream::consumeConnection(bool defer_read_enable) {
+   ASSERT(!new_connections_.empty());
+   auto* const connection_wrapper = new_connections_.front().get();
+   // Skip the thread safety check if the network connection has already been freed since there's no
+@@ -930,10 +940,11 @@
+   connection_wrapper->moveBetweenLists(new_connections_, consumed_connections_);
+   if (read_disable_on_new_connection_ && connection_wrapper->connected() &&
+       http_type_ != Http::CodecType::HTTP3 && !disable_and_do_not_enable_) {
+-    // Re-enable read and early close detection.
+     auto& connection = connection_wrapper->connection();
+     connection.detectEarlyCloseWhenReadDisabled(true);
+-    connection.readDisable(false);
++    if (!defer_read_enable) {
++      connection.readDisable(false);
++    }
+   }
+   return *connection_wrapper;
+ }
+diff --git a/test/integration/fake_upstream.h b/test/integration/fake_upstream.h
+--- a/test/integration/fake_upstream.h
++++ b/test/integration/fake_upstream.h
+@@ -547,6 +547,8 @@
+                      envoy::config::core::v3::HttpProtocolOptions::HeadersWithUnderscoresAction
+                          headers_with_underscores_action);
+
++  void initialize() override;
++
+   ABSL_MUST_USE_RESULT
+   testing::AssertionResult
+   waitForNewStream(Event::Dispatcher& client_dispatcher, FakeStreamPtr& stream,
+@@ -998,7 +1000,8 @@
+   };
+
+   void threadRoutine();
+-  SharedConnectionWrapper& consumeConnection() ABSL_EXCLUSIVE_LOCKS_REQUIRED(lock_);
++  SharedConnectionWrapper& consumeConnection(bool defer_read_enable = false)
++      ABSL_EXCLUSIVE_LOCKS_REQUIRED(lock_);
+   Network::FilterStatus onRecvDatagram(Network::UdpRecvData& data);
+   AssertionResult
+   runOnDispatcherThreadAndWait(std::function<AssertionResult()> cb,

tests/accesslog_server.cc

@@ -16,9 +16,11 @@
 namespace Envoy {
 
 AccessLogServer::AccessLogServer(const std::string path)
-    : UDSServer(path, std::bind(&AccessLogServer::msgCallback, this, std::placeholders::_1)) {}
+    : UDSServer(path, std::bind(&AccessLogServer::msgCallback, this, std::placeholders::_1)) {
+  startServerThread();
+}
 
-AccessLogServer::~AccessLogServer() = default;
+AccessLogServer::~AccessLogServer() { shutdownServerThread(); }
 
 void AccessLogServer::clear() {
   absl::MutexLock lock(&mutex_);

tests/accesslog_server.h

@@ -17,7 +17,7 @@ namespace Envoy {
 class AccessLogServer : public UDSServer {
 public:
   AccessLogServer(const std::string path);
-  ~AccessLogServer();
+  ~AccessLogServer() override;
 
   void clear();
   absl::optional<::cilium::LogEntry>

tests/cilium_http_integration.cc

@@ -33,7 +33,15 @@ CiliumHttpIntegrationTest::CiliumHttpIntegrationTest(const std::string& config)
 #endif
 }
 
-CiliumHttpIntegrationTest::~CiliumHttpIntegrationTest() = default;
+CiliumHttpIntegrationTest::~CiliumHttpIntegrationTest() {
+  // Shut down live downstream/upstream traffic before the access log UDS server member is
+  // destroyed. Otherwise the UDS sink can disappear while HttpIntegrationTest base teardown is
+  // still closing HTTP connections and the Envoy test server.
+  cleanupUpstreamAndDownstream();
+  codec_client_.reset();
+  test_server_.reset();
+  fake_upstreams_.clear();
+}
 
 void CiliumHttpIntegrationTest::createEnvoy() {
   // fake upstreams have been created by now, use the port from the 1st upstream

tests/cilium_tcp_integration.cc

@@ -40,8 +40,7 @@ const std::string TCP_POLICY_fmt = R"EOF(version_info: "0"
 )EOF";
 
 CiliumTcpIntegrationTest::CiliumTcpIntegrationTest(const std::string& config)
-    : BaseIntegrationTest(GetParam(), config),
-      accessLogServer_(TestEnvironment::unixDomainSocketPath("access_log.sock")) {
+    : BaseIntegrationTest(GetParam(), config) {
   enableHalfClose(true);
 #if 1
   for (Logger::Logger& logger : Logger::Registry::loggers()) {

tests/cilium_tcp_integration.h

@@ -8,8 +8,6 @@
 
 #include "test/integration/base_integration_test.h"
 
-#include "tests/accesslog_server.h"
-
 namespace Envoy {
 
 class CiliumTcpIntegrationTest : public BaseIntegrationTest,
@@ -22,8 +20,6 @@ class CiliumTcpIntegrationTest : public BaseIntegrationTest,
   virtual std::string testPolicyFmt();
 
   void initialize() override;
-
-  AccessLogServer accessLogServer_;
 };
 
 } // namespace Envoy

tests/cilium_tls_http_integration_test.cc

@@ -234,7 +234,6 @@ class CiliumHttpTLSIntegrationTest : public CiliumHttpIntegrationTest {
 
   void initialize() override {
     CiliumHttpIntegrationTest::initialize();
-    fake_upstreams_[0]->setReadDisableOnNewConnection(false);
 
     // Set up the SSL client.
     Network::Address::InstanceConstSharedPtr address =