stats: rework stat sink flushing to centralize counter latching

include/envoy/stats/BUILD

@@ -15,7 +15,6 @@ envoy_cc_library(
         "histogram.h",
         "scope.h",
         "sink.h",
-        "source.h",
         "stat_data_allocator.h",
         "stats.h",
         "stats_matcher.h",

include/envoy/stats/sink.h

@@ -4,25 +4,51 @@
 #include <memory>
 
 #include "envoy/common/pure.h"
+#include "envoy/stats/histogram.h"
+#include "envoy/stats/stats.h"
 
 namespace Envoy {
 namespace Stats {
 
 class Histogram;
-class Source;
+
+class MetricSnapshot {
+public:
+  struct CounterSnapshot {
+    uint64_t delta_;
+    std::reference_wrapper<const Counter> counter_;
+  };
+
+  virtual ~MetricSnapshot() = default;
+
+  /**
+   * @return a snapshot of all counters with pre-latched deltas.
+   */
+  virtual const std::vector<CounterSnapshot>& counters() PURE;
+
+  /**
+   * @return a snapshot of all gauges.
+   */
+  virtual const std::vector<std::reference_wrapper<const Gauge>>& gauges() PURE;
+
+  /**
+   * @return a snapshot of all histograms.
+   */
+  virtual const std::vector<std::reference_wrapper<const ParentHistogram>>& histograms() PURE;
+};
 
 /**
  * A sink for stats. Each sink is responsible for writing stats to a backing store.
  */
 class Sink {
 public:
-  virtual ~Sink() {}
+  virtual ~Sink() = default;
 
   /**
    * Periodic metric flush to the sink.
-   * @param source interface through which the sink can access all metrics being flushed.
+   * @param snapshot interface through which the sink can access all metrics being flushed.
    */
-  virtual void flush(Source& source) PURE;
+  virtual void flush(MetricSnapshot& snapshot) PURE;
 
   /**
    * Flush a single histogram sample. Note: this call is called synchronously as a part of recording
@@ -33,7 +59,7 @@ class Sink {
   virtual void onHistogramComplete(const Histogram& histogram, uint64_t value) PURE;
 };
 
-typedef std::unique_ptr<Sink> SinkPtr;
+using SinkPtr = std::unique_ptr<Sink>;
 
 } // namespace Stats
 } // namespace Envoy

include/envoy/stats/store.h

@@ -22,7 +22,6 @@ class Instance;
 namespace Stats {
 
 class Sink;
-class Source;
 
 /**
  * A store for all known counters, gauges, and timers.
@@ -95,12 +94,6 @@ class StoreRoot : public Store {
    * method would be asserted.
    */
   virtual void mergeHistograms(PostMergeCb merge_complete_cb) PURE;
-
-  /**
-   * Returns the Source to provide cached metrics.
-   * @return Source& the source.
-   */
-  virtual Source& source() PURE;
 };
 
 typedef std::unique_ptr<StoreRoot> StoreRootPtr;

source/common/stats/BUILD

@@ -84,18 +84,6 @@ envoy_cc_library(
     ],
 )
 
-envoy_cc_library(
-    name = "source_impl_lib",
-    srcs = ["source_impl.cc"],
-    hdrs = ["source_impl.h"],
-    external_deps = [
-        "abseil_optional",
-    ],
-    deps = [
-        "//include/envoy/stats:stats_interface",
-    ],
-)
-
 envoy_cc_library(
     name = "stat_data_allocator_lib",
     hdrs = ["stat_data_allocator_impl.h"],
@@ -122,7 +110,6 @@ envoy_cc_library(
     deps = [
         ":histogram_lib",
         ":metric_impl_lib",
-        ":source_impl_lib",
         ":symbol_table_lib",
         ":tag_extractor_lib",
         ":utility_lib",

source/common/stats/thread_local_store.cc

@@ -25,7 +25,7 @@ ThreadLocalStoreImpl::ThreadLocalStoreImpl(StatDataAllocator& alloc)
     : alloc_(alloc), default_scope_(createScope("")),
       tag_producer_(std::make_unique<TagProducerImpl>()),
       stats_matcher_(std::make_unique<StatsMatcherImpl>()), heap_allocator_(alloc.symbolTable()),
-      source_(*this), null_counter_(alloc.symbolTable()), null_gauge_(alloc.symbolTable()),
+      null_counter_(alloc.symbolTable()), null_gauge_(alloc.symbolTable()),
       null_histogram_(alloc.symbolTable()) {}
 
 ThreadLocalStoreImpl::~ThreadLocalStoreImpl() {

source/common/stats/thread_local_store.h

@@ -11,7 +11,6 @@
 #include "common/common/hash.h"
 #include "common/stats/heap_stat_data.h"
 #include "common/stats/histogram_impl.h"
-#include "common/stats/source_impl.h"
 #include "common/stats/symbol_table_impl.h"
 #include "common/stats/utility.h"
 
@@ -211,11 +210,8 @@ class ThreadLocalStoreImpl : Logger::Loggable<Logger::Id::stats>, public StoreRo
   void initializeThreading(Event::Dispatcher& main_thread_dispatcher,
                            ThreadLocal::Instance& tls) override;
   void shutdownThreading() override;
-
   void mergeHistograms(PostMergeCb mergeCb) override;
 
-  Source& source() override { return source_; }
-
 private:
   template <class Stat> using StatMap = StatNameHashMap<Stat>;
 
@@ -362,7 +358,6 @@ class ThreadLocalStoreImpl : Logger::Loggable<Logger::Id::stats>, public StoreRo
   std::atomic<bool> shutting_down_{};
   std::atomic<bool> merge_in_progress_{};
   HeapStatDataAllocator heap_allocator_;
-  SourceImpl source_;
 
   NullCounterImpl null_counter_;
   NullGaugeImpl null_gauge_;

source/extensions/stat_sinks/common/statsd/statsd.cc

@@ -48,20 +48,19 @@ UdpStatsdSink::UdpStatsdSink(ThreadLocal::SlotAllocator& tls,
   });
 }
 
-void UdpStatsdSink::flush(Stats::Source& source) {
+void UdpStatsdSink::flush(Stats::MetricSnapshot& snapshot) {
   Writer& writer = tls_->getTyped<Writer>();
-  for (const Stats::CounterSharedPtr& counter : source.cachedCounters()) {
-    if (counter->used()) {
-      uint64_t delta = counter->latch();
-      writer.write(fmt::format("{}.{}:{}|c{}", prefix_, getName(*counter), delta,
-                               buildTagStr(counter->tags())));
+  for (const auto& counter : snapshot.counters()) {
+    if (counter.counter_.get().used()) {
+      writer.write(fmt::format("{}.{}:{}|c{}", prefix_, getName(counter.counter_.get()),
+                               counter.delta_, buildTagStr(counter.counter_.get().tags())));
     }
   }
 
-  for (const Stats::GaugeSharedPtr& gauge : source.cachedGauges()) {
-    if (gauge->used()) {
-      writer.write(fmt::format("{}.{}:{}|g{}", prefix_, getName(*gauge), gauge->value(),
-                               buildTagStr(gauge->tags())));
+  for (const auto& gauge : snapshot.gauges()) {
+    if (gauge.get().used()) {
+      writer.write(fmt::format("{}.{}:{}|g{}", prefix_, getName(gauge.get()), gauge.get().value(),
+                               buildTagStr(gauge.get().tags())));
     }
   }
 }
@@ -110,18 +109,18 @@ TcpStatsdSink::TcpStatsdSink(const LocalInfo::LocalInfo& local_info,
   });
 }
 
-void TcpStatsdSink::flush(Stats::Source& source) {
+void TcpStatsdSink::flush(Stats::MetricSnapshot& snapshot) {
   TlsSink& tls_sink = tls_->getTyped<TlsSink>();
   tls_sink.beginFlush(true);
-  for (const Stats::CounterSharedPtr& counter : source.cachedCounters()) {
-    if (counter->used()) {
-      tls_sink.flushCounter(counter->name(), counter->latch());
+  for (const auto& counter : snapshot.counters()) {
+    if (counter.counter_.get().used()) {
+      tls_sink.flushCounter(counter.counter_.get().name(), counter.delta_);
     }
   }
 
-  for (const Stats::GaugeSharedPtr& gauge : source.cachedGauges()) {
-    if (gauge->used()) {
-      tls_sink.flushGauge(gauge->name(), gauge->value());
+  for (const auto& gauge : snapshot.gauges()) {
+    if (gauge.get().used()) {
+      tls_sink.flushGauge(gauge.get().name(), gauge.get().value());
     }
   }
   tls_sink.endFlush(true);

source/extensions/stat_sinks/common/statsd/statsd.h

@@ -5,7 +5,6 @@
 #include "envoy/stats/histogram.h"
 #include "envoy/stats/scope.h"
 #include "envoy/stats/sink.h"
-#include "envoy/stats/source.h"
 #include "envoy/stats/stats.h"
 #include "envoy/stats/tag.h"
 #include "envoy/thread_local/thread_local.h"
@@ -58,7 +57,7 @@ class UdpStatsdSink : public Stats::Sink {
   }
 
   // Stats::Sink
-  void flush(Stats::Source& source) override;
+  void flush(Stats::MetricSnapshot& snapshot) override;
   void onHistogramComplete(const Stats::Histogram& histogram, uint64_t value) override;
 
   // Called in unit test to validate writer construction and address.
@@ -87,7 +86,7 @@ class TcpStatsdSink : public Stats::Sink {
                 Stats::Scope& scope, const std::string& prefix = getDefaultPrefix());
 
   // Stats::Sink
-  void flush(Stats::Source& source) override;
+  void flush(Stats::MetricSnapshot& snapshot) override;
   void onHistogramComplete(const Stats::Histogram& histogram, uint64_t value) override {
     // For statsd histograms are all timers.
     tls_->getTyped<TlsSink>().onTimespanComplete(histogram.name(),

source/extensions/stat_sinks/hystrix/hystrix.cc

@@ -320,7 +320,7 @@ Http::Code HystrixSink::handlerHystrixEventStream(absl::string_view,
   return Http::Code::OK;
 }
 
-void HystrixSink::flush(Stats::Source& source) {
+void HystrixSink::flush(Stats::MetricSnapshot& snapshot) {
   if (callbacks_list_.empty()) {
     return;
   }
@@ -330,9 +330,10 @@ void HystrixSink::flush(Stats::Source& source) {
 
   // Save a map of the relevant histograms per cluster in a convenient format.
   std::unordered_map<std::string, QuantileLatencyMap> time_histograms;
-  for (const Stats::ParentHistogramSharedPtr& histogram : source.cachedHistograms()) {
-    if (histogram->tagExtractedStatName() == cluster_upstream_rq_time_) {
-      absl::optional<Stats::StatName> value = Stats::Utility::findTag(*histogram, cluster_name_);
+  for (const auto& histogram : snapshot.histograms()) {
+    if (histogram.get().tagExtractedStatName() == cluster_upstream_rq_time_) {
+      absl::optional<Stats::StatName> value =
+          Stats::Utility::findTag(histogram.get(), cluster_name_);
       // Make sure we found the cluster name tag
       ASSERT(value);
       std::string value_str = server_.stats().symbolTable().toString(*value);
@@ -342,12 +343,12 @@ void HystrixSink::flush(Stats::Source& source) {
       QuantileLatencyMap& hist_map = it_bool_pair.first->second;
 
       const std::vector<double>& supported_quantiles =
-          histogram->intervalStatistics().supportedQuantiles();
+          histogram.get().intervalStatistics().supportedQuantiles();
       for (size_t i = 0; i < supported_quantiles.size(); ++i) {
         // binary-search here is likely not worth it, as hystrix_quantiles has <10 elements.
         if (std::find(hystrix_quantiles.begin(), hystrix_quantiles.end(), supported_quantiles[i]) !=
             hystrix_quantiles.end()) {
-          const double value = histogram->intervalStatistics().computedQuantiles()[i];
+          const double value = histogram.get().intervalStatistics().computedQuantiles()[i];
           if (!std::isnan(value)) {
             hist_map[supported_quantiles[i]] = value;
           }