[bp/1.35] Flake stack

changelogs/current.yaml

@@ -8,6 +8,9 @@ minor_behavior_changes:
 
 bug_fixes:
 # *Changes expected to improve the state of the world and are unlikely to have negative effects*
+- 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

@@ -105,9 +105,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

@@ -310,6 +310,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

@@ -28,6 +28,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::RawAsyncClientPtr async_client,
                     Event::Dispatcher& dispatcher);
+  virtual ~LoadStatsReporter();
 
   // Grpc::AsyncStreamCallbacks
   void onCreateInitialMetadata(Http::RequestHeaderMap& metadata) override;

test/common/common/matchers_test.cc

@@ -12,6 +12,7 @@
 #include "test/mocks/server/server_factory_context.h"
 #include "test/test_common/utility.h"
 
+#include "gmock/gmock.h"
 #include "gtest/gtest.h"
 
 namespace Envoy {
@@ -453,49 +454,43 @@ TEST_F(StringMatcher, NoMatcherRejected) {
       fmt::format("Configuration must define a matcher: {}", matcher.DebugString()));
 }
 
-// Validates the amount of memory that is being used by the different string
-// matchers. Requested as part of https://github.com/envoyproxy/envoy/pull/37782.
-TEST_F(StringMatcher, Memory) {
-  const uint32_t matchers_num = 1000;
-  // Prefix matcher.
-  {
-    // Add 1000 Prefix-String Matchers of varying string lengths (1 to 1000).
-    std::vector<Matchers::StringMatcherImpl> all_matchers;
-    all_matchers.reserve(matchers_num);
-    Memory::TestUtil::MemoryTest memory_test;
-    for (uint32_t i = 0; i < matchers_num; ++i) {
-      envoy::type::matcher::v3::StringMatcher matcher;
-      matcher.set_prefix(std::string(i + 1, 'a'));
-      all_matchers.emplace_back(Matchers::StringMatcherImpl(matcher, context_));
-    }
-    const size_t prefix_consumed_bytes = memory_test.consumedBytes();
-    // The memory constraints were added to ensure that the amount of memory
-    // used by matchers is carefully analyzed. These constraints can be relaxed
-    // when additional features are added, but it should be done in a thoughtful manner.
-    // Adding 3*8192 bytes because tcmalloc consumption estimation may return
-    // different values depending on memory alignment.
-    EXPECT_MEMORY_LE(prefix_consumed_bytes, 530176 + 3 * 8192);
-  }
-  // Regex matcher.
-  {
-    // Add 1000 Regex-String Matchers of varying string lengths (1 to 1000).
-    std::vector<Matchers::StringMatcherImpl> all_matchers;
-    all_matchers.reserve(matchers_num);
-    Memory::TestUtil::MemoryTest memory_test;
-    for (uint32_t i = 0; i < matchers_num; ++i) {
-      envoy::type::matcher::v3::StringMatcher matcher;
-      matcher.mutable_safe_regex()->mutable_google_re2();
-      matcher.mutable_safe_regex()->set_regex(std::string(i + 1, 'a'));
-      all_matchers.emplace_back(Matchers::StringMatcherImpl(matcher, context_));
-    }
-    const size_t regex_consumed_bytes = memory_test.consumedBytes();
-    // The memory constraints were added to ensure that the amount of memory
-    // used by matchers is carefully analyzed. These constraints can be relaxed
-    // when additional features are added, but it should be done in a thoughtful  manner.
-    // Adding 10*8192 bytes because tcmalloc consumption estimation may return
-    // different values depending on memory alignment.
-    EXPECT_MEMORY_LE(regex_consumed_bytes, 15038016 + 10 * 8192);
-  }
+MATCHER_P(MemNotMoreThan, sz,
+          "does not use more than " + std::to_string(sz) +
+              ": think carefully before increasing this, and if you're sure, "
+              "update the corresponding expectation") {
+  return arg <= sz;
+}
+
+// Validates the per-matcher memory footprint of the different string matchers.
+// Requested as part of https://github.com/envoyproxy/envoy/pull/37782: each
+// variant alternative should carry only the data it needs, and
+// StringMatcherImpl should not retain the proto used to construct it.
+//
+// Bounds are expressed in terms of sizeof(std::string) and sizeof(void*) so
+// they are portable across libc++, libstdc++, and 32/64-bit builds.
+TEST_F(StringMatcher, SizeIsBounded) {
+  // String-holding alternatives: one std::string + one bool rounded up to
+  // pointer alignment.
+  const size_t string_alt_bound = sizeof(std::string) + sizeof(void*);
+  EXPECT_THAT(sizeof(Matchers::ExactStringMatcher), MemNotMoreThan(string_alt_bound));
+  EXPECT_THAT(sizeof(Matchers::PrefixStringMatcher), MemNotMoreThan(string_alt_bound));
+  EXPECT_THAT(sizeof(Matchers::SuffixStringMatcher), MemNotMoreThan(string_alt_bound));
+  EXPECT_THAT(sizeof(Matchers::ContainsStringMatcher), MemNotMoreThan(string_alt_bound));
+
+  // Pointer-holding alternatives: a single unique_ptr.
+  const size_t ptr_alt_bound = 2 * sizeof(void*);
+  EXPECT_THAT(sizeof(Matchers::RegexStringMatcher), MemNotMoreThan(ptr_alt_bound));
+  EXPECT_THAT(sizeof(Matchers::CustomStringMatcher), MemNotMoreThan(ptr_alt_bound));
+
+  // StringMatcherImpl layout — accounting for all four pointer-sized
+  // contributions:
+  //   [1] vtable pointer for ValueMatcher base       (+1 * sizeof(void*))
+  //   [2] vtable pointer for StringMatcher base      (+1 * sizeof(void*))
+  //   [3] absl::variant payload = max(sizeof(alternatives))
+  //         = sizeof(std::string) + sizeof(void*) (string + padded bool)
+  //   [4] absl::variant discriminant                 (+1 * sizeof(void*))
+  EXPECT_THAT(sizeof(Matchers::StringMatcherImpl),
+              MemNotMoreThan(sizeof(std::string) + 4 * sizeof(void*)));
 }
 
 class PathMatcher : public BaseTest {};

test/common/upstream/load_stats_reporter_test.cc

@@ -36,6 +36,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) {
@@ -84,14 +95,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

@@ -1203,7 +1203,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.
@@ -1219,7 +1219,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.
@@ -1232,7 +1235,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/integration/BUILD

@@ -1571,12 +1571,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",
@@ -1628,9 +1625,8 @@ envoy_cc_test(
 
 envoy_cc_test(
     name = "load_stats_integration_test",
-    size = "large",
     srcs = ["load_stats_integration_test.cc"],
-    rbe_pool = "4core",
+    rbe_pool = "linux_x64_medium",
     deps = [
         ":http_integration_lib",
         "//test/config:utility_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/load_stats_integration_test.cc

@@ -227,6 +227,27 @@ class LoadStatsIntegrationTest : public Grpc::GrpcClientIntegrationParamTest,
               upstream_locality_stats->total_issued_requests() +
               local_upstream_locality_stats.total_issued_requests());
           // Unlike most stats, current requests in progress replaces old requests in progress.
+
+          // Merge load_metric_stats.
+          for (int k = 0; k < local_upstream_locality_stats.load_metric_stats_size(); ++k) {
+            const auto& local_metric = local_upstream_locality_stats.load_metric_stats(k);
+            bool found_metric = false;
+            for (int l = 0; l < upstream_locality_stats->load_metric_stats_size(); ++l) {
+              auto* metric = upstream_locality_stats->mutable_load_metric_stats(l);
+              if (metric->metric_name() == local_metric.metric_name()) {
+                found_metric = true;
+                metric->set_num_requests_finished_with_metric(
+                    metric->num_requests_finished_with_metric() +
+                    local_metric.num_requests_finished_with_metric());
+                metric->set_total_metric_value(metric->total_metric_value() +
+                                               local_metric.total_metric_value());
+                break;
+              }
+            }
+            if (!found_metric) {
+              upstream_locality_stats->add_load_metric_stats()->CopyFrom(local_metric);
+            }
+          }
           break;
         }
       }
@@ -454,7 +475,7 @@ class LoadStatsIntegrationTest : public Grpc::GrpcClientIntegrationParamTest,
 
   const uint64_t request_size_ = 1024;
   const uint64_t response_size_ = 512;
-  const uint32_t load_report_interval_ms_ = 500;
+  const uint32_t load_report_interval_ms_ = 1000;
 };
 
 INSTANTIATE_TEST_SUITE_P(IpVersionsClientType, LoadStatsIntegrationTest,