diff --git a/envoy/stats/allocator.h b/envoy/stats/allocator.h
index f6181553ad851..6f9cc9715ea43 100644
--- a/envoy/stats/allocator.h
+++ b/envoy/stats/allocator.h
@@ -58,6 +58,31 @@ class Allocator {
   virtual const SymbolTable& constSymbolTable() const PURE;
   virtual SymbolTable& symbolTable() PURE;
 
+  /**
+   * Mark rejected stats as deleted by moving them to a different vector, so they don't show up
+   * when iterating over stats, but prevent crashes when trying to access references to them.
+   * Note that allocating a stat with the same name after calling this will
+   * return a new stat. Hence callers should seek to avoid this situation, as is
+   * done in ThreadLocalStore.
+   */
+  virtual void markCounterForDeletion(const CounterSharedPtr& counter) PURE;
+  virtual void markGaugeForDeletion(const GaugeSharedPtr& gauge) PURE;
+  virtual void markTextReadoutForDeletion(const TextReadoutSharedPtr& text_readout) PURE;
+
+  /**
+   * Iterate over all stats that need to be sinked. Note, that implementations can potentially hold
+   * on to a mutex that will deadlock if the passed in functors try to create or delete a stat.
+   * @param f_size functor that is provided the number of all sinked stats. Note this is called
+   *        only once, prior to any calls to f_stat.
+   * @param f_stat functor that is provided one sinked stat at a time.
+   */
+  virtual void forEachCounter(std::function<void(std::size_t)> f_size,
+                              std::function<void(Stats::Counter&)> f_stat) const PURE;
+  virtual void forEachGauge(std::function<void(std::size_t)> f_size,
+                            std::function<void(Stats::Gauge&)> f_stat) const PURE;
+  virtual void forEachTextReadout(std::function<void(std::size_t)> f_size,
+                                  std::function<void(Stats::TextReadout&)> f_stat) const PURE;
+
   // TODO(jmarantz): create a parallel mechanism to instantiate histograms. At
   // the moment, histograms don't fit the same pattern of counters and gauges
   // as they are not actually created in the context of a stats allocator.
diff --git a/envoy/stats/store.h b/envoy/stats/store.h
index 191ed0f8589c9..a682fb0cd3d5f 100644
--- a/envoy/stats/store.h
+++ b/envoy/stats/store.h
@@ -7,6 +7,7 @@
 #include "envoy/common/pure.h"
 #include "envoy/stats/histogram.h"
 #include "envoy/stats/scope.h"
+#include "envoy/stats/stats.h"
 #include "envoy/stats/stats_matcher.h"
 #include "envoy/stats/tag_producer.h"
 
@@ -48,6 +49,21 @@ class Store : public Scope {
    * @return a list of all known histograms.
    */
   virtual std::vector<ParentHistogramSharedPtr> histograms() const PURE;
+
+  /**
+   * Iterate over all stats that need to be sinked. Note, that implementations can potentially  hold
+   * on to a mutex that will deadlock if the passed in functors try to create or delete a stat.
+   * @param f_size functor that is provided the number of all sinked stats.
+   * @param f_stat functor that is provided one sinked stat at a time.
+   */
+  virtual void forEachCounter(std::function<void(std::size_t)> f_size,
+                              std::function<void(Stats::Counter&)> f_stat) const PURE;
+
+  virtual void forEachGauge(std::function<void(std::size_t)> f_size,
+                            std::function<void(Stats::Gauge&)> f_stat) const PURE;
+
+  virtual void forEachTextReadout(std::function<void(std::size_t)> f_size,
+                                  std::function<void(Stats::TextReadout&)> f_stat) const PURE;
 };
 
 using StorePtr = std::unique_ptr<Store>;
diff --git a/source/common/stats/allocator_impl.cc b/source/common/stats/allocator_impl.cc
index 3c64b6ba49511..4464f41a344e6 100644
--- a/source/common/stats/allocator_impl.cc
+++ b/source/common/stats/allocator_impl.cc
@@ -1,5 +1,6 @@
 #include "source/common/stats/allocator_impl.h"
 
+#include <algorithm>
 #include <cstdint>
 
 #include "envoy/stats/stats.h"
@@ -25,6 +26,25 @@ const char AllocatorImpl::DecrementToZeroSyncPoint[] = "decrement-zero";
 AllocatorImpl::~AllocatorImpl() {
   ASSERT(counters_.empty());
   ASSERT(gauges_.empty());
+
+#ifndef NDEBUG
+  // Move deleted stats into the sets for the ASSERTs in removeFromSetLockHeld to function.
+  for (auto& counter : deleted_counters_) {
+    auto insertion = counters_.insert(counter.get());
+    // Assert that there were no duplicates.
+    ASSERT(insertion.second);
+  }
+  for (auto& gauge : deleted_gauges_) {
+    auto insertion = gauges_.insert(gauge.get());
+    // Assert that there were no duplicates.
+    ASSERT(insertion.second);
+  }
+  for (auto& text_readout : deleted_text_readouts_) {
+    auto insertion = text_readouts_.insert(text_readout.get());
+    // Assert that there were no duplicates.
+    ASSERT(insertion.second);
+  }
+#endif
 }
 
 #ifndef ENVOY_CONFIG_COVERAGE
@@ -316,5 +336,77 @@ Counter* AllocatorImpl::makeCounterInternal(StatName name, StatName tag_extracte
   return new CounterImpl(name, *this, tag_extracted_name, stat_name_tags);
 }
 
+void AllocatorImpl::forEachCounter(std::function<void(std::size_t)> f_size,
+                                   std::function<void(Stats::Counter&)> f_stat) const {
+  Thread::LockGuard lock(mutex_);
+  if (f_size != nullptr) {
+    f_size(counters_.size());
+  }
+  for (auto& counter : counters_) {
+    f_stat(*counter);
+  }
+}
+
+void AllocatorImpl::forEachGauge(std::function<void(std::size_t)> f_size,
+                                 std::function<void(Stats::Gauge&)> f_stat) const {
+  Thread::LockGuard lock(mutex_);
+  if (f_size != nullptr) {
+    f_size(gauges_.size());
+  }
+  for (auto& gauge : gauges_) {
+    f_stat(*gauge);
+  }
+}
+
+void AllocatorImpl::forEachTextReadout(std::function<void(std::size_t)> f_size,
+                                       std::function<void(Stats::TextReadout&)> f_stat) const {
+  Thread::LockGuard lock(mutex_);
+  if (f_size != nullptr) {
+    f_size(text_readouts_.size());
+  }
+  for (auto& text_readout : text_readouts_) {
+    f_stat(*text_readout);
+  }
+}
+
+void AllocatorImpl::markCounterForDeletion(const CounterSharedPtr& counter) {
+  Thread::LockGuard lock(mutex_);
+  auto iter = counters_.find(counter->statName());
+  if (iter == counters_.end()) {
+    // This has already been marked for deletion.
+    return;
+  }
+  ASSERT(counter.get() == *iter);
+  // Duplicates are ASSERTed in ~AllocatorImpl.
+  deleted_counters_.emplace_back(*iter);
+  counters_.erase(iter);
+}
+
+void AllocatorImpl::markGaugeForDeletion(const GaugeSharedPtr& gauge) {
+  Thread::LockGuard lock(mutex_);
+  auto iter = gauges_.find(gauge->statName());
+  if (iter == gauges_.end()) {
+    // This has already been marked for deletion.
+    return;
+  }
+  ASSERT(gauge.get() == *iter);
+  // Duplicates are ASSERTed in ~AllocatorImpl.
+  deleted_gauges_.emplace_back(*iter);
+  gauges_.erase(iter);
+}
+
+void AllocatorImpl::markTextReadoutForDeletion(const TextReadoutSharedPtr& text_readout) {
+  Thread::LockGuard lock(mutex_);
+  auto iter = text_readouts_.find(text_readout->statName());
+  if (iter == text_readouts_.end()) {
+    // This has already been marked for deletion.
+    return;
+  }
+  ASSERT(text_readout.get() == *iter);
+  // Duplicates are ASSERTed in ~AllocatorImpl.
+  deleted_text_readouts_.emplace_back(*iter);
+  text_readouts_.erase(iter);
+}
+
 } // namespace Stats
 } // namespace Envoy
diff --git a/source/common/stats/allocator_impl.h b/source/common/stats/allocator_impl.h
index db656e2e40134..806e7dc8612fd 100644
--- a/source/common/stats/allocator_impl.h
+++ b/source/common/stats/allocator_impl.h
@@ -33,6 +33,15 @@ class AllocatorImpl : public Allocator {
   SymbolTable& symbolTable() override { return symbol_table_; }
   const SymbolTable& constSymbolTable() const override { return symbol_table_; }
 
+  void forEachCounter(std::function<void(std::size_t)>,
+                      std::function<void(Stats::Counter&)>) const override;
+
+  void forEachGauge(std::function<void(std::size_t)>,
+                    std::function<void(Stats::Gauge&)>) const override;
+
+  void forEachTextReadout(std::function<void(std::size_t)>,
+                          std::function<void(Stats::TextReadout&)>) const override;
+
 #ifndef ENVOY_CONFIG_COVERAGE
   void debugPrint();
 #endif
@@ -47,6 +56,10 @@ class AllocatorImpl : public Allocator {
    */
   bool isMutexLockedForTest();
 
+  void markCounterForDeletion(const CounterSharedPtr& counter) override;
+  void markGaugeForDeletion(const GaugeSharedPtr& gauge) override;
+  void markTextReadoutForDeletion(const TextReadoutSharedPtr& text_readout) override;
+
 protected:
   virtual Counter* makeCounterInternal(StatName name, StatName tag_extracted_name,
                                        const StatNameTagVector& stat_name_tags);
@@ -58,21 +71,29 @@ class AllocatorImpl : public Allocator {
   friend class TextReadoutImpl;
   friend class NotifyingAllocatorImpl;
 
-  void removeCounterFromSetLockHeld(Counter* counter) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
-  void removeGaugeFromSetLockHeld(Gauge* gauge) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
-  void removeTextReadoutFromSetLockHeld(Counter* counter) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+  // A mutex is needed here to protect both the stats_ object from both
+  // alloc() and free() operations. Although alloc() operations are called under existing locking,
+  // free() operations are made from the destructors of the individual stat objects, which are not
+  // protected by locks.
+  mutable Thread::MutexBasicLockable mutex_;
 
   StatSet<Counter> counters_ ABSL_GUARDED_BY(mutex_);
   StatSet<Gauge> gauges_ ABSL_GUARDED_BY(mutex_);
   StatSet<TextReadout> text_readouts_ ABSL_GUARDED_BY(mutex_);
 
-  SymbolTable& symbol_table_;
+  // Retain storage for deleted stats; these are no longer in maps because
+  // the matcher-pattern was established after they were created. Since the
+  // stats are held by reference in code that expects them to be there, we
+  // can't actually delete the stats.
+  //
+  // It seems like it would be better to have each client that expects a stat
+  // to exist to hold it as (e.g.) a CounterSharedPtr rather than a Counter&
+  // but that would be fairly complex to change.
+  std::vector<CounterSharedPtr> deleted_counters_ ABSL_GUARDED_BY(mutex_);
+  std::vector<GaugeSharedPtr> deleted_gauges_ ABSL_GUARDED_BY(mutex_);
+  std::vector<TextReadoutSharedPtr> deleted_text_readouts_ ABSL_GUARDED_BY(mutex_);
 
-  // A mutex is needed here to protect both the stats_ object from both
-  // alloc() and free() operations. Although alloc() operations are called under existing locking,
-  // free() operations are made from the destructors of the individual stat objects, which are not
-  // protected by locks.
-  Thread::MutexBasicLockable mutex_;
+  SymbolTable& symbol_table_;
 
   Thread::ThreadSynchronizer sync_;
 };
diff --git a/source/common/stats/isolated_store_impl.h b/source/common/stats/isolated_store_impl.h
index f0f924ebda541..ebff944da7eff 100644
--- a/source/common/stats/isolated_store_impl.h
+++ b/source/common/stats/isolated_store_impl.h
@@ -2,6 +2,7 @@
 
 #include <algorithm>
 #include <cstring>
+#include <functional>
 #include <string>
 
 #include "envoy/stats/stats.h"
@@ -100,6 +101,14 @@ template <class Base> class IsolatedStatsCache {
     return true;
   }
 
+  void forEachStat(std::function<void(std::size_t)> f_size,
+                   std::function<void(Base&)> f_stat) const {
+    f_size(stats_.size());
+    for (auto const& stat : stats_) {
+      f_stat(*stat.second);
+    }
+  }
+
 private:
   friend class IsolatedStoreImpl;
 
@@ -205,6 +214,21 @@ class IsolatedStoreImpl : public StoreImpl {
     return textReadoutFromStatName(storage.statName());
   }
 
+  void forEachCounter(std::function<void(std::size_t)> f_size,
+                      std::function<void(Stats::Counter&)> f_stat) const override {
+    counters_.forEachStat(f_size, f_stat);
+  }
+
+  void forEachGauge(std::function<void(std::size_t)> f_size,
+                    std::function<void(Stats::Gauge&)> f_stat) const override {
+    gauges_.forEachStat(f_size, f_stat);
+  }
+
+  void forEachTextReadout(std::function<void(std::size_t)> f_size,
+                          std::function<void(Stats::TextReadout&)> f_stat) const override {
+    text_readouts_.forEachStat(f_size, f_stat);
+  }
+
 private:
   IsolatedStoreImpl(std::unique_ptr<SymbolTable>&& symbol_table);
 
diff --git a/source/common/stats/thread_local_store.cc b/source/common/stats/thread_local_store.cc
index 3b29f8b5d4df5..0e88476e5e47f 100644
--- a/source/common/stats/thread_local_store.cc
+++ b/source/common/stats/thread_local_store.cc
@@ -69,10 +69,19 @@ void ThreadLocalStoreImpl::setStatsMatcher(StatsMatcherPtr&& stats_matcher) {
   Thread::LockGuard lock(lock_);
   const uint32_t first_histogram_index = deleted_histograms_.size();
   for (ScopeImpl* scope : scopes_) {
-    removeRejectedStats(scope->central_cache_->counters_, deleted_counters_);
-    removeRejectedStats(scope->central_cache_->gauges_, deleted_gauges_);
+    removeRejectedStats<CounterSharedPtr>(scope->central_cache_->counters_,
+                                          [this](const CounterSharedPtr& counter) mutable {
+                                            alloc_.markCounterForDeletion(counter);
+                                          });
+    removeRejectedStats<GaugeSharedPtr>(
+        scope->central_cache_->gauges_,
+        [this](const GaugeSharedPtr& gauge) mutable { alloc_.markGaugeForDeletion(gauge); });
     removeRejectedStats(scope->central_cache_->histograms_, deleted_histograms_);
-    removeRejectedStats(scope->central_cache_->text_readouts_, deleted_text_readouts_);
+    removeRejectedStats<TextReadoutSharedPtr>(
+        scope->central_cache_->text_readouts_,
+        [this](const TextReadoutSharedPtr& text_readout) mutable {
+          alloc_.markTextReadoutForDeletion(text_readout);
+        });
   }
 
   // Remove any newly rejected histograms from histogram_set_.
@@ -101,6 +110,23 @@ void ThreadLocalStoreImpl::removeRejectedStats(StatMapClass& map, StatListClass&
   }
 }
 
+template <class StatSharedPtr>
+void ThreadLocalStoreImpl::removeRejectedStats(
+    StatNameHashMap<StatSharedPtr>& map, std::function<void(const StatSharedPtr&)> f_deletion) {
+  StatNameVec remove_list;
+  for (auto& stat : map) {
+    if (rejects(stat.first)) {
+      remove_list.push_back(stat.first);
+    }
+  }
+  for (StatName stat_name : remove_list) {
+    auto iter = map.find(stat_name);
+    ASSERT(iter != map.end());
+    f_deletion(iter->second);
+    map.erase(iter);
+  }
+}
+
 StatsMatcher::FastResult ThreadLocalStoreImpl::fastRejects(StatName stat_name) const {
   return stats_matcher_->fastRejects(stat_name);
 }
@@ -113,16 +139,9 @@ bool ThreadLocalStoreImpl::slowRejects(StatsMatcher::FastResult fast_reject_resu
 std::vector<CounterSharedPtr> ThreadLocalStoreImpl::counters() const {
   // Handle de-dup due to overlapping scopes.
   std::vector<CounterSharedPtr> ret;
-  StatNameHashSet names;
-  Thread::LockGuard lock(lock_);
-  for (ScopeImpl* scope : scopes_) {
-    for (auto& counter : scope->central_cache_->counters_) {
-      if (names.insert(counter.first).second) {
-        ret.push_back(counter.second);
-      }
-    }
-  }
-
+  forEachCounter(
+      [&ret](std::size_t size) mutable { ret.reserve(size); },
+      [&ret](Counter& counter) mutable { ret.emplace_back(CounterSharedPtr(&counter)); });
   return ret;
 }
 
@@ -141,34 +160,22 @@ ScopePtr ThreadLocalStoreImpl::scopeFromStatName(StatName name) {
 std::vector<GaugeSharedPtr> ThreadLocalStoreImpl::gauges() const {
   // Handle de-dup due to overlapping scopes.
   std::vector<GaugeSharedPtr> ret;
-  StatNameHashSet names;
-  Thread::LockGuard lock(lock_);
-  for (ScopeImpl* scope : scopes_) {
-    for (auto& gauge_iter : scope->central_cache_->gauges_) {
-      const GaugeSharedPtr& gauge = gauge_iter.second;
-      if (gauge->importMode() != Gauge::ImportMode::Uninitialized &&
-          names.insert(gauge_iter.first).second) {
-        ret.push_back(gauge);
-      }
-    }
-  }
-
+  forEachGauge([&ret](std::size_t size) mutable { ret.reserve(size); },
+               [&ret](Gauge& gauge) mutable {
+                 if (gauge.importMode() != Gauge::ImportMode::Uninitialized) {
+                   ret.emplace_back(GaugeSharedPtr(&gauge));
+                 }
+               });
   return ret;
 }
 
 std::vector<TextReadoutSharedPtr> ThreadLocalStoreImpl::textReadouts() const {
   // Handle de-dup due to overlapping scopes.
   std::vector<TextReadoutSharedPtr> ret;
-  StatNameHashSet names;
-  Thread::LockGuard lock(lock_);
-  for (ScopeImpl* scope : scopes_) {
-    for (auto& text_readout : scope->central_cache_->text_readouts_) {
-      if (names.insert(text_readout.first).second) {
-        ret.push_back(text_readout.second);
-      }
-    }
-  }
-
+  forEachTextReadout([&ret](std::size_t size) mutable { ret.reserve(size); },
+                     [&ret](TextReadout& text_readout) mutable {
+                       ret.emplace_back(TextReadoutSharedPtr(&text_readout));
+                     });
   return ret;
 }
 
@@ -975,5 +982,24 @@ bool ParentHistogramImpl::usedLockHeld() const {
   return false;
 }
 
+void ThreadLocalStoreImpl::forEachCounter(std::function<void(std::size_t)> f_size,
+                                          std::function<void(Stats::Counter&)> f_stat) const {
+  Thread::LockGuard lock(lock_);
+  alloc_.forEachCounter(f_size, f_stat);
+}
+
+void ThreadLocalStoreImpl::forEachGauge(std::function<void(std::size_t)> f_size,
+                                        std::function<void(Stats::Gauge&)> f_stat) const {
+  Thread::LockGuard lock(lock_);
+  alloc_.forEachGauge(f_size, f_stat);
+}
+
+void ThreadLocalStoreImpl::forEachTextReadout(
+    std::function<void(std::size_t)> f_size,
+    std::function<void(Stats::TextReadout&)> f_stat) const {
+  Thread::LockGuard lock(lock_);
+  alloc_.forEachTextReadout(f_size, f_stat);
+}
+
 } // namespace Stats
 } // namespace Envoy
diff --git a/source/common/stats/thread_local_store.h b/source/common/stats/thread_local_store.h
index eaf2946fd99f7..742e1fc3c04d6 100644
--- a/source/common/stats/thread_local_store.h
+++ b/source/common/stats/thread_local_store.h
@@ -244,6 +244,15 @@ class ThreadLocalStoreImpl : Logger::Loggable<Logger::Id::stats>, public StoreRo
   std::vector<TextReadoutSharedPtr> textReadouts() const override;
   std::vector<ParentHistogramSharedPtr> histograms() const override;
 
+  void forEachCounter(std::function<void(std::size_t)> f_size,
+                      std::function<void(Stats::Counter&)> f_stat) const override;
+
+  void forEachGauge(std::function<void(std::size_t)> f_size,
+                    std::function<void(Stats::Gauge&)> f_stat) const override;
+
+  void forEachTextReadout(std::function<void(std::size_t)> f_size,
+                          std::function<void(Stats::TextReadout&)> f_stat) const override;
+
   // Stats::StoreRoot
   void addSink(Sink& sink) override { timer_sinks_.push_back(sink); }
   void setTagProducer(TagProducerPtr&& tag_producer) override {
@@ -483,6 +492,9 @@ class ThreadLocalStoreImpl : Logger::Loggable<Logger::Id::stats>, public StoreRo
   bool rejectsAll() const { return stats_matcher_->rejectsAll(); }
   template <class StatMapClass, class StatListClass>
   void removeRejectedStats(StatMapClass& map, StatListClass& list);
+  template <class StatSharedPtr>
+  void removeRejectedStats(StatNameHashMap<StatSharedPtr>& map,
+                           std::function<void(const StatSharedPtr&)> f_deletion);
   bool checkAndRememberRejection(StatName name, StatsMatcher::FastResult fast_reject_result,
                                  StatNameStorageSet& central_rejected_stats,
                                  StatNameHashSet* tls_rejected_stats);
@@ -527,10 +539,7 @@ class ThreadLocalStoreImpl : Logger::Loggable<Logger::Id::stats>, public StoreRo
   // It seems like it would be better to have each client that expects a stat
   // to exist to hold it as (e.g.) a CounterSharedPtr rather than a Counter&
   // but that would be fairly complex to change.
-  std::vector<CounterSharedPtr> deleted_counters_ ABSL_GUARDED_BY(lock_);
-  std::vector<GaugeSharedPtr> deleted_gauges_ ABSL_GUARDED_BY(lock_);
   std::vector<HistogramSharedPtr> deleted_histograms_ ABSL_GUARDED_BY(lock_);
-  std::vector<TextReadoutSharedPtr> deleted_text_readouts_ ABSL_GUARDED_BY(lock_);
 
   // Scope IDs and central cache entries that are queued for cross-scope release.
   // Because there can be a large number of scopes, all of which are released at once,
diff --git a/source/server/server.cc b/source/server/server.cc
index 872e4d54b2a3a..152058301be54 100644
--- a/source/server/server.cc
+++ b/source/server/server.cc
@@ -20,6 +20,7 @@
 #include "envoy/server/bootstrap_extension_config.h"
 #include "envoy/server/instance.h"
 #include "envoy/server/options.h"
+#include "envoy/stats/stats.h"
 #include "envoy/upstream/cluster_manager.h"
 
 #include "source/common/api/api_impl.h"
@@ -165,18 +166,16 @@ void InstanceImpl::failHealthcheck(bool fail) {
 }
 
 MetricSnapshotImpl::MetricSnapshotImpl(Stats::Store& store, TimeSource& time_source) {
-  snapped_counters_ = store.counters();
-  counters_.reserve(snapped_counters_.size());
-  for (const auto& counter : snapped_counters_) {
-    counters_.push_back({counter->latch(), *counter});
-  }
+  store.forEachCounter([this](std::size_t size) mutable { counters_.reserve(size); },
+                       [this](Stats::Counter& counter) mutable {
+                         counters_.push_back({counter.latch(), counter});
+                       });
 
-  snapped_gauges_ = store.gauges();
-  gauges_.reserve(snapped_gauges_.size());
-  for (const auto& gauge : snapped_gauges_) {
-    ASSERT(gauge->importMode() != Stats::Gauge::ImportMode::Uninitialized);
-    gauges_.push_back(*gauge);
-  }
+  store.forEachGauge([this](std::size_t size) mutable { gauges_.reserve(size); },
+                     [this](Stats::Gauge& gauge) mutable {
+                       ASSERT(gauge.importMode() != Stats::Gauge::ImportMode::Uninitialized);
+                       gauges_.push_back(gauge);
+                     });
 
   snapped_histograms_ = store.histograms();
   histograms_.reserve(snapped_histograms_.size());
@@ -184,11 +183,9 @@ MetricSnapshotImpl::MetricSnapshotImpl(Stats::Store& store, TimeSource& time_sou
     histograms_.push_back(*histogram);
   }
 
-  snapped_text_readouts_ = store.textReadouts();
-  text_readouts_.reserve(snapped_text_readouts_.size());
-  for (const auto& text_readout : snapped_text_readouts_) {
-    text_readouts_.push_back(*text_readout);
-  }
+  store.forEachTextReadout(
+      [this](std::size_t size) mutable { text_readouts_.reserve(size); },
+      [this](Stats::TextReadout& text_readout) { text_readouts_.push_back(text_readout); });
 
   snapshot_time_ = time_source.systemTime();
 }
diff --git a/test/common/stats/allocator_impl_test.cc b/test/common/stats/allocator_impl_test.cc
index 83fb85d68ea8d..cc06acedaef1e 100644
--- a/test/common/stats/allocator_impl_test.cc
+++ b/test/common/stats/allocator_impl_test.cc
@@ -1,3 +1,4 @@
+#include <cmath>
 #include <string>
 
 #include "source/common/stats/allocator_impl.h"
@@ -6,6 +7,7 @@
 #include "test/test_common/thread_factory_for_test.h"
 
 #include "absl/synchronization/notification.h"
+#include "gmock/gmock-matchers.h"
 #include "gtest/gtest.h"
 
 namespace Envoy {
@@ -25,12 +27,18 @@ class AllocatorImplTest : public testing::Test {
 
   void clearStorage() {
     pool_.clear();
-    EXPECT_EQ(0, symbol_table_.numSymbols());
+    // If stats have been marked for deletion, they are not cleared until the
+    // destructor of alloc_ is called, and hence the symbol_table_.numSymbols()
+    // will be greater than zero at this point.
+    if (!are_stats_marked_for_deletion_) {
+      EXPECT_EQ(0, symbol_table_.numSymbols());
+    }
   }
 
   SymbolTableImpl symbol_table_;
   AllocatorImpl alloc_;
   StatNamePool pool_;
+  bool are_stats_marked_for_deletion_ = false;
 };
 
 // Allocate 2 counters of the same name, and you'll get the same object.
@@ -125,6 +133,283 @@ TEST_F(AllocatorImplTest, RefCountDecAllocRaceSynchronized) {
   EXPECT_FALSE(alloc_.isMutexLockedForTest());
 }
 
+TEST_F(AllocatorImplTest, ForEachCounter) {
+  StatNameHashSet stat_names;
+  std::vector<CounterSharedPtr> counters;
+
+  const size_t num_stats = 11;
+
+  for (size_t idx = 0; idx < num_stats; ++idx) {
+    auto stat_name = makeStat(absl::StrCat("counter.", idx));
+    stat_names.insert(stat_name);
+    counters.emplace_back(alloc_.makeCounter(stat_name, StatName(), {}));
+  }
+
+  size_t num_counters = 0;
+  size_t num_iterations = 0;
+  alloc_.forEachCounter([&num_counters](std::size_t size) { num_counters = size; },
+                        [&num_iterations, &stat_names](Stats::Counter& counter) {
+                          EXPECT_EQ(stat_names.count(counter.statName()), 1);
+                          ++num_iterations;
+                        });
+  EXPECT_EQ(num_counters, 11);
+  EXPECT_EQ(num_iterations, 11);
+
+  // Reject a stat and remove it from "scope".
+  StatName rejected_stat_name = counters[4]->statName();
+  alloc_.markCounterForDeletion(counters[4]);
+  are_stats_marked_for_deletion_ = true;
+  // Save a local reference to rejected stat.
+  Counter& rejected_counter = *counters[4];
+  counters.erase(counters.begin() + 4);
+
+  // Verify that the rejected stat does not show up during iteration.
+  num_iterations = 0;
+  num_counters = 0;
+  alloc_.forEachCounter([&num_counters](std::size_t size) { num_counters = size; },
+                        [&num_iterations, &rejected_stat_name](Stats::Counter& counter) {
+                          EXPECT_THAT(counter.statName(), ::testing::Ne(rejected_stat_name));
+                          ++num_iterations;
+                        });
+  EXPECT_EQ(num_iterations, 10);
+  EXPECT_EQ(num_counters, 10);
+
+  // Verify that we can access the local reference without a crash.
+  rejected_counter.inc();
+
+  // Erase all stats.
+  counters.clear();
+  num_iterations = 0;
+  alloc_.forEachCounter([&num_counters](std::size_t size) { num_counters = size; },
+                        [&num_iterations](Stats::Counter&) { ++num_iterations; });
+  EXPECT_EQ(num_counters, 0);
+  EXPECT_EQ(num_iterations, 0);
+}
+
+TEST_F(AllocatorImplTest, ForEachGauge) {
+  StatNameHashSet stat_names;
+  std::vector<GaugeSharedPtr> gauges;
+
+  const size_t num_stats = 11;
+
+  for (size_t idx = 0; idx < num_stats; ++idx) {
+    auto stat_name = makeStat(absl::StrCat("gauge.", idx));
+    stat_names.insert(stat_name);
+    gauges.emplace_back(alloc_.makeGauge(stat_name, StatName(), {}, Gauge::ImportMode::Accumulate));
+  }
+
+  size_t num_gauges = 0;
+  size_t num_iterations = 0;
+  alloc_.forEachGauge([&num_gauges](std::size_t size) { num_gauges = size; },
+                      [&num_iterations, &stat_names](Stats::Gauge& gauge) {
+                        EXPECT_EQ(stat_names.count(gauge.statName()), 1);
+                        ++num_iterations;
+                      });
+  EXPECT_EQ(num_gauges, 11);
+  EXPECT_EQ(num_iterations, 11);
+
+  // Reject a stat and remove it from "scope".
+  StatName rejected_stat_name = gauges[3]->statName();
+  alloc_.markGaugeForDeletion(gauges[3]);
+  are_stats_marked_for_deletion_ = true;
+  // Save a local reference to rejected stat.
+  Gauge& rejected_gauge = *gauges[3];
+  gauges.erase(gauges.begin() + 3);
+
+  // Verify that the rejected stat does not show up during iteration.
+  num_iterations = 0;
+  num_gauges = 0;
+  alloc_.forEachGauge([&num_gauges](std::size_t size) { num_gauges = size; },
+                      [&num_iterations, &rejected_stat_name](Stats::Gauge& gauge) {
+                        EXPECT_THAT(gauge.statName(), ::testing::Ne(rejected_stat_name));
+                        ++num_iterations;
+                      });
+  EXPECT_EQ(num_iterations, 10);
+  EXPECT_EQ(num_gauges, 10);
+
+  // Verify that we can access the local reference without a crash.
+  rejected_gauge.inc();
+
+  // Erase all stats.
+  gauges.clear();
+  num_iterations = 0;
+  alloc_.forEachGauge([&num_gauges](std::size_t size) { num_gauges = size; },
+                      [&num_iterations](Stats::Gauge&) { ++num_iterations; });
+  EXPECT_EQ(num_gauges, 0);
+  EXPECT_EQ(num_iterations, 0);
+}
+
+TEST_F(AllocatorImplTest, ForEachTextReadout) {
+  StatNameHashSet stat_names;
+  std::vector<TextReadoutSharedPtr> text_readouts;
+
+  const size_t num_stats = 11;
+
+  for (size_t idx = 0; idx < num_stats; ++idx) {
+    auto stat_name = makeStat(absl::StrCat("text_readout.", idx));
+    stat_names.insert(stat_name);
+    text_readouts.emplace_back(alloc_.makeTextReadout(stat_name, StatName(), {}));
+  }
+
+  size_t num_text_readouts = 0;
+  size_t num_iterations = 0;
+  alloc_.forEachTextReadout([&num_text_readouts](std::size_t size) { num_text_readouts = size; },
+                            [&num_iterations, &stat_names](Stats::TextReadout& text_readout) {
+                              EXPECT_EQ(stat_names.count(text_readout.statName()), 1);
+                              ++num_iterations;
+                            });
+  EXPECT_EQ(num_text_readouts, 11);
+  EXPECT_EQ(num_iterations, 11);
+
+  // Reject a stat and remove it from "scope".
+  StatName rejected_stat_name = text_readouts[4]->statName();
+  alloc_.markTextReadoutForDeletion(text_readouts[4]);
+  are_stats_marked_for_deletion_ = true;
+  // Save a local reference to rejected stat.
+  TextReadout& rejected_text_readout = *text_readouts[4];
+  text_readouts.erase(text_readouts.begin() + 4);
+
+  // Verify that the rejected stat does not show up during iteration.
+  num_iterations = 0;
+  num_text_readouts = 0;
+  alloc_.forEachTextReadout(
+      [&num_text_readouts](std::size_t size) { num_text_readouts = size; },
+      [&num_iterations, &rejected_stat_name](Stats::TextReadout& text_readout) {
+        EXPECT_THAT(text_readout.statName(), ::testing::Ne(rejected_stat_name));
+        ++num_iterations;
+      });
+  EXPECT_EQ(num_iterations, 10);
+  EXPECT_EQ(num_text_readouts, 10);
+
+  // Verify that we can access the local reference without a crash.
+  rejected_text_readout.set("no crash");
+
+  // Erase all stats.
+  text_readouts.clear();
+  num_iterations = 0;
+  alloc_.forEachTextReadout([&num_text_readouts](std::size_t size) { num_text_readouts = size; },
+                            [&num_iterations](Stats::TextReadout&) { ++num_iterations; });
+  EXPECT_EQ(num_text_readouts, 0);
+  EXPECT_EQ(num_iterations, 0);
+}
+
+// Verify that we don't crash if a nullptr is passed in for the size lambda for
+// the for each stat methods.
+TEST_F(AllocatorImplTest, ForEachWithNullSizeLambda) {
+  std::vector<CounterSharedPtr> counters;
+  std::vector<TextReadoutSharedPtr> text_readouts;
+  std::vector<GaugeSharedPtr> gauges;
+
+  const size_t num_stats = 3;
+
+  // For each counter.
+  for (size_t idx = 0; idx < num_stats; ++idx) {
+    auto stat_name = makeStat(absl::StrCat("counter.", idx));
+    counters.emplace_back(alloc_.makeCounter(stat_name, StatName(), {}));
+  }
+  size_t num_iterations = 0;
+  alloc_.forEachCounter(nullptr, [&num_iterations](Stats::Counter& counter) {
+    (void)counter;
+    ++num_iterations;
+  });
+  EXPECT_EQ(num_iterations, num_stats);
+
+  // For each gauge.
+  for (size_t idx = 0; idx < num_stats; ++idx) {
+    auto stat_name = makeStat(absl::StrCat("gauge.", idx));
+    gauges.emplace_back(alloc_.makeGauge(stat_name, StatName(), {}, Gauge::ImportMode::Accumulate));
+  }
+  num_iterations = 0;
+  alloc_.forEachGauge(nullptr, [&num_iterations](Stats::Gauge& gauge) {
+    (void)gauge;
+    ++num_iterations;
+  });
+  EXPECT_EQ(num_iterations, num_stats);
+
+  // For each text readout.
+  for (size_t idx = 0; idx < num_stats; ++idx) {
+    auto stat_name = makeStat(absl::StrCat("text_readout.", idx));
+    text_readouts.emplace_back(alloc_.makeTextReadout(stat_name, StatName(), {}));
+  }
+  num_iterations = 0;
+  alloc_.forEachTextReadout(nullptr, [&num_iterations](Stats::TextReadout& text_readout) {
+    (void)text_readout;
+    ++num_iterations;
+  });
+  EXPECT_EQ(num_iterations, num_stats);
+}
+
+// Currently, if we ask for a stat from the Allocator that has already been
+// marked for deletion (i.e. rejected) we get a new stat with the same name.
+// This test documents this behavior.
+TEST_F(AllocatorImplTest, AskForDeletedStat) {
+  const size_t num_stats = 10;
+  are_stats_marked_for_deletion_ = true;
+
+  std::vector<CounterSharedPtr> counters;
+  for (size_t idx = 0; idx < num_stats; ++idx) {
+    auto stat_name = makeStat(absl::StrCat("counter.", idx));
+    counters.emplace_back(alloc_.makeCounter(stat_name, StatName(), {}));
+  }
+  // Reject a stat and remove it from "scope".
+  StatName const rejected_counter_name = counters[4]->statName();
+  alloc_.markCounterForDeletion(counters[4]);
+  // Save a local reference to rejected stat.
+  Counter& rejected_counter = *counters[4];
+  counters.erase(counters.begin() + 4);
+
+  rejected_counter.inc();
+  rejected_counter.inc();
+
+  // Make the deleted stat again.
+  CounterSharedPtr deleted_counter = alloc_.makeCounter(rejected_counter_name, StatName(), {});
+
+  EXPECT_EQ(deleted_counter->value(), 0);
+  EXPECT_EQ(rejected_counter.value(), 2);
+
+  std::vector<GaugeSharedPtr> gauges;
+  for (size_t idx = 0; idx < num_stats; ++idx) {
+    auto stat_name = makeStat(absl::StrCat("gauge.", idx));
+    gauges.emplace_back(alloc_.makeGauge(stat_name, StatName(), {}, Gauge::ImportMode::Accumulate));
+  }
+  // Reject a stat and remove it from "scope".
+  StatName const rejected_gauge_name = gauges[4]->statName();
+  alloc_.markGaugeForDeletion(gauges[4]);
+  // Save a local reference to rejected stat.
+  Gauge& rejected_gauge = *gauges[4];
+  gauges.erase(gauges.begin() + 4);
+
+  rejected_gauge.set(10);
+
+  // Make the deleted stat again.
+  GaugeSharedPtr deleted_gauge =
+      alloc_.makeGauge(rejected_gauge_name, StatName(), {}, Gauge::ImportMode::Accumulate);
+
+  EXPECT_EQ(deleted_gauge->value(), 0);
+  EXPECT_EQ(rejected_gauge.value(), 10);
+
+  std::vector<TextReadoutSharedPtr> text_readouts;
+  for (size_t idx = 0; idx < num_stats; ++idx) {
+    auto stat_name = makeStat(absl::StrCat("text_readout.", idx));
+    text_readouts.emplace_back(alloc_.makeTextReadout(stat_name, StatName(), {}));
+  }
+  // Reject a stat and remove it from "scope".
+  StatName const rejected_text_readout_name = text_readouts[4]->statName();
+  alloc_.markTextReadoutForDeletion(text_readouts[4]);
+  // Save a local reference to rejected stat.
+  TextReadout& rejected_text_readout = *text_readouts[4];
+  text_readouts.erase(text_readouts.begin() + 4);
+
+  rejected_text_readout.set("deleted value");
+
+  // Make the deleted stat again.
+  TextReadoutSharedPtr deleted_text_readout =
+      alloc_.makeTextReadout(rejected_text_readout_name, StatName(), {});
+
+  EXPECT_EQ(deleted_text_readout->value(), "");
+  EXPECT_EQ(rejected_text_readout.value(), "deleted value");
+}
+
 } // namespace
 } // namespace Stats
 } // namespace Envoy
diff --git a/test/common/stats/thread_local_store_test.cc b/test/common/stats/thread_local_store_test.cc
index dd77935494dfb..5bf5e67395437 100644
--- a/test/common/stats/thread_local_store_test.cc
+++ b/test/common/stats/thread_local_store_test.cc
@@ -508,10 +508,15 @@ TEST_F(StatsThreadLocalStoreTest, ScopeDelete) {
   EXPECT_CALL(main_thread_dispatcher_, post(_));
   EXPECT_CALL(tls_, runOnAllThreads(_, _)).Times(testing::AtLeast(1));
   scope1.reset();
-  EXPECT_EQ(0UL, store_->counters().size());
+  // The counter is gone from all scopes, but is still held in the local
+  // variable c1. Hence, it will not be removed from the allocator or store.
+  EXPECT_EQ(1UL, store_->counters().size());
 
   EXPECT_EQ(1L, c1.use_count());
   c1.reset();
+  // Removing the counter from the local variable, should now remove it from the
+  // allocator.
+  EXPECT_EQ(0UL, store_->counters().size());
 
   tls_.shutdownGlobalThreading();
   store_->shutdownThreading();
@@ -1192,6 +1197,48 @@ TEST_F(StatsThreadLocalStoreTest, RemoveRejectedStats) {
   tls_.shutdownThread();
 }
 
+// Verify that asking for deleted stats by name does not create new copies on
+// the allocator.
+TEST_F(StatsThreadLocalStoreTest, AskForRejectedStat) {
+  store_->initializeThreading(main_thread_dispatcher_, tls_);
+  Counter& counter = store_->counterFromString("c1");
+  Gauge& gauge = store_->gaugeFromString("g1", Gauge::ImportMode::Accumulate);
+  TextReadout& text_readout = store_->textReadoutFromString("t1");
+  ASSERT_EQ(1, store_->counters().size()); // "c1".
+  ASSERT_EQ(1, store_->gauges().size());
+  ASSERT_EQ(1, store_->textReadouts().size());
+
+  // Will effectively block all stats, and remove all the non-matching stats.
+  envoy::config::metrics::v3::StatsConfig stats_config;
+  stats_config.mutable_stats_matcher()->mutable_inclusion_list()->add_patterns()->set_exact(
+      "no-such-stat");
+  store_->setStatsMatcher(std::make_unique<StatsMatcherImpl>(stats_config, symbol_table_));
+
+  // They can no longer be found.
+  EXPECT_EQ(0, store_->counters().size());
+  EXPECT_EQ(0, store_->gauges().size());
+  EXPECT_EQ(0, store_->textReadouts().size());
+
+  // Ask for the rejected stats again by name.
+  Counter& counter2 = store_->counterFromString("c1");
+  Gauge& gauge2 = store_->gaugeFromString("g1", Gauge::ImportMode::Accumulate);
+  TextReadout& text_readout2 = store_->textReadoutFromString("t1");
+
+  // Verify we got the same stats.
+  EXPECT_EQ(&counter, &counter2);
+  EXPECT_EQ(&gauge, &gauge2);
+  EXPECT_EQ(&text_readout, &text_readout2);
+
+  // Verify that new stats were not created.
+  EXPECT_EQ(0, store_->counters().size());
+  EXPECT_EQ(0, store_->gauges().size());
+  EXPECT_EQ(0, store_->textReadouts().size());
+
+  tls_.shutdownGlobalThreading();
+  store_->shutdownThreading();
+  tls_.shutdownThread();
+}
+
 TEST_F(StatsThreadLocalStoreTest, NonHotRestartNoTruncation) {
   InSequence s;
   store_->initializeThreading(main_thread_dispatcher_, tls_);
diff --git a/test/integration/server.h b/test/integration/server.h
index 109743e551b23..da51efe2eabf7 100644
--- a/test/integration/server.h
+++ b/test/integration/server.h
@@ -281,6 +281,21 @@ class TestIsolatedStoreImpl : public StoreRoot {
     Thread::LockGuard lock(lock_);
     return store_.counterFromStatNameWithTags(name, tags);
   }
+  void forEachCounter(std::function<void(std::size_t)> f_size,
+                      std::function<void(Stats::Counter&)> f_stat) const override {
+    Thread::LockGuard lock(lock_);
+    store_.forEachCounter(f_size, f_stat);
+  }
+  void forEachGauge(std::function<void(std::size_t)> f_size,
+                    std::function<void(Stats::Gauge&)> f_stat) const override {
+    Thread::LockGuard lock(lock_);
+    store_.forEachGauge(f_size, f_stat);
+  }
+  void forEachTextReadout(std::function<void(std::size_t)> f_size,
+                          std::function<void(Stats::TextReadout&)> f_stat) const override {
+    Thread::LockGuard lock(lock_);
+    store_.forEachTextReadout(f_size, f_stat);
+  }
   Counter& counterFromString(const std::string& name) override {
     Thread::LockGuard lock(lock_);
     return store_.counterFromString(name);
diff --git a/test/mocks/stats/mocks.h b/test/mocks/stats/mocks.h
index 3adec4d4a8ae2..ba81b5922f306 100644
--- a/test/mocks/stats/mocks.h
+++ b/test/mocks/stats/mocks.h
@@ -2,6 +2,7 @@
 
 #include <chrono>
 #include <cstdint>
+#include <functional>
 #include <list>
 #include <string>
 
@@ -285,6 +286,13 @@ class MockStore : public TestUtil::TestStore {
   MOCK_METHOD(Histogram&, histogramFromString, (const std::string& name, Histogram::Unit unit));
   MOCK_METHOD(TextReadout&, textReadout, (const std::string&));
   MOCK_METHOD(std::vector<TextReadoutSharedPtr>, text_readouts, (), (const));
+  MOCK_METHOD(void, forEachCounter,
+              (std::function<void(std::size_t)>, std::function<void(Stats::Counter&)>), (const));
+  MOCK_METHOD(void, forEachGauge,
+              (std::function<void(std::size_t)>, std::function<void(Stats::Gauge&)>), (const));
+  MOCK_METHOD(void, forEachTextReadout,
+              (std::function<void(std::size_t)>, std::function<void(Stats::TextReadout&)>),
+              (const));
 
   MOCK_METHOD(CounterOptConstRef, findCounter, (StatName), (const));
   MOCK_METHOD(GaugeOptConstRef, findGauge, (StatName), (const));
diff --git a/test/server/BUILD b/test/server/BUILD
index 5e50e3cbacb4d..952131ac020fe 100644
--- a/test/server/BUILD
+++ b/test/server/BUILD
@@ -499,3 +499,22 @@ envoy_benchmark_test(
     timeout = "long",
     benchmark_binary = "filter_chain_benchmark_test",
 )
+
+envoy_cc_benchmark_binary(
+    name = "server_stats_flush_benchmark",
+    srcs = ["server_stats_flush_benchmark_test.cc"],
+    external_deps = [
+        "benchmark",
+    ],
+    deps = [
+        "//envoy/stats:stats_interface",
+        "//source/common/stats:thread_local_store_lib",
+        "//source/server:server_lib",
+        "//test/test_common:simulated_time_system_lib",
+    ],
+)
+
+envoy_benchmark_test(
+    name = "server_stats_flush_benchmark_test",
+    benchmark_binary = "server_stats_flush_benchmark",
+)
diff --git a/test/server/server_stats_flush_benchmark_test.cc b/test/server/server_stats_flush_benchmark_test.cc
new file mode 100644
index 0000000000000..98790c5c65d56
--- /dev/null
+++ b/test/server/server_stats_flush_benchmark_test.cc
@@ -0,0 +1,74 @@
+#include <cstdint>
+#include <memory>
+
+#include "envoy/stats/sink.h"
+#include "envoy/stats/stats.h"
+
+#include "source/common/stats/thread_local_store.h"
+#include "source/server/server.h"
+
+#include "test/benchmark/main.h"
+#include "test/mocks/stats/mocks.h"
+#include "test/test_common/simulated_time_system.h"
+#include "test/test_common/utility.h"
+
+#include "absl/strings/str_cat.h"
+#include "benchmark/benchmark.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace Envoy {
+
+class StatsSinkFlushSpeedTest {
+public:
+  StatsSinkFlushSpeedTest(size_t const num_stats)
+      : pool_(symbol_table_), stats_allocator_(symbol_table_), stats_store_(stats_allocator_) {
+
+    // Create counters
+    for (uint64_t idx = 0; idx < num_stats; ++idx) {
+      auto stat_name = pool_.add(absl::StrCat("counter.", idx));
+      stats_store_.counterFromStatName(stat_name).inc();
+    }
+    // Create gauges
+    for (uint64_t idx = 0; idx < num_stats; ++idx) {
+      auto stat_name = pool_.add(absl::StrCat("gauge.", idx));
+      stats_store_.gaugeFromStatName(stat_name, Stats::Gauge::ImportMode::NeverImport).set(idx);
+    }
+
+    // Create text readouts
+    for (uint64_t idx = 0; idx < num_stats; ++idx) {
+      auto stat_name = pool_.add(absl::StrCat("text_readout.", idx));
+      stats_store_.textReadoutFromStatName(stat_name).set(absl::StrCat("text_readout.", idx));
+    }
+  }
+
+  void test(::benchmark::State& state) {
+    for (auto _ : state) {
+      UNREFERENCED_PARAMETER(_);
+      std::list<Stats::SinkPtr> sinks;
+      sinks.emplace_back(new testing::NiceMock<Stats::MockSink>());
+      Server::InstanceUtil::flushMetricsToSinks(sinks, stats_store_, time_system_);
+    }
+  }
+
+private:
+  Stats::SymbolTableImpl symbol_table_;
+  Stats::StatNamePool pool_;
+  Stats::AllocatorImpl stats_allocator_;
+  Stats::ThreadLocalStoreImpl stats_store_;
+  Event::SimulatedTimeSystem time_system_;
+};
+
+static void bmFlushToSinks(::benchmark::State& state) {
+  // Skip expensive benchmarks for unit tests.
+  if (benchmark::skipExpensiveBenchmarks() && state.range(0) > 100) {
+    state.SkipWithError("Skipping expensive benchmark");
+    return;
+  }
+
+  StatsSinkFlushSpeedTest speed_test(state.range(0));
+  speed_test.test(state);
+}
+BENCHMARK(bmFlushToSinks)->Unit(::benchmark::kMillisecond)->RangeMultiplier(10)->Range(10, 1000000);
+
+} // namespace Envoy
diff --git a/tools/spelling/spelling_dictionary.txt b/tools/spelling/spelling_dictionary.txt
index 324cc3eb544c0..96dda9ee1dc88 100644
--- a/tools/spelling/spelling_dictionary.txt
+++ b/tools/spelling/spelling_dictionary.txt
@@ -15,6 +15,7 @@ ARN
 ASAN
 ASCII
 ASM
+ASSERTed
 ASSERTs
 AST
 AWS