Stats: Filter stats to be flushed to sinks

envoy/server/instance.h

@@ -2,6 +2,7 @@
 
 #include <chrono>
 #include <cstdint>
+#include <memory>
 #include <string>
 
 #include "envoy/access_log/access_log.h"
@@ -31,6 +32,11 @@
 #include "envoy/upstream/cluster_manager.h"
 
 namespace Envoy {
+
+namespace Stats {
+class SinkPredicates;
+}
+
 namespace Server {
 
 /**
@@ -269,6 +275,9 @@ class Instance {
    * TODO(mattklein123): This can be removed when version 1.20.0 is no longer supported.
    */
   virtual bool enableReusePortDefault() PURE;
+
+  virtual void
+  setSinkPredicates(std::unique_ptr<Envoy::Stats::SinkPredicates> sink_predicates) PURE;
 };
 
 } // namespace Server

envoy/stats/allocator.h

@@ -18,6 +18,9 @@
 namespace Envoy {
 namespace Stats {
 
+class Sink;
+class SinkPredicates;
+
 /**
  * Abstract interface for allocating statistics. Implementations can
  * be created utilizing a single fixed-size block suitable for
@@ -84,6 +87,18 @@ class Allocator {
   virtual void forEachTextReadout(std::function<void(std::size_t)> f_size,
                                   std::function<void(Stats::TextReadout&)> f_stat) const PURE;
 
+  virtual void forEachSinkedCounter(std::function<void(std::size_t)> f_size,
+                                    std::function<void(Stats::Counter&)> f_stat) const PURE;
+  virtual void forEachSinkedGauge(std::function<void(std::size_t)> f_size,
+                                  std::function<void(Stats::Gauge&)> f_stat) const PURE;
+  virtual void forEachSinkedTextReadout(std::function<void(std::size_t)> f_size,
+                                        std::function<void(Stats::TextReadout&)> f_stat) const PURE;
+
+  /**
+   * Set the predicates to filter stats for sink.
+   */
+  virtual void setSinkPredicates(const SinkPredicates& sink_predicates) 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.

envoy/stats/sink.h

@@ -48,6 +48,23 @@ class MetricSnapshot {
   virtual SystemTime snapshotTime() const PURE;
 };
 
+/**
+ * A class to define predicates to filter stats for flushing to sinks.
+ */
+class SinkPredicates {
+public:
+  virtual ~SinkPredicates() = default;
+
+  /// @return true if @param counter needs to be flushed to sinks.
+  virtual bool includeCounter(const Counter& counter) const PURE;
+  /// @return true if @param gague needs to be flushed to sinks.
+  virtual bool includeGauge(const Gauge& gauge) const PURE;
+  /// @return true if @param text_readout needs to be flushed to sinks.
+  virtual bool includeTextReadout(const TextReadout& text_readout) const PURE;
+  /// @return true if @param histogram needs to be flushed to sinks.
+  virtual bool includeHistogram(const Histogram& histogram) const PURE;
+};
+
 /**
  * A sink for stats. Each sink is responsible for writing stats to a backing store.
  */

envoy/stats/store.h

@@ -24,6 +24,7 @@ class Instance;
 namespace Stats {
 
 class Sink;
+class SinkPredicates;
 
 /**
  * A store for all known counters, gauges, and timers.
@@ -65,6 +66,21 @@ class Store : public Scope {
 
   virtual void forEachTextReadout(std::function<void(std::size_t)> f_size,
                                   std::function<void(Stats::TextReadout&)> f_stat) const PURE;
+
+  virtual void forEachHistogram(std::function<void(std::size_t)> f_size,
+                                std::function<void(Stats::ParentHistogram&)> f_stat) const PURE;
+  /**
+   * Iterate over all stats that need to be flushed for sink.
+   */
+  virtual void forEachSinkedCounter(std::function<void(std::size_t)> f_size,
+                                    std::function<void(Stats::Counter&)> f_stat) const PURE;
+  virtual void forEachSinkedGauge(std::function<void(std::size_t)> f_size,
+                                  std::function<void(Stats::Gauge&)> f_stat) const PURE;
+  virtual void forEachSinkedTextReadout(std::function<void(std::size_t)> f_size,
+                                        std::function<void(Stats::TextReadout&)> f_stat) const PURE;
+  virtual void
+  forEachSinkedHistogram(std::function<void(std::size_t)> f_size,
+                         std::function<void(Stats::ParentHistogram&)> f_stat) const PURE;
 };
 
 using StorePtr = std::unique_ptr<Store>;
@@ -123,6 +139,11 @@ class StoreRoot : public Store {
    * method would be asserted.
    */
   virtual void mergeHistograms(PostMergeCb merge_complete_cb) PURE;
+
+  /**
+   * Set the predicates to filter stats for sink.
+   */
+  virtual void setSinkPredicates(const SinkPredicates& sink_predicates) PURE;
 };
 
 using StoreRootPtr = std::unique_ptr<StoreRoot>;

source/common/stats/allocator_impl.cc

@@ -3,6 +3,7 @@
 #include <algorithm>
 #include <cstdint>
 
+#include "envoy/stats/sink.h"
 #include "envoy/stats/stats.h"
 #include "envoy/stats/symbol_table.h"
 
@@ -144,6 +145,7 @@ class CounterImpl : public StatsSharedImpl<Counter> {
   void removeFromSetLockHeld() ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) override {
     const size_t count = alloc_.counters_.erase(statName());
     ASSERT(count == 1);
+    alloc_.sinked_counters_.erase(this);
   }
 
   // Stats::Counter
@@ -188,6 +190,7 @@ class GaugeImpl : public StatsSharedImpl<Gauge> {
   void removeFromSetLockHeld() override ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) {
     const size_t count = alloc_.gauges_.erase(statName());
     ASSERT(count == 1);
+    alloc_.sinked_gauges_.erase(this);
   }
 
   // Stats::Gauge
@@ -260,6 +263,7 @@ class TextReadoutImpl : public StatsSharedImpl<TextReadout> {
   void removeFromSetLockHeld() ABSL_EXCLUSIVE_LOCKS_REQUIRED(alloc_.mutex_) override {
     const size_t count = alloc_.text_readouts_.erase(statName());
     ASSERT(count == 1);
+    alloc_.sinked_text_readouts_.erase(this);
   }
 
   // Stats::TextReadout
@@ -289,6 +293,11 @@ CounterSharedPtr AllocatorImpl::makeCounter(StatName name, StatName tag_extracte
   }
   auto counter = CounterSharedPtr(makeCounterInternal(name, tag_extracted_name, stat_name_tags));
   counters_.insert(counter.get());
+  // Add counter to sinked_counters_ if it matches the sink predicate.
+  if (sink_predicates_ && sink_predicates_->includeCounter(*counter)) {
+    auto val = sinked_counters_.insert(counter.get());
+    ASSERT(val.second);
+  }
   return counter;
 }
 
@@ -305,6 +314,11 @@ GaugeSharedPtr AllocatorImpl::makeGauge(StatName name, StatName tag_extracted_na
   auto gauge =
       GaugeSharedPtr(new GaugeImpl(name, *this, tag_extracted_name, stat_name_tags, import_mode));
   gauges_.insert(gauge.get());
+  // Add gauge to sinked_gauges_ if it matches the sink predicate.
+  if (sink_predicates_ && sink_predicates_->includeGauge(*gauge)) {
+    auto val = sinked_gauges_.insert(gauge.get());
+    ASSERT(val.second);
+  }
   return gauge;
 }
 
@@ -320,6 +334,11 @@ TextReadoutSharedPtr AllocatorImpl::makeTextReadout(StatName name, StatName tag_
   auto text_readout =
       TextReadoutSharedPtr(new TextReadoutImpl(name, *this, tag_extracted_name, stat_name_tags));
   text_readouts_.insert(text_readout.get());
+  // Add text_readout to sinked_text_readouts_ if it matches the sink predicate.
+  if (sink_predicates_ && sink_predicates_->includeTextReadout(*text_readout)) {
+    auto val = sinked_text_readouts_.insert(text_readout.get());
+    ASSERT(val.second);
+  }
   return text_readout;
 }
 
@@ -369,6 +388,71 @@ void AllocatorImpl::forEachTextReadout(std::function<void(std::size_t)> f_size,
   }
 }
 
+void AllocatorImpl::forEachSinkedCounter(std::function<void(std::size_t)> f_size,
+                                         std::function<void(Stats::Counter&)> f_stat) const {
+  if (sink_predicates_ != nullptr) {
+    Thread::LockGuard lock(mutex_);
+    f_size(sinked_counters_.size());
+    for (auto counter : sinked_counters_) {
+      f_stat(*counter);
+    }
+  } else {
+    forEachCounter(f_size, f_stat);
+  }
+}
+
+void AllocatorImpl::forEachSinkedGauge(std::function<void(std::size_t)> f_size,
+                                       std::function<void(Stats::Gauge&)> f_stat) const {
+  if (sink_predicates_ != nullptr) {
+    Thread::LockGuard lock(mutex_);
+    f_size(sinked_gauges_.size());
+    for (auto gauge : sinked_gauges_) {
+      f_stat(*gauge);
+    }
+  } else {
+    forEachGauge(f_size, f_stat);
+  }
+}
+
+void AllocatorImpl::forEachSinkedTextReadout(
+    std::function<void(std::size_t)> f_size,
+    std::function<void(Stats::TextReadout&)> f_stat) const {
+  if (sink_predicates_ != nullptr) {
+    Thread::LockGuard lock(mutex_);
+    f_size(sinked_text_readouts_.size());
+    for (auto text_readout : sinked_text_readouts_) {
+      f_stat(*text_readout);
+    }
+  } else {
+    forEachTextReadout(f_size, f_stat);
+  }
+}
+
+void AllocatorImpl::setSinkPredicates(const SinkPredicates& sink_predicates) {
+  Thread::LockGuard lock(mutex_);
+  ASSERT(sink_predicates_ == nullptr);
+  sink_predicates_ = &sink_predicates;
+
+  // Add counters to the set of sinked counters.
+  for (auto& counter : counters_) {
+    if (sink_predicates_->includeCounter(*counter)) {
+      sinked_counters_.emplace(counter);
+    }
+  }
+  // Add gauges to the set of sinked gauges.
+  for (auto& gauge : gauges_) {
+    if (sink_predicates_->includeGauge(*gauge)) {
+      sinked_gauges_.insert(gauge);
+    }
+  }
+  // Add text_readouts to the set of sinked text readouts.
+  for (auto& text_readout : text_readouts_) {
+    if (sink_predicates_->includeTextReadout(*text_readout)) {
+      sinked_text_readouts_.insert(text_readout);
+    }
+  }
+}
+
 void AllocatorImpl::markCounterForDeletion(const CounterSharedPtr& counter) {
   Thread::LockGuard lock(mutex_);
   auto iter = counters_.find(counter->statName());
@@ -380,6 +464,7 @@ void AllocatorImpl::markCounterForDeletion(const CounterSharedPtr& counter) {
   // Duplicates are ASSERTed in ~AllocatorImpl.
   deleted_counters_.emplace_back(*iter);
   counters_.erase(iter);
+  sinked_counters_.erase(counter.get());
 }
 
 void AllocatorImpl::markGaugeForDeletion(const GaugeSharedPtr& gauge) {
@@ -393,6 +478,7 @@ void AllocatorImpl::markGaugeForDeletion(const GaugeSharedPtr& gauge) {
   // Duplicates are ASSERTed in ~AllocatorImpl.
   deleted_gauges_.emplace_back(*iter);
   gauges_.erase(iter);
+  sinked_gauges_.erase(gauge.get());
 }
 
 void AllocatorImpl::markTextReadoutForDeletion(const TextReadoutSharedPtr& text_readout) {
@@ -406,6 +492,7 @@ void AllocatorImpl::markTextReadoutForDeletion(const TextReadoutSharedPtr& text_
   // Duplicates are ASSERTed in ~AllocatorImpl.
   deleted_text_readouts_.emplace_back(*iter);
   text_readouts_.erase(iter);
+  sinked_text_readouts_.erase(text_readout.get());
 }
 
 } // namespace Stats

source/common/stats/allocator_impl.h

@@ -15,6 +15,8 @@
 namespace Envoy {
 namespace Stats {
 
+class SinkPredicates;
+
 class AllocatorImpl : public Allocator {
 public:
   static const char DecrementToZeroSyncPoint[];
@@ -42,6 +44,14 @@ class AllocatorImpl : public Allocator {
   void forEachTextReadout(std::function<void(std::size_t)>,
                           std::function<void(Stats::TextReadout&)>) const override;
 
+  void forEachSinkedCounter(std::function<void(std::size_t)> f_size,
+                            std::function<void(Stats::Counter&)> f_stat) const override;
+  void forEachSinkedGauge(std::function<void(std::size_t)> f_size,
+                          std::function<void(Stats::Gauge&)> f_stat) const override;
+  void forEachSinkedTextReadout(std::function<void(std::size_t)> f_size,
+                                std::function<void(Stats::TextReadout&)> f_stat) const override;
+
+  void setSinkPredicates(SinkPredicates const& sink_predicates) override;
 #ifndef ENVOY_CONFIG_COVERAGE
   void debugPrint();
 #endif
@@ -93,6 +103,15 @@ class AllocatorImpl : public Allocator {
   std::vector<GaugeSharedPtr> deleted_gauges_ ABSL_GUARDED_BY(mutex_);
   std::vector<TextReadoutSharedPtr> deleted_text_readouts_ ABSL_GUARDED_BY(mutex_);
 
+  template <typename StatType>
+  using StatPointerSet = absl::flat_hash_set<StatType*, absl::Hash<StatType*>>;
+  // Stat pointers that participate in the flush to sink process.
+  StatPointerSet<Counter> sinked_counters_ ABSL_GUARDED_BY(mutex_);
+  StatPointerSet<Gauge> sinked_gauges_ ABSL_GUARDED_BY(mutex_);
+  StatPointerSet<TextReadout> sinked_text_readouts_ ABSL_GUARDED_BY(mutex_);
+
+  // Predicates used to filter stats to be flushed.
+  const SinkPredicates* sink_predicates_ = nullptr;
   SymbolTable& symbol_table_;
 
   Thread::ThreadSynchronizer sync_;

source/common/stats/isolated_store_impl.h

@@ -103,7 +103,9 @@ template <class Base> class IsolatedStatsCache {
 
   void forEachStat(std::function<void(std::size_t)> f_size,
                    std::function<void(Base&)> f_stat) const {
-    f_size(stats_.size());
+    if (f_size != nullptr) {
+      f_size(stats_.size());
+    }
     for (auto const& stat : stats_) {
       f_stat(*stat.second);
     }
@@ -229,6 +231,33 @@ class IsolatedStoreImpl : public StoreImpl {
     text_readouts_.forEachStat(f_size, f_stat);
   }
 
+  void forEachHistogram(std::function<void(std::size_t)> f_size,
+                        std::function<void(Stats::ParentHistogram&)> f_stat) const override {
+    UNREFERENCED_PARAMETER(f_size);
+    UNREFERENCED_PARAMETER(f_stat);
+  }
+
+  void forEachSinkedCounter(std::function<void(std::size_t)> f_size,
+                            std::function<void(Stats::Counter&)> f_stat) const override {
+    forEachCounter(f_size, f_stat);
+  }
+
+  void forEachSinkedGauge(std::function<void(std::size_t)> f_size,
+                          std::function<void(Stats::Gauge&)> f_stat) const override {
+    forEachGauge(f_size, f_stat);
+  }
+
+  void forEachSinkedTextReadout(std::function<void(std::size_t)> f_size,
+                                std::function<void(Stats::TextReadout&)> f_stat) const override {
+    forEachTextReadout(f_size, f_stat);
+  }
+
+  void forEachSinkedHistogram(std::function<void(std::size_t)> f_size,
+                              std::function<void(Stats::ParentHistogram&)> f_stat) const override {
+    (void)f_size;
+    (void)f_stat;
+  }
+
 private:
   IsolatedStoreImpl(std::unique_ptr<SymbolTable>&& symbol_table);