StatsAccessLogger: fixes connection gauge underflow crashes when decrementing metrics after Scope evictions.

source/common/stats/allocator_impl.cc

@@ -211,7 +211,6 @@ class GaugeImpl : public StatsSharedImpl<Gauge> {
   }
   void sub(uint64_t amount) override {
     ASSERT(child_value_ >= amount);
-    ASSERT(used() || amount == 0);
     child_value_ -= amount;
   }
   uint64_t value() const override { return child_value_ + parent_value_; }

source/common/stats/thread_local_store.cc

@@ -1117,15 +1117,21 @@ void ThreadLocalStoreImpl::evictUnused() {
         MetricBag metrics(scope->scope_id_);
         CentralCacheEntrySharedPtr& central_cache = scope->centralCacheMutableNoThreadAnalysis();
         auto filter_unused = []<typename T>(StatNameHashMap<T>& unused_metrics) {
-          return [&unused_metrics](std::pair<StatName, T> kv) {
+          return [&unused_metrics](const auto& kv) {
             const auto& [name, metric] = kv;
             if (metric->used()) {
               metric->markUnused();
               return false;
-            } else {
-              unused_metrics.try_emplace(name, metric);
-              return true;
             }
+            if constexpr (std::is_same_v<T, GaugeSharedPtr>) {
+              if (metric->value() != 0 &&
+                  (metric->importMode() == Gauge::ImportMode::Accumulate ||
+                   metric->importMode() == Gauge::ImportMode::HiddenAccumulate)) {
+                return false;
+              }
+            }
+            unused_metrics.try_emplace(name, metric);
+            return true;
           };
         };
         absl::erase_if(central_cache->counters_, filter_unused(metrics.counters_));

source/extensions/access_loggers/stats/stats.cc

@@ -15,64 +15,37 @@ namespace {
 
 class AccessLogState : public StreamInfo::FilterState::Object {
 public:
-  AccessLogState(Stats::ScopeSharedPtr scope) : scope_(std::move(scope)) {}
+  explicit AccessLogState(Stats::ScopeSharedPtr scope) : scope_(std::move(scope)) {}
 
-  // When the request is destroyed, we need to subtract the value from the gauge.
-  // We need to look up the gauge again in the scope because it might have been evicted.
-  // The gauge object itself is kept alive by the shared_ptr in the state, so we can access its
-  // name and tags to re-lookup/re-create it in the scope.
   ~AccessLogState() override {
-    for (const auto& [gauge_ptr, state] : inflight_gauges_) {
-      // TODO(taoxuy):  make this as an accessor of the
-      // Stat class.
-      Stats::StatNameTagVector tag_names;
-      state.gauge_->iterateTagStatNames(
-          [&tag_names](Stats::StatName name, Stats::StatName value) -> bool {
-            tag_names.emplace_back(name, value);
-            return true;
-          });
-
-      // Using state.gauge_->statName() directly would be incorrect because it returns the fully
-      // qualified name (including tags). Passing this full name to scope_->gaugeFromStatName(...)
-      // would cause the scope to attempt tag extraction on the full name. Since the tags in
-      // AccessLogState are often dynamic and not configured in the global tag extractors, this
-      // extraction would likely fail to identify the tags correctly, resulting in a gauge with a
-      // different identity (the full name as the stat name and no tags).
-      auto& gauge = scope_->gaugeFromStatNameWithTags(
-          state.gauge_->tagExtractedStatName(), tag_names, Stats::Gauge::ImportMode::Accumulate);
-      gauge.sub(state.value_);
+    for (const auto& [gauge, value] : inflight_gauges_) {
+      gauge->sub(value);
     }
   }
 
   void addInflightGauge(Stats::Gauge* gauge, uint64_t value) {
-    inflight_gauges_.try_emplace(gauge, Stats::GaugeSharedPtr(gauge), value);
+    auto [it, inserted] = inflight_gauges_.try_emplace(gauge, 0);
+    it->second += value;
   }
 
   absl::optional<uint64_t> removeInflightGauge(Stats::Gauge* gauge) {
     auto it = inflight_gauges_.find(gauge);
     if (it == inflight_gauges_.end()) {
       return absl::nullopt;
     }
-    uint64_t value = it->second.value_;
+    uint64_t value = it->second;
     inflight_gauges_.erase(it);
     return value;
   }
 
   static constexpr absl::string_view key() { return "envoy.access_loggers.stats.access_log_state"; }
 
 private:
-  struct State {
-    State(Stats::GaugeSharedPtr gauge, uint64_t value) : gauge_(std::move(gauge)), value_(value) {}
-
-    Stats::GaugeSharedPtr gauge_;
-    uint64_t value_;
-  };
-
   Stats::ScopeSharedPtr scope_;
 
-  // The map key holds a raw pointer to the gauge. The value holds a ref-counted pointer to ensure
-  // the gauge is not destroyed if it is evicted from the stats scope.
-  absl::flat_hash_map<Stats::Gauge*, State> inflight_gauges_;
+  // The map key holds a raw pointer to the gauge to enable fast O(1) lookups during
+  // PAIRED_SUBTRACT.
+  absl::flat_hash_map<Stats::Gauge*, uint64_t> inflight_gauges_;
 };
 
 Formatter::FormatterProviderPtr

test/common/stats/thread_local_store_test.cc

@@ -540,6 +540,9 @@ TEST_F(StatsThreadLocalStoreTest, StatsNumLimitsWithEviction) {
     EXPECT_FALSE(c1.used());
     EXPECT_FALSE(g1.used());
     EXPECT_FALSE(h1.used());
+
+    // Make gauge 0 so it can be evicted in the second pass.
+    g1.sub(1);
   }
 
   // Second eviction removes stats.
@@ -731,6 +734,10 @@ TEST_F(StatsThreadLocalStoreTest, Eviction) {
     EXPECT_EQ(1UL, store_->histograms().size());
   }
 
+  // Make gauge value zero so it can be evicted
+  Gauge& g1 = scope->gaugeFromString("g1", Gauge::ImportMode::Accumulate);
+  g1.sub(5);
+
   // Eviction removes here.
   EXPECT_CALL(tls_, runOnAllThreads(_, _)).Times(testing::AtLeast(1));
   store_->evictUnused();
@@ -783,6 +790,8 @@ TEST_F(StatsThreadLocalStoreTest, EvictionGaugesInterleavedOperations) {
   EXPECT_EQ(1UL, store_->gauges().size());
 
   // Second pass evicts from scope cache because it is now unused.
+  // We need to set it to 0 otherwise it won't be evicted.
+  g1.sub(10);
   EXPECT_CALL(tls_, runOnAllThreads(_, _)).Times(testing::AtLeast(1));
   store_->evictUnused();
 
@@ -795,7 +804,7 @@ TEST_F(StatsThreadLocalStoreTest, EvictionGaugesInterleavedOperations) {
 
   // 3. Interleaved PAIRED_ADD (add) on the held reference
   g1_ref->add(5);
-  EXPECT_EQ(15, g1_ref->value());
+  EXPECT_EQ(5, g1_ref->value());
   EXPECT_TRUE(g1_ref->used());
 
   // 4. Re-resolve and PAIRED_SUBTRACT (sub)
@@ -805,17 +814,66 @@ TEST_F(StatsThreadLocalStoreTest, EvictionGaugesInterleavedOperations) {
   EXPECT_EQ(g1_ref.get(), &g1_resurrected);
 
   // Value should be preserved
-  EXPECT_EQ(15, g1_resurrected.value());
+  EXPECT_EQ(5, g1_resurrected.value());
 
   // Perform subtract
-  g1_resurrected.sub(15);
+  g1_resurrected.sub(5);
   EXPECT_EQ(0, g1_resurrected.value());
 
   tls_.shutdownGlobalThreading();
   store_->shutdownThreading();
   tls_.shutdownThread();
 }
 
+TEST_F(StatsThreadLocalStoreTest, EvictionGaugesNonZero) {
+  InSequence s;
+  store_->initializeThreading(main_thread_dispatcher_, tls_);
+
+  ScopeSharedPtr scope = store_->rootScope()->createScope("scope.", /*evictable=*/true);
+
+  // 1. Create gauge and add to make it non-zero
+  Gauge& g1 = scope->gaugeFromString("g1", Gauge::ImportMode::Accumulate);
+  g1.add(10);
+  EXPECT_EQ(10, g1.value());
+  EXPECT_TRUE(g1.used());
+
+  // First pass marks unused. Note that evictUnused() only removes if it was ALREADY unused.
+  store_->evictUnused();
+  EXPECT_FALSE(g1.used());
+  EXPECT_EQ(1UL, store_->gauges().size());
+
+  // Second pass would normally evict from scope cache if it was zero, but since it's non-zero,
+  // it should stay.
+  store_->evictUnused();
+
+  // Verify STILL in scope
+  StatNameManagedStorage g1_name("scope.g1", symbol_table_);
+  auto found_gauge = scope->findGauge(g1_name.statName());
+  ASSERT_TRUE(found_gauge.has_value());
+  EXPECT_EQ(&g1, &(found_gauge.value().get()));
+
+  // Verify still in store (allocator)
+  EXPECT_EQ(1UL, store_->gauges().size());
+
+  // 2. Now subtract to make it zero and evict
+  g1.sub(10);
+  EXPECT_EQ(0, g1.value());
+
+  // First pass actually evicts it here because it is already unused
+  EXPECT_CALL(tls_, runOnAllThreads(_, _)).Times(testing::AtLeast(1));
+  store_->evictUnused();
+
+  // Verify removed from scope
+  EXPECT_FALSE(scope->findGauge(g1_name.statName()).has_value());
+
+  // Verify removed from store
+  EXPECT_EQ(0UL, store_->gauges().size());
+
+  tls_.shutdownGlobalThreading();
+  store_->shutdownThreading();
+  tls_.shutdownThread();
+}
+
 TEST_F(StatsThreadLocalStoreTest, NestedScopes) {
   InSequence s;
   store_->initializeThreading(main_thread_dispatcher_, tls_);

test/extensions/access_loggers/stats/integration_test.cc

@@ -279,5 +279,75 @@ TEST_P(StatsAccessLogIntegrationTest, SubtractWithoutAdd) {
   test_server_->waitForGaugeEq("test_stat_prefix.active_requests.request_header_tag.my-tag", 0);
 }
 
+TEST_P(StatsAccessLogIntegrationTest, ActiveRequestsGaugeEvictionResetsValueIfUnprotected) {
+  const std::string config_yaml = R"(
+              name: envoy.access_loggers.stats
+              typed_config:
+                "@type": type.googleapis.com/envoy.extensions.access_loggers.stats.v3.Config
+                stat_prefix: test_stat_prefix
+                gauges:
+                  - stat:
+                      name: active_requests
+                      tags:
+                        - name: request_header_tag
+                          value_format: '%REQUEST_HEADER(tag-value)%'
+                    value_fixed: 1
+                    add_subtract:
+                      add_log_type: DownstreamStart
+                      sub_log_type: DownstreamEnd
+)";
+
+  init(config_yaml, /*autonomous_upstream=*/false,
+       /*flush_access_log_on_new_request=*/true);
+
+  Http::TestRequestHeaderMapImpl request_headers{
+      {":method", "GET"},  {":authority", "envoyproxy.io"},       {":path", "/"},
+      {":scheme", "http"}, {"tag-value", "my-eviction-test-tag"},
+  };
+
+  // Request 1: starts gauge at 1.
+  auto codec_client1 = makeHttpConnection(lookupPort("http"));
+  auto response1 = codec_client1->makeHeaderOnlyRequest(request_headers);
+  waitForNextUpstreamRequest();
+  test_server_->waitForGaugeEq(
+      "test_stat_prefix.active_requests.request_header_tag.my-eviction-test-tag", 1);
+
+  // Simulate eviction from the store.
+  absl::Notification evict_done;
+  test_server_->server().dispatcher().post([this, &evict_done]() {
+    test_server_->statStore().evictUnused();
+    test_server_->statStore().evictUnused();
+    evict_done.Notify();
+  });
+  evict_done.WaitForNotification();
+
+  // Request 2: starts another concurrent request using the same tag.
+  auto codec_client2 = makeHttpConnection(lookupPort("http"));
+  auto response2 = codec_client2->makeHeaderOnlyRequest(request_headers);
+
+  // Wait for the second request to reach upstream.
+  // We need to keep track of the second upstream request.
+  FakeStreamPtr upstream_request2;
+  FakeHttpConnectionPtr fake_upstream_connection2;
+  ASSERT_TRUE(fake_upstreams_[0]->waitForHttpConnection(*dispatcher_, fake_upstream_connection2));
+  ASSERT_TRUE(fake_upstream_connection2->waitForNewStream(*dispatcher_, upstream_request2));
+  ASSERT_TRUE(upstream_request2->waitForEndStream(*dispatcher_));
+
+  // The gauge should be kept even with eviction happened and the active request is 2.
+  test_server_->waitForGaugeEq(
+      "test_stat_prefix.active_requests.request_header_tag.my-eviction-test-tag", 2);
+
+  // Clean up.
+  Http::TestResponseHeaderMapImpl response_headers{{":status", "200"}};
+  upstream_request_->encodeHeaders(response_headers, true);
+  ASSERT_TRUE(response1->waitForEndStream());
+  test_server_->waitForGaugeEq(
+      "test_stat_prefix.active_requests.request_header_tag.my-eviction-test-tag", 1);
+  upstream_request2->encodeHeaders(response_headers, true);
+  ASSERT_TRUE(response2->waitForEndStream());
+  test_server_->waitForGaugeEq(
+      "test_stat_prefix.active_requests.request_header_tag.my-eviction-test-tag", 0);
+}
+
 } // namespace
 } // namespace Envoy

test/extensions/access_loggers/stats/stats_test.cc

@@ -100,6 +100,7 @@ class StatsAccessLoggerTest : public testing::Test {
   void initialize(const envoy::extensions::access_loggers::stats::v3::Config& config) {
     auto* gauge = new NiceMock<MockGaugeWithTags>();
     gauge_ = gauge;
+    ON_CALL(*gauge_, value()).WillByDefault(testing::Return(10));
     gauge_ptr_ = Stats::GaugeSharedPtr(gauge_);
     gauge_->name_ = "gauge";
     gauge_->setTagExtractedName("gauge");
@@ -637,18 +638,21 @@ TEST_F(StatsAccessLoggerTest, DestructionSubtractsRemainingValue) {
 
   NiceMock<StreamInfo::MockStreamInfo> local_stream_info;
 
-  // Called once on log() and once on destruction.
-  EXPECT_CALL(store_, gauge(_, Stats::Gauge::ImportMode::Accumulate)).Times(2);
+  // Called once on log().
+  EXPECT_CALL(store_, gauge(_, Stats::Gauge::ImportMode::Accumulate));
   EXPECT_CALL(*gauge_, add(10));
   logger_->log(formatter_context_, local_stream_info);
 
   // Expect subtraction on destruction
   EXPECT_CALL(*gauge_, sub(10));
 
+  // Destroy logger before stream_info to simulate logger config deletion while stream is active
+  logger_.reset();
+
   // local_stream_info goes out of scope here.
 }
 
-TEST_F(StatsAccessLoggerTest, AccessLogStateDestructorReconstructsGauge) {
+TEST_F(StatsAccessLoggerTest, AccessLogStateDestructorSubtractsFromSavedGauge) {
   const std::string yaml = R"EOF(
     stat_prefix: test_stat_prefix
     gauges:
@@ -680,44 +684,28 @@ TEST_F(StatsAccessLoggerTest, AccessLogStateDestructorReconstructsGauge) {
 
   // Initial lookup and add
   EXPECT_CALL(*mock_scope, gaugeFromStatNameWithTags(_, _, Stats::Gauge::ImportMode::Accumulate))
-      .WillOnce(Invoke([&](const Stats::StatName& name, Stats::StatNameTagVectorOptConstRef tags,
-                           Stats::Gauge::ImportMode) -> Stats::Gauge& {
-        saved_name = name;
-        if (tags) {
-          for (const auto& tag : tags->get()) {
-            saved_tags_strs.emplace_back(store_.symbolTable().toString(tag.first),
-                                         store_.symbolTable().toString(tag.second));
-          }
-        }
-        EXPECT_FALSE(saved_tags_strs.empty());
-        auto* gauge_with_tags = dynamic_cast<MockGaugeWithTags*>(gauge_);
-        EXPECT_TRUE(gauge_with_tags != nullptr);
-        gauge_with_tags->setTags(tags->get(), store_.symbolTable());
-        return *gauge_;
-      }));
+      .WillRepeatedly(
+          Invoke([&](const Stats::StatName& name, Stats::StatNameTagVectorOptConstRef tags,
+                     Stats::Gauge::ImportMode) -> Stats::Gauge& {
+            saved_name = name;
+            if (tags) {
+              for (const auto& tag : tags->get()) {
+                saved_tags_strs.emplace_back(store_.symbolTable().toString(tag.first),
+                                             store_.symbolTable().toString(tag.second));
+              }
+              EXPECT_FALSE(saved_tags_strs.empty());
+              auto* gauge_with_tags = dynamic_cast<MockGaugeWithTags*>(gauge_);
+              EXPECT_TRUE(gauge_with_tags != nullptr);
+              gauge_with_tags->setTags(tags->get(), store_.symbolTable());
+            }
+            return *gauge_;
+          }));
+
   EXPECT_CALL(*gauge_, add(10));
   logger_->log(formatter_context_, local_stream_info);
 
-  // Simulate eviction from scope (or just verify lookup happens again)
-  // The destructor of AccessLogState should call gaugeFromStatNameWithTags again.
-  EXPECT_CALL(*mock_scope, gaugeFromStatNameWithTags(_, _, Stats::Gauge::ImportMode::Accumulate))
-      .WillOnce(Invoke([&](const Stats::StatName& name, Stats::StatNameTagVectorOptConstRef tags,
-                           Stats::Gauge::ImportMode) -> Stats::Gauge& {
-        EXPECT_EQ(name, saved_name);
-        EXPECT_TRUE(tags.has_value());
-        if (tags) {
-          const auto& tags_vec = tags->get();
-          // Detailed comparison
-          EXPECT_EQ(tags_vec.size(), 2);
-          if (tags_vec.size() == 2) {
-            EXPECT_EQ(store_.symbolTable().toString(tags_vec[0].first), "tag_name");
-            EXPECT_EQ(store_.symbolTable().toString(tags_vec[0].second), "200");
-            EXPECT_EQ(store_.symbolTable().toString(tags_vec[1].first), "another_tag");
-            EXPECT_EQ(store_.symbolTable().toString(tags_vec[1].second), "value_fixed");
-          }
-        }
-        return *gauge_;
-      }));
+  // The destructor of AccessLogState should call sub(10) directly on the saved gauge
+  // This will trigger a second lookup using gaugeFromString (tags == absl::nullopt).
   EXPECT_CALL(*gauge_, sub(10));
 
   // local_stream_info goes out of scope here, triggering AccessLogState destructor.