[bp/1.37] Flakes stack

changelogs/current.yaml

@@ -11,6 +11,9 @@ bug_fixes:
 - area: http2
   change: |
     Apply nghttp2 CVE-2026-27135 patch.
+- area: load_report
+  change: |
+    Fixed an issue upon load-report shutdown race with ADS stream. Introduced proper cleanup of the gRPC stream.
 
 removed_config_or_runtime:
 # *Normally occurs at the end of the* :ref:`deprecation period <deprecated>`

source/common/quic/envoy_quic_client_session.h

@@ -117,9 +117,10 @@ class EnvoyQuicClientSession : public QuicFilterManagerConnectionImpl,
   quic::QuicSpdyStream* CreateIncomingStream(quic::PendingStream* pending) override;
   std::unique_ptr<quic::QuicCryptoClientStreamBase> CreateQuicCryptoStream() override;
   bool ShouldCreateOutgoingBidirectionalStream() override {
-    ASSERT(quic::QuicSpdyClientSession::ShouldCreateOutgoingBidirectionalStream());
-    // Prefer creating an "invalid" stream outside of current stream bounds to
-    // crashing when dereferencing a nullptr in QuicHttpClientConnectionImpl::newStream
+    // quic::QuicSpdyClientSession::ShouldCreateOutgoingBidirectionalStream()
+    // might return false, but we want to create the stream anyway
+    // because otherwise we crash dereferencing a nullptr, so we
+    // don't even ask it, and just return true.
     return true;
   }
   // QuicFilterManagerConnectionImpl

source/common/upstream/cluster_manager_impl.h

@@ -312,6 +312,7 @@ class ClusterManagerImpl : public ClusterManager,
     // Make sure we destroy all potential outgoing connections before this returns.
     cds_api_.reset();
     xds_manager_.shutdown();
+    load_stats_reporter_.reset();
     active_clusters_.clear();
     warming_clusters_.clear();
     updateClusterCounts();

source/common/upstream/load_stats_reporter.cc

@@ -39,6 +39,17 @@ LoadStatsReporter::LoadStatsReporter(const LocalInfo::LocalInfo& local_info,
   establishNewStream();
 }
 
+LoadStatsReporter::~LoadStatsReporter() {
+  // Disable the timer.
+  ENVOY_LOG_MISC(info, "Destroying LoadStatsReporter");
+  retry_timer_->disableTimer();
+  response_timer_->disableTimer();
+  if (stream_ != nullptr) {
+    stream_->resetStream();
+    stream_ = nullptr;
+  }
+}
+
 void LoadStatsReporter::setRetryTimer() {
   ENVOY_LOG(info, "Load reporter stats stream/connection will retry in {} ms.", RETRY_DELAY_MS);
   retry_timer_->enableTimer(std::chrono::milliseconds(RETRY_DELAY_MS));

source/common/upstream/load_stats_reporter.h

@@ -36,6 +36,7 @@ class LoadStatsReporter
   LoadStatsReporter(const LocalInfo::LocalInfo& local_info, ClusterManager& cluster_manager,
                     Stats::Scope& scope, Grpc::RawAsyncClientSharedPtr&& async_client,
                     Event::Dispatcher& dispatcher);
+  virtual ~LoadStatsReporter();
 
   // Grpc::AsyncStreamCallbacks
   void onCreateInitialMetadata(Http::RequestHeaderMap& metadata) override;

test/common/upstream/load_stats_reporter_test.cc

@@ -37,6 +37,17 @@ class LoadStatsReporterTest : public testing::Test {
       : retry_timer_(new Event::MockTimer()), response_timer_(new Event::MockTimer()),
         async_client_(new Grpc::MockAsyncClient()) {}
 
+  void TearDown() override {
+    if (load_stats_reporter_ != nullptr) {
+      // Validate that LoadStatsReporter correctly shuts down by disabling
+      // timers and resetting the stream.
+      EXPECT_CALL(*retry_timer_, disableTimer());
+      EXPECT_CALL(*response_timer_, disableTimer());
+      EXPECT_CALL(async_stream_, resetStream());
+      load_stats_reporter_.reset();
+    }
+  }
+
   void createLoadStatsReporter() {
     InSequence s;
     EXPECT_CALL(dispatcher_, createTimer_(_)).WillOnce(Invoke([this](Event::TimerCb timer_cb) {
@@ -97,14 +108,14 @@ class LoadStatsReporterTest : public testing::Test {
   NiceMock<Upstream::MockClusterManager> cm_;
   Event::MockDispatcher dispatcher_;
   Stats::IsolatedStoreImpl stats_store_;
-  std::unique_ptr<LoadStatsReporter> load_stats_reporter_;
   Event::MockTimer* retry_timer_;
   Event::TimerCb retry_timer_cb_;
   Event::MockTimer* response_timer_;
   Event::TimerCb response_timer_cb_;
   Grpc::MockAsyncStream async_stream_;
   Grpc::MockAsyncClient* async_client_;
   NiceMock<LocalInfo::MockLocalInfo> local_info_;
+  std::unique_ptr<LoadStatsReporter> load_stats_reporter_;
 };
 
 // Validate that stream creation results in a timer based retry.

test/extensions/filters/http/grpc_json_transcoder/grpc_json_transcoder_integration_test.cc

@@ -1299,7 +1299,7 @@ TEST_P(GrpcJsonTranscoderIntegrationTest, UTF8) {
       false);
 }
 
-TEST_P(GrpcJsonTranscoderIntegrationTest, DisableRequestValidation) {
+TEST_P(GrpcJsonTranscoderIntegrationTest, DisableRequestValidationGrpcContentType) {
   HttpIntegrationTest::initialize();
 
   // Transcoding does not occur from a request with the gRPC content type.
@@ -1315,7 +1315,10 @@ TEST_P(GrpcJsonTranscoderIntegrationTest, DisableRequestValidation) {
       Http::TestResponseHeaderMapImpl{
           {":status", "200"}, {"grpc-status", "5"}, {"grpc-message", "Shelf 9999 Not Found"}},
       "", true, false, R"({ "theme" : "Children")");
+}
 
+TEST_P(GrpcJsonTranscoderIntegrationTest, DisableRequestValidationUnknownPath) {
+  HttpIntegrationTest::initialize();
   // Transcoding does not occur when unknown path is called.
   // HTTP Request to is passed directly to gRPC backend.
   // gRPC response is passed directly to HTTP client.
@@ -1328,7 +1331,10 @@ TEST_P(GrpcJsonTranscoderIntegrationTest, DisableRequestValidation) {
       Http::TestResponseHeaderMapImpl{
           {":status", "200"}, {"grpc-status", "5"}, {"grpc-message", "Shelf 9999 Not Found"}},
       "", true, false, R"({ "theme" : "Children")");
+}
 
+TEST_P(GrpcJsonTranscoderIntegrationTest, DisableRequestValidationUnknownQueryParam) {
+  HttpIntegrationTest::initialize();
   // Transcoding does not occur when unknown query param is included.
   // HTTP Request to is passed directly to gRPC backend.
   // gRPC response is passed directly to HTTP client.

test/extensions/filters/udp/dns_filter/dns_filter_integration_test.cc

@@ -89,6 +89,13 @@ class DnsFilterIntegrationTest : public testing::TestWithParam<Network::Address:
 
   void setupResponseParser() { histogram_.unit_ = Stats::Histogram::Unit::Milliseconds; }
 
+  void TearDown() override {
+    if (test_server_) {
+      test_server_.reset();
+    }
+    fake_upstreams_.clear();
+  }
+
   static std::string configToUse() {
     return fmt::format(R"EOF(
 admin:

test/integration/BUILD

@@ -1665,12 +1665,9 @@ envoy_cc_test_library(
 
 envoy_cc_test(
     name = "websocket_integration_test",
-    size = "large",
     srcs = ["websocket_integration_test.cc"],
-    rbe_pool = "2core",
-    tags = [
-        "cpu:3",
-    ],
+    rbe_pool = "linux_x64_small",
+    shard_count = 8,
     deps = [
         ":http_protocol_integration_lib",
         ":websocket_integration_test_lib",

test/integration/idle_timeout_integration_test.cc

@@ -117,7 +117,7 @@ class IdleTimeoutIntegrationTest : public HttpProtocolIntegrationTest {
   }
 
   static constexpr uint64_t IdleTimeoutMs = 300 * TIMEOUT_FACTOR;
-  static constexpr uint64_t RequestTimeoutMs = 200;
+  static constexpr uint64_t RequestTimeoutMs = 200 * TIMEOUT_FACTOR;
   bool enable_global_idle_timeout_{false};
   bool enable_per_stream_idle_timeout_{false};
   bool enable_request_timeout_{false};

test/integration/websocket_integration_test.cc

@@ -262,8 +262,8 @@ TEST_P(WebsocketIntegrationTest, WebSocketConnectionIdleTimeout) {
         auto* route_config = hcm.mutable_route_config();
         auto* virtual_host = route_config->mutable_virtual_hosts(0);
         auto* route = virtual_host->mutable_routes(0)->mutable_route();
-        route->mutable_idle_timeout()->set_seconds(0);
-        route->mutable_idle_timeout()->set_nanos(200 * 1000 * 1000);
+        route->mutable_idle_timeout()->set_seconds(1);
+        route->mutable_idle_timeout()->set_nanos(0);
       });
   initialize();
 
@@ -850,8 +850,8 @@ TEST_P(WebsocketIntegrationTest, WebSocketUpgradePerTryTimeout) {
         auto* route_config = hcm.mutable_route_config();
         auto* virtual_host = route_config->mutable_virtual_hosts(0);
         auto* route = virtual_host->mutable_routes(0)->mutable_route();
-        route->mutable_retry_policy()->mutable_per_try_timeout()->set_nanos(
-            200 * 1000 * 1000); // 200ms per-try timeout
+        route->mutable_retry_policy()->mutable_per_try_timeout()->set_seconds(
+            2); // 2s per-try timeout (safe under msan/tsan)
       });
   initialize();
 
@@ -886,7 +886,7 @@ TEST_P(WebsocketIntegrationTest, WebSocketUpgradeRouteTimeout) {
         auto* route_config = hcm.mutable_route_config();
         auto* virtual_host = route_config->mutable_virtual_hosts(0);
         auto* route = virtual_host->mutable_routes(0)->mutable_route();
-        route->mutable_timeout()->set_nanos(200 * 1000 * 1000); // 200ms route timeout
+        route->mutable_timeout()->set_seconds(2); // 2s route timeout (safe under msan/tsan)
       });
   initialize();
 
@@ -922,7 +922,7 @@ TEST_P(WebsocketIntegrationTest, WebSocketUpgradeRouteTimeoutWithRetries) {
         auto* route_config = hcm.mutable_route_config();
         auto* virtual_host = route_config->mutable_virtual_hosts(0);
         auto* route = virtual_host->mutable_routes(0)->mutable_route();
-        route->mutable_timeout()->set_nanos(200 * 1000 * 1000); // 200ms route timeout
+        route->mutable_timeout()->set_seconds(2); // 2s route timeout
       });
   initialize();
 
@@ -936,7 +936,7 @@ TEST_P(WebsocketIntegrationTest, WebSocketUpgradeRouteTimeoutWithRetries) {
   ASSERT_TRUE(fake_upstreams_[0]->waitForHttpConnection(*dispatcher_, fake_upstream_connection_));
   ASSERT_TRUE(fake_upstream_connection_->waitForNewStream(*dispatcher_, upstream_request_));
   ASSERT_TRUE(upstream_request_->waitForHeadersComplete());
-  upstream_request_->encodeHeaders(upgradeFailedResponseHeaders(), false);
+  upstream_request_->encodeHeaders(upgradeFailedResponseHeaders(), true);
 
   // Wait for the first request to be reset or disconnected
   ASSERT_TRUE(waitForUpstreamDisconnectOrReset());