feat statistics: add HistogramAggregator

Anton3 · Anton3 · commit 4363c8d1dc85 · 2023-12-14T15:37:18.000+03:00
It can:
- add together multiple histograms faster than Histogram would do it
- drop some buckets, e.g. to write approximate by-handler metrics and precise aggregate metrics
diff --git a/.mapping.json b/.mapping.json
@@ -562,6 +562,7 @@
   "core/include/userver/utils/statistics/fwd.hpp":"taxi/uservices/userver/core/include/userver/utils/statistics/fwd.hpp",
   "core/include/userver/utils/statistics/graphite.hpp":"taxi/uservices/userver/core/include/userver/utils/statistics/graphite.hpp",
   "core/include/userver/utils/statistics/histogram.hpp":"taxi/uservices/userver/core/include/userver/utils/statistics/histogram.hpp",
+  "core/include/userver/utils/statistics/histogram_aggregator.hpp":"taxi/uservices/userver/core/include/userver/utils/statistics/histogram_aggregator.hpp",
   "core/include/userver/utils/statistics/histogram_view.hpp":"taxi/uservices/userver/core/include/userver/utils/statistics/histogram_view.hpp",
   "core/include/userver/utils/statistics/impl/histogram_bucket.hpp":"taxi/uservices/userver/core/include/userver/utils/statistics/impl/histogram_bucket.hpp",
   "core/include/userver/utils/statistics/json.hpp":"taxi/uservices/userver/core/include/userver/utils/statistics/json.hpp",
@@ -1298,6 +1299,7 @@
   "core/src/utils/statistics/entry_impl.hpp":"taxi/uservices/userver/core/src/utils/statistics/entry_impl.hpp",
   "core/src/utils/statistics/graphite.cpp":"taxi/uservices/userver/core/src/utils/statistics/graphite.cpp",
   "core/src/utils/statistics/histogram.cpp":"taxi/uservices/userver/core/src/utils/statistics/histogram.cpp",
+  "core/src/utils/statistics/histogram_aggregator.cpp":"taxi/uservices/userver/core/src/utils/statistics/histogram_aggregator.cpp",
   "core/src/utils/statistics/histogram_benchmark.cpp":"taxi/uservices/userver/core/src/utils/statistics/histogram_benchmark.cpp",
   "core/src/utils/statistics/histogram_test.cpp":"taxi/uservices/userver/core/src/utils/statistics/histogram_test.cpp",
   "core/src/utils/statistics/histogram_view.cpp":"taxi/uservices/userver/core/src/utils/statistics/histogram_view.cpp",
diff --git a/clickhouse/src/storages/clickhouse/cluster.cpp b/clickhouse/src/storages/clickhouse/cluster.cpp
@@ -5,7 +5,6 @@
 #include <userver/formats/common/merge.hpp>
 #include <userver/formats/json/value_builder.hpp>
 #include <userver/utils/async.hpp>
-#include <userver/utils/statistics/metadata.hpp>
 
 #include <storages/clickhouse/impl/settings.hpp>
 
diff --git a/core/include/userver/utils/statistics/histogram.hpp b/core/include/userver/utils/statistics/histogram.hpp
@@ -55,6 +55,9 @@ struct BoundsBlock;
 /// This can be useful for writing custom metric serialization formats or
 /// for testing.
 ///
+/// Histogram metrics can be summed using
+/// utils::statistics::HistogramAggregator.
+///
 /// Histogram can be used in utils::statistics::MetricTag:
 /// @snippet utils/statistics/histogram_test.cpp  metric tag
 class Histogram final {
@@ -75,10 +78,6 @@ class Histogram final {
   /// Atomically increment the bucket corresponding to the given value.
   void Account(double value, std::uint64_t count = 1) noexcept;
 
-  /// Add the other histogram to the current one. Bucket borders in `this` and
-  /// `other` must be identical. Writes to `*this` are non-atomic.
-  void Add(HistogramView other);
-
   /// Atomically reset all counters to zero.
   friend void ResetMetric(Histogram& histogram) noexcept;
 
diff --git a/core/include/userver/utils/statistics/histogram_aggregator.hpp b/core/include/userver/utils/statistics/histogram_aggregator.hpp
@@ -0,0 +1,53 @@
+#pragma once
+
+#include <memory>
+
+#include <userver/utils/span.hpp>
+#include <userver/utils/statistics/fwd.hpp>
+#include <userver/utils/statistics/histogram_view.hpp>
+
+/// @file userver/utils/statistics/histogram_aggregator.hpp
+/// @brief @copybrief utils::statistics::HistogramAggregator
+
+USERVER_NAMESPACE_BEGIN
+
+namespace utils::statistics {
+
+/// @brief Used to aggregate multiple utils::statistics::Histogram metrics.
+///
+/// Usage example:
+/// @snippet utils/statistics/histogram_test.cpp  HistogramAggregator
+class HistogramAggregator final {
+ public:
+  explicit HistogramAggregator(utils::span<const double> upper_bounds);
+
+  HistogramAggregator(HistogramAggregator&&) noexcept;
+  HistogramAggregator& operator=(HistogramAggregator&&) noexcept;
+  ~HistogramAggregator();
+
+  /// @brief Add the other histogram to the current one.
+  ///
+  /// Bucket borders in `this` and `other` must be either identical, or bucket
+  /// borders in `this` must be a strict subset of bucket borders in `other`.
+  ///
+  /// Writes to `*this` are non-atomic.
+  void Add(HistogramView other);
+
+  /// Allows reading the histogram.
+  HistogramView GetView() const& noexcept;
+
+  /// @cond
+  // Store Histogram in a variable before taking a view on it.
+  HistogramView GetView() && noexcept = delete;
+  /// @endcond
+
+ private:
+  std::unique_ptr<impl::histogram::Bucket[]> buckets_;
+};
+
+/// Metric serialization support for HistogramAggregator.
+void DumpMetric(Writer& writer, const HistogramAggregator& histogram);
+
+}  // namespace utils::statistics
+
+USERVER_NAMESPACE_END
diff --git a/core/include/userver/utils/statistics/recentperiod.hpp b/core/include/userver/utils/statistics/recentperiod.hpp
@@ -176,16 +176,14 @@ class RecentPeriod {
   mutable std::vector<EpochBucket> items_;
 };
 
-/// @brief @a Writer support for @a RecentPeriod. Forwards to `DumpMetric`
-/// overload for `Result`.
-///
-/// @param args if any, are forwarded to `DumpMetric` for `Result`
+/// @a Writer support for @a RecentPeriod
 template <typename Counter, typename Result, typename Timer>
 void DumpMetric(Writer& writer,
                 const RecentPeriod<Counter, Result, Timer>& recent_period) {
   writer = recent_period.GetStatsForPeriod();
 }
 
+/// Reset support for @a RecentPeriod
 template <typename Counter, typename Result, typename Timer>
 void ResetMetric(RecentPeriod<Counter, Result, Timer>& recent_period) {
   recent_period.Reset();
diff --git a/core/src/cache/cache_update_trait_impl.cpp b/core/src/cache/cache_update_trait_impl.cpp
@@ -14,7 +14,6 @@
 #include <userver/utils/async.hpp>
 #include <userver/utils/atomic.hpp>
 #include <userver/utils/datetime.hpp>
-#include <userver/utils/statistics/metadata.hpp>
 
 #include <cache/cache_dependencies.hpp>
 #include <dump/dump_locator.hpp>
diff --git a/core/src/cache/lru_cache_component_base.cpp b/core/src/cache/lru_cache_component_base.cpp
@@ -7,7 +7,6 @@
 #include <userver/dump/dumper.hpp>
 #include <userver/dynamic_config/storage/component.hpp>
 #include <userver/testsuite/testsuite_support.hpp>
-#include <userver/utils/statistics/metadata.hpp>
 #include <userver/yaml_config/merge_schemas.hpp>
 #include <userver/yaml_config/schema.hpp>
 
diff --git a/core/src/clients/dns/component.cpp b/core/src/clients/dns/component.cpp
@@ -4,9 +4,6 @@
 #include <userver/components/component.hpp>
 #include <userver/components/statistics_storage.hpp>
 #include <userver/formats/json/inline.hpp>
-#include <userver/formats/json/value_builder.hpp>
-#include <userver/formats/parse/common_containers.hpp>
-#include <userver/utils/statistics/metadata.hpp>
 #include <userver/yaml_config/merge_schemas.hpp>
 
 USERVER_NAMESPACE_BEGIN
diff --git a/core/src/clients/http/component.cpp b/core/src/clients/http/component.cpp
@@ -8,7 +8,6 @@
 #include <userver/components/statistics_storage.hpp>
 #include <userver/dynamic_config/storage/component.hpp>
 #include <userver/testsuite/testsuite_support.hpp>
-#include <userver/utils/statistics/metadata.hpp>
 
 #include <clients/http/destination_statistics.hpp>
 #include <clients/http/statistics.hpp>
diff --git a/core/src/components/manager_controller_component.cpp b/core/src/components/manager_controller_component.cpp
@@ -8,7 +8,6 @@
 #include <userver/dynamic_config/storage/component.hpp>
 #include <userver/dynamic_config/value.hpp>
 #include <userver/logging/component.hpp>
-#include <userver/utils/statistics/metadata.hpp>
 
 #include <components/manager.hpp>
 
diff --git a/core/src/utils/statistics/histogram.cpp b/core/src/utils/statistics/histogram.cpp
@@ -156,11 +156,6 @@ void Histogram::Account(double value, std::uint64_t count) noexcept {
   bucket.counter.fetch_add(count, std::memory_order_relaxed);
 }
 
-// NOLINTNEXTLINE(readability-make-member-function-const)
-void Histogram::Add(HistogramView other) {
-  impl::histogram::Add(buckets_.get(), other);
-}
-
 void ResetMetric(Histogram& histogram) noexcept {
   impl::histogram::ResetMetric(histogram.buckets_.get());
 }
diff --git a/core/src/utils/statistics/histogram_aggregator.cpp b/core/src/utils/statistics/histogram_aggregator.cpp
@@ -0,0 +1,37 @@
+#include <userver/utils/statistics/histogram_aggregator.hpp>
+
+#include <userver/utils/assert.hpp>
+#include <userver/utils/statistics/impl/histogram_bucket.hpp>
+#include <userver/utils/statistics/writer.hpp>
+
+USERVER_NAMESPACE_BEGIN
+
+namespace utils::statistics {
+
+HistogramAggregator::HistogramAggregator(utils::span<const double> upper_bounds)
+    : buckets_(std::make_unique<impl::histogram::Bucket[]>(upper_bounds.size() +
+                                                           1)) {
+  impl::histogram::CopyBounds(buckets_.get(), upper_bounds);
+}
+
+HistogramAggregator::HistogramAggregator(HistogramAggregator&&) noexcept =
+    default;
+
+HistogramAggregator& HistogramAggregator::operator=(
+    HistogramAggregator&&) noexcept = default;
+
+HistogramAggregator::~HistogramAggregator() = default;
+
+void HistogramAggregator::Add(HistogramView other) {
+  UASSERT(buckets_);
+  impl::histogram::Add(buckets_.get(), other);
+}
+
+HistogramView HistogramAggregator::GetView() const& noexcept {
+  UASSERT(buckets_);
+  return impl::histogram::MakeView(buckets_.get());
+}
+
+}  // namespace utils::statistics
+
+USERVER_NAMESPACE_END
diff --git a/core/src/utils/statistics/histogram_test.cpp b/core/src/utils/statistics/histogram_test.cpp
@@ -9,6 +9,7 @@
 #include <userver/utils/enumerate.hpp>
 #include <userver/utils/statistics/fmt.hpp>
 #include <userver/utils/statistics/graphite.hpp>
+#include <userver/utils/statistics/histogram_aggregator.hpp>
 #include <userver/utils/statistics/json.hpp>
 #include <userver/utils/statistics/metric_tag.hpp>
 #include <userver/utils/statistics/metrics_storage.hpp>
@@ -111,17 +112,21 @@ UTEST(StatisticsHistogram, Copy) {
 }
 
 UTEST(StatisticsHistogram, Add) {
+  /// [HistogramAggregator]
   utils::statistics::Histogram histogram1{Bounds()};
   AccountSome(histogram1);
   AccountSome(histogram1);
 
   utils::statistics::Histogram histogram2{Bounds()};
   AccountSome(histogram2);
 
-  histogram1.Add(histogram2.GetView());
+  utils::statistics::HistogramAggregator aggregator{Bounds()};
+  aggregator.Add(histogram1.GetView());
+  aggregator.Add(histogram2.GetView());
 
-  EXPECT_EQ(fmt::to_string(histogram1.GetView()),
+  EXPECT_EQ(fmt::to_string(aggregator.GetView()),
             "[1.5]=3,[5]=3,[42]=15,[60]=0,[inf]=3");
+  /// [HistogramAggregator]
 }
 
 UTEST(StatisticsHistogram, Reset) {
@@ -143,10 +148,41 @@ UTEST(StatisticsHistogram, ZeroBuckets) {
   utils::statistics::Histogram histogram2{histogram};
   histogram2.Account(42);
 
-  histogram.Add(histogram2.GetView());
-  EXPECT_EQ(histogram.GetView().GetValueAtInf(), 17);
+  utils::statistics::HistogramAggregator aggregator{std::vector<double>{}};
+  aggregator.Add(histogram.GetView());
+  aggregator.Add(histogram2.GetView());
+  EXPECT_EQ(aggregator.GetView().GetValueAtInf(), 17);
 
-  EXPECT_EQ(fmt::to_string(histogram.GetView()), "[inf]=17");
+  EXPECT_EQ(fmt::to_string(aggregator.GetView()), "[inf]=17");
+}
+
+utils::statistics::Histogram MakeHistogramForNarrowing() {
+  utils::statistics::Histogram histogram{
+      std::vector<double>{2.0, 4.0, 6.0, 8.0}};
+  histogram.Account(1.0, 1);
+  histogram.Account(3.0, 10);
+  histogram.Account(5.0, 100);
+  histogram.Account(7.0, 1000);
+  histogram.Account(9.0, 10000);
+  return histogram;
+}
+
+UTEST(StatisticsHistogram, BucketNarrowing) {
+  const auto histogram = MakeHistogramForNarrowing();
+  {
+    utils::statistics::HistogramAggregator aggregator{
+        std::vector<double>{2.0, 8.0}};
+    aggregator.Add(histogram.GetView());
+    EXPECT_EQ(fmt::to_string(aggregator.GetView()),
+              "[2]=1,[8]=1110,[inf]=10000");
+  }
+  {
+    utils::statistics::HistogramAggregator aggregator{
+        std::vector<double>{4.0, 6.0}};
+    aggregator.Add(histogram.GetView());
+    EXPECT_EQ(fmt::to_string(aggregator.GetView()),
+              "[4]=11,[6]=100,[inf]=11000");
+  }
 }
 
 UTEST_DEATH(StatisticsHistogramDeathTest, InvalidBuckets) {
diff --git a/core/src/utils/statistics/impl/histogram_view_utils.hpp b/core/src/utils/statistics/impl/histogram_view_utils.hpp
@@ -6,6 +6,7 @@
 #include <boost/range/adaptor/transformed.hpp>
 #include <boost/range/algorithm/copy.hpp>
 #include <boost/range/algorithm/equal.hpp>
+#include <boost/range/algorithm/set_algorithm.hpp>
 #include <boost/range/algorithm/upper_bound.hpp>
 #include <boost/range/combine.hpp>
 
@@ -125,16 +126,21 @@ class MutableView final {
 
   // Non-atomic
   void Add(HistogramView other) const {
-    UINVARIANT(HistogramView{*this}.GetBucketCount() == other.GetBucketCount(),
-               "Mismatch in added Histogram sizes");
-    UASSERT_MSG(HasSameBounds(*this, other),
-                "Mismatch in added Histogram bounds");
+    UINVARIANT(
+        boost::range::includes(Access::Bounds(other), Access::Bounds(*this)),
+        "Buckets can be merged, but not added during Histogram conversion.");
     AddNonAtomic(buckets_[0].counter, other.GetValueAtInf());
-    for (const auto tuple_of_refs :
-         boost::combine(Access::Buckets(*this), Access::Values(other))) {
-      auto& self_bucket = boost::get<0>(tuple_of_refs);
-      const auto other_value = boost::get<1>(tuple_of_refs);
-      AddNonAtomic(self_bucket.counter, other_value);
+    const auto self_bounds = Access::Bounds(*this);
+    auto current_self_bound = self_bounds.begin();
+    for (const auto& other_bucket : Access::Buckets(other)) {
+      while (current_self_bound != self_bounds.end() &&
+             other_bucket.upper_bound.bound > *current_self_bound) {
+        ++current_self_bound;
+      }
+      auto& self_bucket = current_self_bound == self_bounds.end()
+                              ? buckets_[0]
+                              : *current_self_bound.base();
+      AddNonAtomic(self_bucket.counter, other_bucket.counter);
     }
   }
 
