Stats: Add iteration and sink predicates for histograms

envoy/server/instance.h

@@ -277,7 +277,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;

envoy/stats/allocator.h

@@ -96,7 +96,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;
 

envoy/stats/sink.h

@@ -69,6 +69,11 @@ 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.
+   */
+  virtual bool includeHistogram(const Histogram& histogram) PURE;
 };
 
 /**

envoy/stats/store.h

@@ -61,6 +61,7 @@ class Store : public Scope {
   virtual void forEachCounter(SizeFn f_size, StatFn<Counter> f_stat) const PURE;
   virtual void forEachGauge(SizeFn f_size, StatFn<Gauge> f_stat) const PURE;
   virtual void forEachTextReadout(SizeFn f_size, StatFn<TextReadout> f_stat) const PURE;
+  virtual void forEachHistogram(SizeFn f_size, StatFn<ParentHistogram> f_stat) const PURE;
 
   /**
    * Iterate over all stats that need to be flushed to sinks. Note, that implementations can
@@ -73,6 +74,7 @@ class Store : public Scope {
   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;
 };
 
 using StorePtr = std::unique_ptr<Store>;
@@ -133,7 +135,7 @@ 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.

source/common/stats/isolated_store_impl.h

@@ -227,6 +227,11 @@ class IsolatedStoreImpl : public StoreImpl {
     text_readouts_.forEachStat(f_size, f_stat);
   }
 
+  void forEachHistogram(SizeFn f_size, StatFn<ParentHistogram> f_stat) const override {
+    UNREFERENCED_PARAMETER(f_size);
+    UNREFERENCED_PARAMETER(f_stat);
+  }
+
   void forEachSinkedCounter(SizeFn f_size, StatFn<Counter> f_stat) const override {
     forEachCounter(f_size, f_stat);
   }
@@ -239,6 +244,11 @@ class IsolatedStoreImpl : public StoreImpl {
     forEachTextReadout(f_size, f_stat);
   }
 
+  void forEachSinkedHistogram(SizeFn f_size, StatFn<ParentHistogram> f_stat) const override {
+    UNREFERENCED_PARAMETER(f_size);
+    UNREFERENCED_PARAMETER(f_stat);
+  }
+
 private:
   IsolatedStoreImpl(std::unique_ptr<SymbolTable>&& symbol_table);
 

source/common/stats/thread_local_store.cc

@@ -90,6 +90,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());
     }
   }
 }
@@ -181,14 +182,10 @@ std::vector<TextReadoutSharedPtr> ThreadLocalStoreImpl::textReadouts() const {
 
 std::vector<ParentHistogramSharedPtr> ThreadLocalStoreImpl::histograms() const {
   std::vector<ParentHistogramSharedPtr> ret;
-  Thread::LockGuard lock(hist_mutex_);
-  {
-    ret.reserve(histogram_set_.size());
-    for (const auto& histogram_ptr : histogram_set_) {
-      ret.emplace_back(histogram_ptr);
-    }
-  }
-
+  forEachHistogram([&ret](std::size_t size) mutable { ret.reserve(size); },
+                   [&ret](ParentHistogram& histogram) mutable {
+                     ret.emplace_back(ParentHistogramSharedPtr(&histogram));
+                   });
   return ret;
 }
 
@@ -221,6 +218,7 @@ void ThreadLocalStoreImpl::shutdownThreading() {
     histogram->setShuttingDown(true);
   }
   histogram_set_.clear();
+  sinked_histograms_.clear();
 }
 
 void ThreadLocalStoreImpl::mergeHistograms(PostMergeCb merge_complete_cb) {
@@ -243,9 +241,7 @@ void ThreadLocalStoreImpl::mergeHistograms(PostMergeCb merge_complete_cb) {
 
 void ThreadLocalStoreImpl::mergeInternal(PostMergeCb merge_complete_cb) {
   if (!shutting_down_) {
-    for (const ParentHistogramSharedPtr& histogram : histograms()) {
-      histogram->merge();
-    }
+    forEachHistogram(nullptr, [](ParentHistogram& histogram) { histogram.merge(); });
     merge_complete_cb();
     merge_in_progress_ = false;
   }
@@ -678,6 +674,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());
+          }
         }
       }
     }
@@ -867,6 +867,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;
   }
@@ -966,6 +967,16 @@ void ThreadLocalStoreImpl::forEachTextReadout(SizeFn f_size, StatFn<TextReadout>
   alloc_.forEachTextReadout(f_size, f_stat);
 }
 
+void ThreadLocalStoreImpl::forEachHistogram(SizeFn f_size, StatFn<ParentHistogram> f_stat) const {
+  Thread::LockGuard lock(hist_mutex_);
+  if (f_size != nullptr) {
+    f_size(histogram_set_.size());
+  }
+  for (auto histogram : histogram_set_) {
+    f_stat(*histogram);
+  }
+}
+
 void ThreadLocalStoreImpl::forEachSinkedCounter(SizeFn f_size, StatFn<Counter> f_stat) const {
   alloc_.forEachSinkedCounter(f_size, f_stat);
 }
@@ -979,11 +990,35 @@ 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()) {
+    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);
+      }
+    }
   }
 }
 

source/common/stats/thread_local_store.h

@@ -247,6 +247,7 @@ class ThreadLocalStoreImpl : Logger::Loggable<Logger::Id::stats>, public StoreRo
   void forEachCounter(SizeFn f_size, StatFn<Counter> f_stat) const override;
   void forEachGauge(SizeFn f_size, StatFn<Gauge> f_stat) const override;
   void forEachTextReadout(SizeFn f_size, StatFn<TextReadout> f_stat) const override;
+  void forEachHistogram(SizeFn f_size, StatFn<ParentHistogram> f_stat) const override;
 
   // Stats::StoreRoot
   void addSink(Sink& sink) override { timer_sinks_.push_back(sink); }
@@ -265,6 +266,7 @@ 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;
 
@@ -532,6 +534,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

source/server/server.cc

@@ -180,34 +180,38 @@ void InstanceImpl::failHealthcheck(bool fail) {
 
 MetricSnapshotImpl::MetricSnapshotImpl(Stats::Store& store, TimeSource& time_source) {
   store.forEachSinkedCounter(
-      [this](std::size_t size) mutable {
+      [this](std::size_t size) {
         snapped_counters_.reserve(size);
         counters_.reserve(size);
       },
-      [this](Stats::Counter& counter) mutable {
+      [this](Stats::Counter& counter) {
         snapped_counters_.push_back(Stats::CounterSharedPtr(&counter));
         counters_.push_back({counter.latch(), counter});
       });
 
   store.forEachSinkedGauge(
-      [this](std::size_t size) mutable {
+      [this](std::size_t size) {
         snapped_gauges_.reserve(size);
         gauges_.reserve(size);
       },
-      [this](Stats::Gauge& gauge) mutable {
+      [this](Stats::Gauge& gauge) {
         ASSERT(gauge.importMode() != Stats::Gauge::ImportMode::Uninitialized);
         snapped_gauges_.push_back(Stats::GaugeSharedPtr(&gauge));
         gauges_.push_back(gauge);
       });
 
-  snapped_histograms_ = store.histograms();
-  histograms_.reserve(snapped_histograms_.size());
-  for (const auto& histogram : snapped_histograms_) {
-    histograms_.push_back(*histogram);
-  }
+  store.forEachSinkedHistogram(
+      [this](std::size_t size) {
+        snapped_histograms_.reserve(size);
+        histograms_.reserve(size);
+      },
+      [this](Stats::ParentHistogram& histogram) {
+        snapped_histograms_.push_back(Stats::ParentHistogramSharedPtr(&histogram));
+        histograms_.push_back(histogram);
+      });
 
   store.forEachSinkedTextReadout(
-      [this](std::size_t size) mutable {
+      [this](std::size_t size) {
         snapped_text_readouts_.reserve(size);
         text_readouts_.reserve(size);
       },

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"
 
@@ -44,26 +45,6 @@ class AllocatorImplTest : public testing::Test {
   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");
@@ -434,9 +415,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<Stats::TestUtil::TestSinkPredicates> moved_sink_predicates =
+      std::make_unique<Stats::TestUtil::TestSinkPredicates>();
+  Stats::TestUtil::TestSinkPredicates* sink_predicates = moved_sink_predicates.get();
   std::vector<CounterSharedPtr> sinked_counters;
   std::vector<CounterSharedPtr> unsinked_counters;
 
@@ -480,9 +461,9 @@ TEST_F(AllocatorImplTest, ForEachSinkedCounter) {
 }
 
 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<Stats::TestUtil::TestSinkPredicates> moved_sink_predicates =
+      std::make_unique<Stats::TestUtil::TestSinkPredicates>();
+  Stats::TestUtil::TestSinkPredicates* sink_predicates = moved_sink_predicates.get();
   std::vector<GaugeSharedPtr> sinked_gauges;
   std::vector<GaugeSharedPtr> unsinked_gauges;
 
@@ -526,9 +507,9 @@ TEST_F(AllocatorImplTest, ForEachSinkedGauge) {
 }
 
 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<Stats::TestUtil::TestSinkPredicates> moved_sink_predicates =
+      std::make_unique<Stats::TestUtil::TestSinkPredicates>();
+  Stats::TestUtil::TestSinkPredicates* sink_predicates = moved_sink_predicates.get();
   std::vector<TextReadoutSharedPtr> sinked_text_readouts;
   std::vector<TextReadoutSharedPtr> unsinked_text_readouts;
 

test/common/stats/stat_test_utility.h

@@ -219,6 +219,29 @@ 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;
+  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();
+  }
+  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