diff --git a/changelogs/current.yaml b/changelogs/current.yaml
index 7f583e31d0080..c8bd1ef1530e8 100644
--- a/changelogs/current.yaml
+++ b/changelogs/current.yaml
@@ -271,6 +271,9 @@ new_features:
 - area: tracing
   change: |
     added support for setting the hostname used when sending spans to a Datadog collector using the :ref:`collector_hostname <envoy_v3_api_field_config.trace.v3.DatadogConfig.collector_hostname>` field.
+- area: stats
+  change: |
+    added ``includeHistogram()`` method to ``Stats::SinkPredicates`` to filter histograms to be flushed to stat sinks. Use ``envoy.reloadable_features.enable_include_histograms`` to enable this feature, which is disabled by default.
 - area: http
   change: |
     added support of :ref:`early header mutation <envoy_v3_api_field_extensions.filters.network.http_connection_manager.v3.HttpConnectionManager.early_header_mutation_extensions>` to the HTTP connection manager.
diff --git a/envoy/server/instance.h b/envoy/server/instance.h
index f3fdd98be6baf..a3686473f3efd 100644
--- a/envoy/server/instance.h
+++ b/envoy/server/instance.h
@@ -282,7 +282,7 @@ class Instance {
   virtual bool enableReusePortDefault() PURE;
 
   /**
-   * Set predicates for filtering counters, gauges and text readouts to be flushed to sinks.
+   * Set predicates for filtering stats to be flushed to sinks.
    */
   virtual void
   setSinkPredicates(std::unique_ptr<Envoy::Stats::SinkPredicates>&& sink_predicates) PURE;
diff --git a/envoy/stats/allocator.h b/envoy/stats/allocator.h
index 0d61c0d97eeb2..fc45ca8af4f14 100644
--- a/envoy/stats/allocator.h
+++ b/envoy/stats/allocator.h
@@ -95,7 +95,7 @@ class Allocator {
   virtual void forEachSinkedTextReadout(SizeFn f_size, StatFn<TextReadout> f_stat) const PURE;
 
   /**
-   * Set the predicates to filter counters, gauges and text readouts for sink.
+   * Set the predicates to filter stats for sink.
    */
   virtual void setSinkPredicates(std::unique_ptr<SinkPredicates>&& sink_predicates) PURE;
 
diff --git a/envoy/stats/sink.h b/envoy/stats/sink.h
index a0d914416bd9d..e833c44c554c6 100644
--- a/envoy/stats/sink.h
+++ b/envoy/stats/sink.h
@@ -69,6 +69,13 @@ class SinkPredicates {
    * @return true if @param text_readout needs to be flushed to sinks.
    */
   virtual bool includeTextReadout(const TextReadout& text_readout) PURE;
+
+  /*
+   * @return true if @param histogram needs to be flushed to sinks.
+   * Note that this is used only if runtime flag envoy.reloadable_features.enable_include_histograms
+   * (which is false by default) is set to true.
+   */
+  virtual bool includeHistogram(const Histogram& histogram) PURE;
 };
 
 /**
diff --git a/envoy/stats/store.h b/envoy/stats/store.h
index db34ccb93cf7b..249d7e9a4d795 100644
--- a/envoy/stats/store.h
+++ b/envoy/stats/store.h
@@ -4,6 +4,7 @@
 #include <memory>
 #include <vector>
 
+#include "envoy/common/optref.h"
 #include "envoy/common/pure.h"
 #include "envoy/stats/histogram.h"
 #include "envoy/stats/scope.h"
@@ -136,6 +137,7 @@ class Store {
   virtual void forEachSinkedCounter(SizeFn f_size, StatFn<Counter> f_stat) const PURE;
   virtual void forEachSinkedGauge(SizeFn f_size, StatFn<Gauge> f_stat) const PURE;
   virtual void forEachSinkedTextReadout(SizeFn f_size, StatFn<TextReadout> f_stat) const PURE;
+  virtual void forEachSinkedHistogram(SizeFn f_size, StatFn<ParentHistogram> f_stat) const PURE;
 
   /**
    * Calls 'fn' for every stat. Note that in the case of overlapping scopes, the
@@ -240,12 +242,14 @@ class StoreRoot : public Store {
   virtual void mergeHistograms(PostMergeCb merge_complete_cb) PURE;
 
   /**
-   * Set predicates for filtering counters, gauges and text readouts to be flushed to sinks.
+   * Set predicates for filtering stats to be flushed to sinks.
    * Note that if the sink predicates object is set, we do not send non-sink stats over to the
    * child process during hot restart. This will result in the admin stats console being wrong
    * during hot restart.
    */
   virtual void setSinkPredicates(std::unique_ptr<SinkPredicates>&& sink_predicates) PURE;
+
+  virtual OptRef<SinkPredicates> sinkPredicates() PURE;
 };
 
 using StoreRootPtr = std::unique_ptr<StoreRoot>;
diff --git a/source/common/runtime/runtime_features.cc b/source/common/runtime/runtime_features.cc
index 5afce54218c94..fa8b1d8ccda66 100644
--- a/source/common/runtime/runtime_features.cc
+++ b/source/common/runtime/runtime_features.cc
@@ -98,6 +98,8 @@ FALSE_RUNTIME_GUARD(envoy_reloadable_features_runtime_initialized);
 FALSE_RUNTIME_GUARD(envoy_reloadable_features_always_use_v6);
 // TODO(alyssawilk) remove in Q2.
 FALSE_RUNTIME_GUARD(envoy_reloadable_features_no_delay_close_for_upgrades);
+// TODO(pradeepcrao) reset this to true after 2 releases (1.27)
+FALSE_RUNTIME_GUARD(envoy_reloadable_features_enable_include_histograms);
 
 // Block of non-boolean flags. These are deprecated. Do not add more.
 ABSL_FLAG(uint64_t, envoy_headermap_lazy_map_min_size, 3, "");  // NOLINT
diff --git a/source/common/stats/isolated_store_impl.h b/source/common/stats/isolated_store_impl.h
index cd155e6b34821..ee25f2f924d07 100644
--- a/source/common/stats/isolated_store_impl.h
+++ b/source/common/stats/isolated_store_impl.h
@@ -196,6 +196,11 @@ class IsolatedStoreImpl : public Store {
     forEachTextReadout(f_size, f_stat);
   }
 
+  void forEachSinkedHistogram(SizeFn f_size, StatFn<ParentHistogram> f_stat) const override {
+    UNREFERENCED_PARAMETER(f_size);
+    UNREFERENCED_PARAMETER(f_stat);
+  }
+
   NullCounterImpl& nullCounter() override { return *null_counter_; }
   NullGaugeImpl& nullGauge() override { return *null_gauge_; }
 
diff --git a/source/common/stats/thread_local_store.cc b/source/common/stats/thread_local_store.cc
index d157f06a9f557..edbb1f5957624 100644
--- a/source/common/stats/thread_local_store.cc
+++ b/source/common/stats/thread_local_store.cc
@@ -12,6 +12,7 @@
 #include "envoy/stats/stats.h"
 
 #include "source/common/common/lock_guard.h"
+#include "source/common/runtime/runtime_features.h"
 #include "source/common/stats/histogram_impl.h"
 #include "source/common/stats/stats_matcher_impl.h"
 #include "source/common/stats/tag_producer_impl.h"
@@ -95,6 +96,7 @@ void ThreadLocalStoreImpl::setStatsMatcher(StatsMatcherPtr&& stats_matcher) {
     for (uint32_t i = first_histogram_index; i < deleted_histograms_.size(); ++i) {
       uint32_t erased = histogram_set_.erase(deleted_histograms_[i].get());
       ASSERT(erased == 1);
+      sinked_histograms_.erase(deleted_histograms_[i].get());
     }
   }
 }
@@ -225,6 +227,7 @@ void ThreadLocalStoreImpl::shutdownThreading() {
     histogram->setShuttingDown(true);
   }
   histogram_set_.clear();
+  sinked_histograms_.clear();
 }
 
 void ThreadLocalStoreImpl::mergeHistograms(PostMergeCb merge_complete_cb) {
@@ -678,6 +681,10 @@ Histogram& ThreadLocalStoreImpl::ScopeImpl::histogramFromStatNameWithTags(
                                        *buckets, parent_.next_histogram_id_++);
         if (!parent_.shutting_down_) {
           parent_.histogram_set_.insert(stat.get());
+          if (parent_.sink_predicates_.has_value() &&
+              parent_.sink_predicates_->includeHistogram(*stat)) {
+            parent_.sinked_histograms_.insert(stat.get());
+          }
         }
       }
     }
@@ -877,6 +884,7 @@ bool ThreadLocalStoreImpl::decHistogramRefCount(ParentHistogramImpl& hist,
     if (!shutting_down_) {
       const size_t count = histogram_set_.erase(hist.statName());
       ASSERT(shutting_down_ || count == 1);
+      sinked_histograms_.erase(&hist);
     }
     return true;
   }
@@ -1034,11 +1042,36 @@ void ThreadLocalStoreImpl::forEachSinkedTextReadout(SizeFn f_size,
   alloc_.forEachSinkedTextReadout(f_size, f_stat);
 }
 
+void ThreadLocalStoreImpl::forEachSinkedHistogram(SizeFn f_size,
+                                                  StatFn<ParentHistogram> f_stat) const {
+  if (sink_predicates_.has_value() &&
+      Runtime::runtimeFeatureEnabled("envoy.reloadable_features.enable_include_histograms")) {
+    Thread::LockGuard lock(hist_mutex_);
+
+    if (f_size != nullptr) {
+      f_size(sinked_histograms_.size());
+    }
+    for (auto histogram : sinked_histograms_) {
+      f_stat(*histogram);
+    }
+  } else {
+    forEachHistogram(f_size, f_stat);
+  }
+}
+
 void ThreadLocalStoreImpl::setSinkPredicates(std::unique_ptr<SinkPredicates>&& sink_predicates) {
   ASSERT(sink_predicates != nullptr);
   if (sink_predicates != nullptr) {
     sink_predicates_.emplace(*sink_predicates);
     alloc_.setSinkPredicates(std::move(sink_predicates));
+    // Add histograms to the set of sinked histograms.
+    Thread::LockGuard lock(hist_mutex_);
+    sinked_histograms_.clear();
+    for (auto& histogram : histogram_set_) {
+      if (sink_predicates_->includeHistogram(*histogram)) {
+        sinked_histograms_.insert(histogram);
+      }
+    }
   }
 }
 
diff --git a/source/common/stats/thread_local_store.h b/source/common/stats/thread_local_store.h
index 124bdf77e13ab..04d0c216b5d1c 100644
--- a/source/common/stats/thread_local_store.h
+++ b/source/common/stats/thread_local_store.h
@@ -192,8 +192,10 @@ class ThreadLocalStoreImpl : Logger::Loggable<Logger::Id::stats>, public StoreRo
   void forEachSinkedCounter(SizeFn f_size, StatFn<Counter> f_stat) const override;
   void forEachSinkedGauge(SizeFn f_size, StatFn<Gauge> f_stat) const override;
   void forEachSinkedTextReadout(SizeFn f_size, StatFn<TextReadout> f_stat) const override;
+  void forEachSinkedHistogram(SizeFn f_size, StatFn<ParentHistogram> f_stat) const override;
 
   void setSinkPredicates(std::unique_ptr<SinkPredicates>&& sink_predicates) override;
+  OptRef<SinkPredicates> sinkPredicates() override { return sink_predicates_; }
 
   /**
    * @return a thread synchronizer object used for controlling thread behavior in tests.
@@ -519,6 +521,7 @@ class ThreadLocalStoreImpl : Logger::Loggable<Logger::Id::stats>, public StoreRo
 
   mutable Thread::MutexBasicLockable hist_mutex_;
   StatSet<ParentHistogramImpl> histogram_set_ ABSL_GUARDED_BY(hist_mutex_);
+  StatSet<ParentHistogramImpl> sinked_histograms_ ABSL_GUARDED_BY(hist_mutex_);
 
   // Retain storage for deleted stats; these are no longer in maps because the
   // matcher-pattern was established after they were created. Since the stats
diff --git a/source/server/server.cc b/source/server/server.cc
index 8d864aedc5c8b..70e19cee251df 100644
--- a/source/server/server.cc
+++ b/source/server/server.cc
@@ -219,7 +219,7 @@ MetricSnapshotImpl::MetricSnapshotImpl(Stats::Store& store, TimeSource& time_sou
         gauges_.push_back(gauge);
       });
 
-  store.forEachHistogram(
+  store.forEachSinkedHistogram(
       [this](std::size_t size) {
         snapped_histograms_.reserve(size);
         histograms_.reserve(size);
diff --git a/test/common/stats/BUILD b/test/common/stats/BUILD
index af53d9d98bce2..8bd9d56c4acb3 100644
--- a/test/common/stats/BUILD
+++ b/test/common/stats/BUILD
@@ -265,6 +265,7 @@ envoy_cc_test(
     deps = [
         ":stat_test_utility_lib",
         "//source/common/memory:stats_lib",
+        "//source/common/runtime:runtime_lib",
         "//source/common/stats:stats_matcher_lib",
         "//source/common/stats:thread_local_store_lib",
         "//source/common/thread_local:thread_local_lib",
diff --git a/test/common/stats/allocator_impl_test.cc b/test/common/stats/allocator_impl_test.cc
index a4c0eb2d271b4..bbb0da55f25eb 100644
--- a/test/common/stats/allocator_impl_test.cc
+++ b/test/common/stats/allocator_impl_test.cc
@@ -6,6 +6,7 @@
 
 #include "source/common/stats/allocator_impl.h"
 
+#include "test/common/stats/stat_test_utility.h"
 #include "test/test_common/logging.h"
 #include "test/test_common/thread_factory_for_test.h"
 
@@ -19,7 +20,7 @@ namespace {
 
 class AllocatorImplTest : public testing::Test {
 protected:
-  AllocatorImplTest() : alloc_(symbol_table_), pool_(symbol_table_) {}
+  AllocatorImplTest() : pool_(symbol_table_), alloc_(symbol_table_) {}
   ~AllocatorImplTest() override { clearStorage(); }
 
   StatNameStorage makeStatStorage(absl::string_view name) {
@@ -39,31 +40,13 @@ class AllocatorImplTest : public testing::Test {
   }
 
   SymbolTableImpl symbol_table_;
-  AllocatorImpl alloc_;
+  // Declare the pool before the allocator because the allocator could contain
+  // a TestSinkPredicates object whose lifetime should be bounded by that of the pool.
   StatNamePool pool_;
+  AllocatorImpl alloc_;
   bool are_stats_marked_for_deletion_ = false;
 };
 
-class TestSinkPredicates : public SinkPredicates {
-public:
-  ~TestSinkPredicates() override = default;
-  StatNameHashSet& sinkedStatNames() { return sinked_stat_names_; }
-
-  // SinkPredicates
-  bool includeCounter(const Counter& counter) override {
-    return sinked_stat_names_.find(counter.statName()) != sinked_stat_names_.end();
-  }
-  bool includeGauge(const Gauge& gauge) override {
-    return sinked_stat_names_.find(gauge.statName()) != sinked_stat_names_.end();
-  }
-  bool includeTextReadout(const TextReadout& text_readout) override {
-    return sinked_stat_names_.find(text_readout.statName()) != sinked_stat_names_.end();
-  }
-
-private:
-  StatNameHashSet sinked_stat_names_;
-};
-
 // Allocate 2 counters of the same name, and you'll get the same object.
 TEST_F(AllocatorImplTest, CountersWithSameName) {
   StatName counter_name = makeStat("counter.name");
@@ -171,7 +154,7 @@ TEST_F(AllocatorImplTest, ForEachCounter) {
   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) {
+                        [&num_iterations, &stat_names](Counter& counter) {
                           EXPECT_EQ(stat_names.count(counter.statName()), 1);
                           ++num_iterations;
                         });
@@ -190,7 +173,7 @@ TEST_F(AllocatorImplTest, ForEachCounter) {
   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) {
+                        [&num_iterations, &rejected_stat_name](Counter& counter) {
                           EXPECT_THAT(counter.statName(), ::testing::Ne(rejected_stat_name));
                           ++num_iterations;
                         });
@@ -204,7 +187,7 @@ TEST_F(AllocatorImplTest, ForEachCounter) {
   counters.clear();
   num_iterations = 0;
   alloc_.forEachCounter([&num_counters](std::size_t size) { num_counters = size; },
-                        [&num_iterations](Stats::Counter&) { ++num_iterations; });
+                        [&num_iterations](Counter&) { ++num_iterations; });
   EXPECT_EQ(num_counters, 0);
   EXPECT_EQ(num_iterations, 0);
 }
@@ -224,7 +207,7 @@ TEST_F(AllocatorImplTest, ForEachGauge) {
   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) {
+                      [&num_iterations, &stat_names](Gauge& gauge) {
                         EXPECT_EQ(stat_names.count(gauge.statName()), 1);
                         ++num_iterations;
                       });
@@ -243,7 +226,7 @@ TEST_F(AllocatorImplTest, ForEachGauge) {
   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) {
+                      [&num_iterations, &rejected_stat_name](Gauge& gauge) {
                         EXPECT_THAT(gauge.statName(), ::testing::Ne(rejected_stat_name));
                         ++num_iterations;
                       });
@@ -257,7 +240,7 @@ TEST_F(AllocatorImplTest, ForEachGauge) {
   gauges.clear();
   num_iterations = 0;
   alloc_.forEachGauge([&num_gauges](std::size_t size) { num_gauges = size; },
-                      [&num_iterations](Stats::Gauge&) { ++num_iterations; });
+                      [&num_iterations](Gauge&) { ++num_iterations; });
   EXPECT_EQ(num_gauges, 0);
   EXPECT_EQ(num_iterations, 0);
 }
@@ -277,7 +260,7 @@ TEST_F(AllocatorImplTest, ForEachTextReadout) {
   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) {
+                            [&num_iterations, &stat_names](TextReadout& text_readout) {
                               EXPECT_EQ(stat_names.count(text_readout.statName()), 1);
                               ++num_iterations;
                             });
@@ -295,12 +278,12 @@ TEST_F(AllocatorImplTest, ForEachTextReadout) {
   // 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;
-      });
+  alloc_.forEachTextReadout([&num_text_readouts](std::size_t size) { num_text_readouts = size; },
+                            [&num_iterations, &rejected_stat_name](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);
 
@@ -311,7 +294,7 @@ TEST_F(AllocatorImplTest, ForEachTextReadout) {
   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; });
+                            [&num_iterations](TextReadout&) { ++num_iterations; });
   EXPECT_EQ(num_text_readouts, 0);
   EXPECT_EQ(num_iterations, 0);
 }
@@ -331,7 +314,7 @@ TEST_F(AllocatorImplTest, ForEachWithNullSizeLambda) {
     counters.emplace_back(alloc_.makeCounter(stat_name, StatName(), {}));
   }
   size_t num_iterations = 0;
-  alloc_.forEachCounter(nullptr, [&num_iterations](Stats::Counter& counter) {
+  alloc_.forEachCounter(nullptr, [&num_iterations](Counter& counter) {
     UNREFERENCED_PARAMETER(counter);
     ++num_iterations;
   });
@@ -343,7 +326,7 @@ TEST_F(AllocatorImplTest, ForEachWithNullSizeLambda) {
     gauges.emplace_back(alloc_.makeGauge(stat_name, StatName(), {}, Gauge::ImportMode::Accumulate));
   }
   num_iterations = 0;
-  alloc_.forEachGauge(nullptr, [&num_iterations](Stats::Gauge& gauge) {
+  alloc_.forEachGauge(nullptr, [&num_iterations](Gauge& gauge) {
     UNREFERENCED_PARAMETER(gauge);
     ++num_iterations;
   });
@@ -355,7 +338,7 @@ TEST_F(AllocatorImplTest, ForEachWithNullSizeLambda) {
     text_readouts.emplace_back(alloc_.makeTextReadout(stat_name, StatName(), {}));
   }
   num_iterations = 0;
-  alloc_.forEachTextReadout(nullptr, [&num_iterations](Stats::TextReadout& text_readout) {
+  alloc_.forEachTextReadout(nullptr, [&num_iterations](TextReadout& text_readout) {
     UNREFERENCED_PARAMETER(text_readout);
     ++num_iterations;
   });
@@ -434,9 +417,9 @@ TEST_F(AllocatorImplTest, AskForDeletedStat) {
 }
 
 TEST_F(AllocatorImplTest, ForEachSinkedCounter) {
-  std::unique_ptr<TestSinkPredicates> moved_sink_predicates =
-      std::make_unique<TestSinkPredicates>();
-  TestSinkPredicates* sink_predicates = moved_sink_predicates.get();
+  std::unique_ptr<TestUtil::TestSinkPredicates> moved_sink_predicates =
+      std::make_unique<TestUtil::TestSinkPredicates>();
+  TestUtil::TestSinkPredicates* sink_predicates = moved_sink_predicates.get();
   std::vector<CounterSharedPtr> sinked_counters;
   std::vector<CounterSharedPtr> unsinked_counters;
 
@@ -448,7 +431,7 @@ TEST_F(AllocatorImplTest, ForEachSinkedCounter) {
     auto stat_name = makeStat(absl::StrCat("counter.", idx));
     // sink every 3rd stat
     if ((idx + 1) % 3 == 0) {
-      sink_predicates->sinkedStatNames().insert(stat_name);
+      sink_predicates->add(stat_name);
       sinked_counters.emplace_back(alloc_.makeCounter(stat_name, StatName(), {}));
     } else {
       unsinked_counters.emplace_back(alloc_.makeCounter(stat_name, StatName(), {}));
@@ -462,8 +445,8 @@ TEST_F(AllocatorImplTest, ForEachSinkedCounter) {
   size_t num_iterations = 0;
   alloc_.forEachSinkedCounter(
       [&num_sinked_counters](std::size_t size) { num_sinked_counters = size; },
-      [&num_iterations, sink_predicates](Stats::Counter& counter) {
-        EXPECT_EQ(sink_predicates->sinkedStatNames().count(counter.statName()), 1);
+      [&num_iterations, sink_predicates](Counter& counter) {
+        EXPECT_TRUE(sink_predicates->has(counter.statName()));
         ++num_iterations;
       });
   EXPECT_EQ(num_sinked_counters, 3);
@@ -474,15 +457,15 @@ TEST_F(AllocatorImplTest, ForEachSinkedCounter) {
   num_iterations = 0;
   alloc_.forEachSinkedCounter(
       [&num_sinked_counters](std::size_t size) { num_sinked_counters = size; },
-      [&num_iterations](Stats::Counter&) { ++num_iterations; });
+      [&num_iterations](Counter&) { ++num_iterations; });
   EXPECT_EQ(num_sinked_counters, 0);
   EXPECT_EQ(num_iterations, 0);
 }
 
 TEST_F(AllocatorImplTest, ForEachSinkedGauge) {
-  std::unique_ptr<TestSinkPredicates> moved_sink_predicates =
-      std::make_unique<TestSinkPredicates>();
-  TestSinkPredicates* sink_predicates = moved_sink_predicates.get();
+  std::unique_ptr<TestUtil::TestSinkPredicates> moved_sink_predicates =
+      std::make_unique<TestUtil::TestSinkPredicates>();
+  TestUtil::TestSinkPredicates* sink_predicates = moved_sink_predicates.get();
   std::vector<GaugeSharedPtr> sinked_gauges;
   std::vector<GaugeSharedPtr> unsinked_gauges;
 
@@ -493,7 +476,7 @@ TEST_F(AllocatorImplTest, ForEachSinkedGauge) {
     auto stat_name = makeStat(absl::StrCat("gauge.", idx));
     // sink every 5th stat
     if ((idx + 1) % 5 == 0) {
-      sink_predicates->sinkedStatNames().insert(stat_name);
+      sink_predicates->add(stat_name);
       sinked_gauges.emplace_back(
           alloc_.makeGauge(stat_name, StatName(), {}, Gauge::ImportMode::Accumulate));
     } else {
@@ -508,9 +491,8 @@ TEST_F(AllocatorImplTest, ForEachSinkedGauge) {
   size_t num_sinked_gauges = 0;
   size_t num_iterations = 0;
   alloc_.forEachSinkedGauge([&num_sinked_gauges](std::size_t size) { num_sinked_gauges = size; },
-                            [&num_iterations, sink_predicates](Stats::Gauge& gauge) {
-                              EXPECT_EQ(sink_predicates->sinkedStatNames().count(gauge.statName()),
-                                        1);
+                            [&num_iterations, sink_predicates](Gauge& gauge) {
+                              EXPECT_TRUE(sink_predicates->has(gauge.statName()));
                               ++num_iterations;
                             });
   EXPECT_EQ(num_sinked_gauges, 2);
@@ -520,15 +502,15 @@ TEST_F(AllocatorImplTest, ForEachSinkedGauge) {
   sinked_gauges.clear();
   num_iterations = 0;
   alloc_.forEachSinkedGauge([&num_sinked_gauges](std::size_t size) { num_sinked_gauges = size; },
-                            [&num_iterations](Stats::Gauge&) { ++num_iterations; });
+                            [&num_iterations](Gauge&) { ++num_iterations; });
   EXPECT_EQ(num_sinked_gauges, 0);
   EXPECT_EQ(num_iterations, 0);
 }
 
 TEST_F(AllocatorImplTest, ForEachSinkedTextReadout) {
-  std::unique_ptr<TestSinkPredicates> moved_sink_predicates =
-      std::make_unique<TestSinkPredicates>();
-  TestSinkPredicates* sink_predicates = moved_sink_predicates.get();
+  std::unique_ptr<TestUtil::TestSinkPredicates> moved_sink_predicates =
+      std::make_unique<TestUtil::TestSinkPredicates>();
+  TestUtil::TestSinkPredicates* sink_predicates = moved_sink_predicates.get();
   std::vector<TextReadoutSharedPtr> sinked_text_readouts;
   std::vector<TextReadoutSharedPtr> unsinked_text_readouts;
 
@@ -539,7 +521,7 @@ TEST_F(AllocatorImplTest, ForEachSinkedTextReadout) {
     auto stat_name = makeStat(absl::StrCat("text_readout.", idx));
     // sink every 2nd stat
     if ((idx + 1) % 2 == 0) {
-      sink_predicates->sinkedStatNames().insert(stat_name);
+      sink_predicates->add(stat_name);
       sinked_text_readouts.emplace_back(alloc_.makeTextReadout(stat_name, StatName(), {}));
     } else {
       unsinked_text_readouts.emplace_back(alloc_.makeTextReadout(stat_name, StatName(), {}));
@@ -553,8 +535,8 @@ TEST_F(AllocatorImplTest, ForEachSinkedTextReadout) {
   size_t num_iterations = 0;
   alloc_.forEachSinkedTextReadout(
       [&num_sinked_text_readouts](std::size_t size) { num_sinked_text_readouts = size; },
-      [&num_iterations, sink_predicates](Stats::TextReadout& text_readout) {
-        EXPECT_EQ(sink_predicates->sinkedStatNames().count(text_readout.statName()), 1);
+      [&num_iterations, sink_predicates](TextReadout& text_readout) {
+        EXPECT_TRUE(sink_predicates->has(text_readout.statName()));
         ++num_iterations;
       });
   EXPECT_EQ(num_sinked_text_readouts, 5);
@@ -565,7 +547,7 @@ TEST_F(AllocatorImplTest, ForEachSinkedTextReadout) {
   num_iterations = 0;
   alloc_.forEachSinkedTextReadout(
       [&num_sinked_text_readouts](std::size_t size) { num_sinked_text_readouts = size; },
-      [&num_iterations](Stats::TextReadout&) { ++num_iterations; });
+      [&num_iterations](TextReadout&) { ++num_iterations; });
   EXPECT_EQ(num_sinked_text_readouts, 0);
   EXPECT_EQ(num_iterations, 0);
 }
diff --git a/test/common/stats/stat_test_utility.h b/test/common/stats/stat_test_utility.h
index f2efebe818ad4..05fb2fcbb773a 100644
--- a/test/common/stats/stat_test_utility.h
+++ b/test/common/stats/stat_test_utility.h
@@ -239,6 +239,34 @@ std::vector<uint8_t> serializeDeserializeNumber(uint64_t number);
 // Serializes a string into a MemBlock and then decodes it.
 void serializeDeserializeString(absl::string_view in);
 
+class TestSinkPredicates : public SinkPredicates {
+public:
+  ~TestSinkPredicates() override = default;
+
+  bool has(StatName name) { return sinked_stat_names_.find(name) != sinked_stat_names_.end(); }
+
+  // Note: The backing store for the StatName needs to live longer than the
+  // TestSinkPredicates object.
+  void add(StatName name) { sinked_stat_names_.insert(name); }
+
+  // SinkPredicates
+  bool includeCounter(const Counter& counter) override {
+    return sinked_stat_names_.find(counter.statName()) != sinked_stat_names_.end();
+  }
+  bool includeGauge(const Gauge& gauge) override {
+    return sinked_stat_names_.find(gauge.statName()) != sinked_stat_names_.end();
+  }
+  bool includeTextReadout(const TextReadout& text_readout) override {
+    return sinked_stat_names_.find(text_readout.statName()) != sinked_stat_names_.end();
+  }
+  bool includeHistogram(const Histogram& histogram) override {
+    return sinked_stat_names_.find(histogram.statName()) != sinked_stat_names_.end();
+  }
+
+private:
+  StatNameHashSet sinked_stat_names_;
+};
+
 } // namespace TestUtil
 } // 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 54982edc0f8c3..4d31938c5bb51 100644
--- a/test/common/stats/thread_local_store_test.cc
+++ b/test/common/stats/thread_local_store_test.cc
@@ -11,6 +11,8 @@
 #include "source/common/common/c_smart_ptr.h"
 #include "source/common/event/dispatcher_impl.h"
 #include "source/common/memory/stats.h"
+#include "source/common/runtime/runtime_impl.h"
+#include "source/common/stats/histogram_impl.h"
 #include "source/common/stats/stats_matcher_impl.h"
 #include "source/common/stats/symbol_table.h"
 #include "source/common/stats/tag_producer_impl.h"
@@ -131,8 +133,8 @@ class HistogramTest : public testing::Test {
   using NameHistogramMap = std::map<std::string, ParentHistogramSharedPtr>;
 
   HistogramTest()
-      : alloc_(symbol_table_), store_(std::make_unique<ThreadLocalStoreImpl>(alloc_)),
-        scope_(*store_->rootScope()) {
+      : pool_(symbol_table_), alloc_(symbol_table_),
+        store_(std::make_unique<ThreadLocalStoreImpl>(alloc_)), scope_(*store_->rootScope()) {
     store_->addSink(sink_);
     store_->initializeThreading(main_thread_dispatcher_, tls_);
   }
@@ -218,9 +220,16 @@ class HistogramTest : public testing::Test {
     }
   }
 
+  TestUtil::TestSinkPredicates& testSinkPredicatesOrDie() {
+    auto predicates = dynamic_cast<TestUtil::TestSinkPredicates*>(store_->sinkPredicates().ptr());
+    ASSERT(predicates != nullptr);
+    return *predicates;
+  }
+
   SymbolTableImpl symbol_table_;
   NiceMock<Event::MockDispatcher> main_thread_dispatcher_;
   NiceMock<ThreadLocal::MockInstance> tls_;
+  StatNamePool pool_;
   AllocatorImpl alloc_;
   MockSink sink_;
   ThreadLocalStoreImplPtr store_;
@@ -1976,6 +1985,7 @@ class TestSinkPredicates : public Stats::SinkPredicates {
   bool includeTextReadout(const Stats::TextReadout&) override {
     return (++num_text_readouts_) % 10 == 0;
   }
+  bool includeHistogram(const Stats::Histogram&) override { return false; }
 
 private:
   size_t num_counters_ = 0;
@@ -2024,5 +2034,188 @@ TEST_F(StatsThreadLocalStoreTest, SetSinkPredicates) {
   EXPECT_EQ(expected_sinked_stats, num_sinked_text_readouts);
 }
 
+enum class EnableIncludeHistograms { No = 0, Yes };
+class HistogramParameterisedTest : public HistogramTest,
+                                   public ::testing::WithParamInterface<EnableIncludeHistograms> {
+public:
+  HistogramParameterisedTest() { local_info_.node_.set_cluster(""); }
+
+protected:
+  void SetUp() override {
+    HistogramTest::SetUp();
+
+    // Set the feature flag in SetUp as store_ is constructed in HistogramTest::SetUp.
+    api_ = Api::createApiForTest(*store_);
+    ProtobufWkt::Struct base = TestUtility::parseYaml<ProtobufWkt::Struct>(
+        GetParam() == EnableIncludeHistograms::Yes ? R"EOF(
+    envoy.reloadable_features.enable_include_histograms: true
+    )EOF"
+                                                   : R"EOF(
+    envoy.reloadable_features.enable_include_histograms: false
+    )EOF");
+    envoy::config::bootstrap::v3::LayeredRuntime layered_runtime;
+    {
+      auto* layer = layered_runtime.add_layers();
+      layer->set_name("base");
+      layer->mutable_static_layer()->MergeFrom(base);
+    }
+    {
+      auto* layer = layered_runtime.add_layers();
+      layer->set_name("admin");
+      layer->mutable_admin_layer();
+    }
+    loader_ =
+        std::make_unique<Runtime::LoaderImpl>(dispatcher_, tls_, layered_runtime, local_info_,
+                                              *store_, generator_, validation_visitor_, *api_);
+  }
+
+  Event::MockDispatcher dispatcher_;
+  Api::ApiPtr api_;
+  NiceMock<LocalInfo::MockLocalInfo> local_info_;
+  Random::MockRandomGenerator generator_;
+  NiceMock<ProtobufMessage::MockValidationVisitor> validation_visitor_;
+  std::unique_ptr<Runtime::LoaderImpl> loader_;
+};
+
+TEST_P(HistogramParameterisedTest, ForEachSinkedHistogram) {
+  std::unique_ptr<TestUtil::TestSinkPredicates> test_sink_predicates =
+      std::make_unique<TestUtil::TestSinkPredicates>();
+  std::vector<std::reference_wrapper<Histogram>> sinked_histograms;
+  std::vector<std::reference_wrapper<Histogram>> unsinked_histograms;
+  auto scope = store_->rootScope();
+
+  const size_t num_stats = 11;
+  // Create some histograms before setting the predicates.
+  for (size_t idx = 0; idx < num_stats / 2; ++idx) {
+    auto name = absl::StrCat("histogram.", idx);
+    StatName stat_name = pool_.add(name);
+    //  sink every 3rd stat
+    if ((idx + 1) % 3 == 0) {
+      test_sink_predicates->add(stat_name);
+      sinked_histograms.emplace_back(
+          scope->histogramFromStatName(stat_name, Histogram::Unit::Unspecified));
+    } else {
+      unsinked_histograms.emplace_back(
+          scope->histogramFromStatName(stat_name, Histogram::Unit::Unspecified));
+    }
+  }
+
+  store_->setSinkPredicates(std::move(test_sink_predicates));
+  auto& sink_predicates = testSinkPredicatesOrDie();
+
+  // Create some histograms after setting the predicates.
+  for (size_t idx = num_stats / 2; idx < num_stats; ++idx) {
+    auto name = absl::StrCat("histogram.", idx);
+    StatName stat_name = pool_.add(name);
+    // sink every 3rd stat
+    if ((idx + 1) % 3 == 0) {
+      sink_predicates.add(stat_name);
+      sinked_histograms.emplace_back(
+          scope->histogramFromStatName(stat_name, Histogram::Unit::Unspecified));
+    } else {
+      unsinked_histograms.emplace_back(
+          scope->histogramFromStatName(stat_name, Histogram::Unit::Unspecified));
+    }
+  }
+
+  EXPECT_EQ(sinked_histograms.size(), 3);
+  EXPECT_EQ(unsinked_histograms.size(), 8);
+
+  size_t num_sinked_histograms = 0;
+  size_t num_iterations = 0;
+  store_->forEachSinkedHistogram(
+      [&num_sinked_histograms](std::size_t size) { num_sinked_histograms = size; },
+      [&num_iterations, &sink_predicates](ParentHistogram& histogram) {
+        if (GetParam() == EnableIncludeHistograms::Yes) {
+          EXPECT_TRUE(sink_predicates.has(histogram.statName()));
+        }
+        ++num_iterations;
+      });
+  if (GetParam() == EnableIncludeHistograms::Yes) {
+    EXPECT_EQ(num_sinked_histograms, 3);
+    EXPECT_EQ(num_iterations, 3);
+  } else {
+    EXPECT_EQ(num_sinked_histograms, 11);
+    EXPECT_EQ(num_iterations, 11);
+  }
+  // Verify that rejecting histograms removes them from the sink set.
+  envoy::config::metrics::v3::StatsConfig stats_config_;
+  stats_config_.mutable_stats_matcher()->set_reject_all(true);
+  store_->setStatsMatcher(std::make_unique<StatsMatcherImpl>(stats_config_, symbol_table_));
+  num_sinked_histograms = 0;
+  num_iterations = 0;
+  store_->forEachSinkedHistogram(
+      [&num_sinked_histograms](std::size_t size) { num_sinked_histograms = size; },
+      [&num_iterations](ParentHistogram&) { ++num_iterations; });
+  EXPECT_EQ(num_sinked_histograms, 0);
+  EXPECT_EQ(num_iterations, 0);
+}
+
+// Verify that histograms that are not flushed to sinks are merged in the call
+// to mergeHistograms
+TEST_P(HistogramParameterisedTest, UnsinkedHistogramsAreMerged) {
+  store_->setSinkPredicates(std::make_unique<TestUtil::TestSinkPredicates>());
+  auto& sink_predicates = testSinkPredicatesOrDie();
+  StatName stat_name = pool_.add("h1");
+  sink_predicates.add(stat_name);
+  auto scope = store_->rootScope();
+
+  auto& h1 = static_cast<ParentHistogramImpl&>(
+      scope->histogramFromStatName(stat_name, Histogram::Unit::Unspecified));
+  stat_name = pool_.add("h2");
+  auto& h2 = static_cast<ParentHistogramImpl&>(
+      scope->histogramFromStatName(stat_name, Histogram::Unit::Unspecified));
+
+  EXPECT_EQ("h1", h1.name());
+  EXPECT_EQ("h2", h2.name());
+  EXPECT_CALL(sink_, onHistogramComplete(Ref(h1), 5));
+  EXPECT_CALL(sink_, onHistogramComplete(Ref(h2), 5));
+
+  h1.recordValue(5);
+  h2.recordValue(5);
+
+  EXPECT_THAT(h1.cumulativeStatistics().bucketSummary(), HasSubstr(" B10: 0,"));
+  EXPECT_THAT(h2.cumulativeStatistics().bucketSummary(), HasSubstr(" B10: 0,"));
+
+  // Verify that all the histograms have not been merged yet.
+  EXPECT_EQ(h1.used(), false);
+  EXPECT_EQ(h2.used(), false);
+
+  store_->mergeHistograms([this, &sink_predicates]() -> void {
+    size_t num_iterations = 0;
+    size_t num_sinked_histograms = 0;
+    store_->forEachSinkedHistogram(
+        [&num_sinked_histograms](std::size_t size) { num_sinked_histograms = size; },
+        [&num_iterations, &sink_predicates](ParentHistogram& histogram) {
+          if (GetParam() == EnableIncludeHistograms::Yes) {
+            EXPECT_TRUE(sink_predicates.has(histogram.statName()));
+          }
+          ++num_iterations;
+        });
+    if (GetParam() == EnableIncludeHistograms::Yes) {
+      EXPECT_EQ(num_sinked_histograms, 1);
+      EXPECT_EQ(num_iterations, 1);
+    } else {
+      EXPECT_EQ(num_sinked_histograms, 2);
+      EXPECT_EQ(num_iterations, 2);
+    }
+  });
+
+  EXPECT_THAT(h1.cumulativeStatistics().bucketSummary(), HasSubstr(" B10: 1,"));
+  EXPECT_THAT(h2.cumulativeStatistics().bucketSummary(), HasSubstr(" B10: 1,"));
+  EXPECT_EQ(h1.cumulativeStatistics().bucketSummary(), h2.cumulativeStatistics().bucketSummary());
+
+  // Verify that all the histograms have been merged.
+  EXPECT_EQ(h1.used(), true);
+  EXPECT_EQ(h2.used(), true);
+}
+
+INSTANTIATE_TEST_SUITE_P(HistogramParameterisedTestGroup, HistogramParameterisedTest,
+                         testing::Values(EnableIncludeHistograms::Yes, EnableIncludeHistograms::No),
+                         [](const testing::TestParamInfo<EnableIncludeHistograms>& info) {
+                           return info.param == EnableIncludeHistograms::No
+                                      ? "DisableIncludeHistograms"
+                                      : "EnableIncludeHistograms";
+                         });
 } // namespace Stats
 } // namespace Envoy
diff --git a/test/integration/server.h b/test/integration/server.h
index 950b006aea14b..17301e6a5e60e 100644
--- a/test/integration/server.h
+++ b/test/integration/server.h
@@ -307,9 +307,14 @@ class TestIsolatedStoreImpl : public StoreRoot {
     Thread::LockGuard lock(lock_);
     store_.forEachSinkedTextReadout(f_size, f_stat);
   }
+  void forEachSinkedHistogram(Stats::SizeFn f_size, StatFn<ParentHistogram> f_stat) const override {
+    Thread::LockGuard lock(lock_);
+    store_.forEachSinkedHistogram(f_size, f_stat);
+  }
   void setSinkPredicates(std::unique_ptr<SinkPredicates>&& sink_predicates) override {
     UNREFERENCED_PARAMETER(sink_predicates);
   }
+  OptRef<SinkPredicates> sinkPredicates() override { return OptRef<SinkPredicates>{}; }
   void deliverHistogramToSinks(const Histogram& histogram, uint64_t value) override {
     Thread::LockGuard lock(lock_);
     store_.deliverHistogramToSinks(histogram, value);
diff --git a/test/mocks/stats/mocks.h b/test/mocks/stats/mocks.h
index ce4c5cba87586..84372a7959b12 100644
--- a/test/mocks/stats/mocks.h
+++ b/test/mocks/stats/mocks.h
@@ -270,6 +270,7 @@ class MockSinkPredicates : public SinkPredicates {
   MOCK_METHOD(bool, includeCounter, (const Counter&));
   MOCK_METHOD(bool, includeGauge, (const Gauge&));
   MOCK_METHOD(bool, includeTextReadout, (const TextReadout&));
+  MOCK_METHOD(bool, includeHistogram, (const Histogram&));
 };
 
 class MockStore;
@@ -315,6 +316,7 @@ class MockStore : public TestUtil::TestStore {
   MOCK_METHOD(void, forEachGauge, (SizeFn, StatFn<Gauge>), (const));
   MOCK_METHOD(void, forEachTextReadout, (SizeFn, StatFn<TextReadout>), (const));
   MOCK_METHOD(void, forEachHistogram, (SizeFn, StatFn<ParentHistogram>), (const));
+  MOCK_METHOD(void, forEachSinkedHistogram, (SizeFn, StatFn<ParentHistogram>), (const));
   MOCK_METHOD(Counter&, counter, (const std::string&));
   MOCK_METHOD(Gauge&, gauge, (const std::string&, Gauge::ImportMode));
   MOCK_METHOD(Histogram&, histogram, (const std::string&, Histogram::Unit));
diff --git a/test/server/server_stats_flush_benchmark_test.cc b/test/server/server_stats_flush_benchmark_test.cc
index 4b09ea31aaebf..138f30ddc0d60 100644
--- a/test/server/server_stats_flush_benchmark_test.cc
+++ b/test/server/server_stats_flush_benchmark_test.cc
@@ -26,11 +26,13 @@ class TestSinkPredicates : public Stats::SinkPredicates {
   bool includeTextReadout(const Stats::TextReadout&) override {
     return (++num_text_readouts_) % 10 == 0;
   }
+  bool includeHistogram(const Stats::Histogram&) override { return (++num_histograms_) % 10 == 0; }
 
 private:
   size_t num_counters_ = 0;
   size_t num_gauges_ = 0;
   size_t num_text_readouts_ = 0;
+  size_t num_histograms_ = 0;
 };
 
 class StatsSinkFlushSpeedTest {
diff --git a/test/server/server_test.cc b/test/server/server_test.cc
index 3b24e2feccd2a..9b30dd4dfa211 100644
--- a/test/server/server_test.cc
+++ b/test/server/server_test.cc
@@ -92,7 +92,7 @@ TEST(ServerInstanceUtil, flushHelper) {
   NiceMock<Stats::MockStore> mock_store;
   Stats::ParentHistogramSharedPtr parent_histogram(new Stats::MockParentHistogram());
   std::vector<Stats::ParentHistogramSharedPtr> parent_histograms = {parent_histogram};
-  ON_CALL(mock_store, forEachHistogram)
+  ON_CALL(mock_store, forEachSinkedHistogram)
       .WillByDefault([&](std::function<void(std::size_t)> f_size,
                          std::function<void(Stats::ParentHistogram&)> f_stat) {
         if (f_size != nullptr) {