From ccdd278d888a7a954ccd2ffe1d2664b44130963c Mon Sep 17 00:00:00 2001
From: Amnon Heiman <amnon@scylladb.com>
Date: Thu, 23 Jun 2022 14:27:11 +0300
Subject: [PATCH 1/4] metrics.hh: Support SUMMARY type

This patch add support for the summary type on the metrics layer.

A summary is a different kind of histogram, it's buckets are percentile
so the reporting layer (i.e. Prometheus for example) would know to
report it correctly.

Signed-off-by: Amnon Heiman <amnon@scylladb.com>
---
 include/seastar/core/metrics.hh | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/include/seastar/core/metrics.hh b/include/seastar/core/metrics.hh
index 450db26944f..8d8dca3a8f0 100644
--- a/include/seastar/core/metrics.hh
+++ b/include/seastar/core/metrics.hh
@@ -256,6 +256,7 @@ enum class data_type : uint8_t {
     REAL_COUNTER,
     GAUGE,
     HISTOGRAM,
+    SUMMARY,
 };
 
 template <bool callable, typename T>
@@ -595,6 +596,18 @@ impl::metric_definition_impl make_histogram(metric_name_type name,
     return  {name, {impl::data_type::HISTOGRAM, "histogram"}, make_function(std::forward<T>(val), impl::data_type::HISTOGRAM), d, {}};
 }
 
+/*!
+ * \brief create a summary metric.
+ *
+ * Summaries are a different kind of histograms. It reports in quantiles.
+ * For example, the p99 and p95 latencies.
+ */
+template<typename T>
+impl::metric_definition_impl make_summary(metric_name_type name,
+        description d, T&& val) {
+    return  {name, {impl::data_type::SUMMARY, "summary"}, make_function(std::forward<T>(val), impl::data_type::SUMMARY), d, {}};
+}
+
 
 /*!
  * \brief create a total_bytes metric.

From 893e2de1af512dee5f21ce6ba5cc082c962ea56e Mon Sep 17 00:00:00 2001
From: Amnon Heiman <amnon@scylladb.com>
Date: Fri, 27 May 2022 16:45:10 +0300
Subject: [PATCH 2/4] metrics.cc: missing count and sum when aggregating
 histograms

This patch adds a missing part to how histograms are being aggregated,
it needs to aggregate the sum and count as well.

Signed-off-by: Amnon Heiman <amnon@scylladb.com>
---
 src/core/metrics.cc | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/src/core/metrics.cc b/src/core/metrics.cc
index 98a14874a02..ea39b3e72d0 100644
--- a/src/core/metrics.cc
+++ b/src/core/metrics.cc
@@ -365,6 +365,8 @@ histogram& histogram::operator+=(const histogram& c) {
             buckets[i].count += c.buckets[i].count;
         }
     }
+    sample_count += c.sample_count;
+    sample_sum += c.sample_sum;
     return *this;
 }
 

From 9c84d52b49a3f637b4833b342abc07183765971b Mon Sep 17 00:00:00 2001
From: Amnon Heiman <amnon@scylladb.com>
Date: Thu, 23 Jun 2022 14:28:16 +0300
Subject: [PATCH 3/4] metrics: support aggregate by labels and skip_when_empty

Aggregate labels are a mechanism for reporting aggregated results. Most
commonly it allows to report one histogram per node instead of per
shard.

This patch adds an option to mark a metric with a vector of labels.
That vector will be part of the metric meta-data so the reporting
layer would be able to aggregate over it.

Skip when empty, means that metrics that are not in used, will not be
reported.

A common scenario is that user register a metrics, but that metrics is
never used.
The most common case is histogram and summary but it it can also happen
with counters.

This patch adds an option to mark a metric with skip_when_empty.
When done so, if a metric was never used (true for histogram, counters
and summary) it will not be reported.

Signed-off-by: Amnon Heiman <amnon@scylladb.com>
---
 include/seastar/core/metrics.hh     | 18 ++++++++++++++++--
 include/seastar/core/metrics_api.hh | 10 +++++++---
 src/core/metrics.cc                 | 26 +++++++++++++++++++++-----
 3 files changed, 44 insertions(+), 10 deletions(-)

diff --git a/include/seastar/core/metrics.hh b/include/seastar/core/metrics.hh
index 8d8dca3a8f0..570cc3651ec 100644
--- a/include/seastar/core/metrics.hh
+++ b/include/seastar/core/metrics.hh
@@ -341,7 +341,16 @@ public:
     const histogram& get_histogram() const {
         return std::get<histogram>(u);
     }
-
+    /*!
+     * \brief return true if this is metrics was never used
+     *
+     * Histograms, Summaries and counters are ever growing by nature, so
+     * it possible to check if they have been used or not.
+     */
+    bool is_empty() const {
+        return ((_type == data_type::HISTOGRAM || _type == data_type::SUMMARY) && get_histogram().sample_count == 0) ||
+                ((_type == data_type::COUNTER || _type == data_type::REAL_COUNTER) && d() == 0);
+    }
 private:
     static void ulong_conversion_error(double d);
 };
@@ -359,16 +368,21 @@ struct metric_definition_impl {
     metric_function f;
     description d;
     bool enabled = true;
+    bool _skip_when_empty = false;
+    std::vector<std::string> aggregate_labels;
     std::map<sstring, sstring> labels;
     metric_definition_impl& operator ()(bool enabled);
     metric_definition_impl& operator ()(const label_instance& label);
+    metric_definition_impl& aggregate(const std::vector<label>& labels);
+    metric_definition_impl& skip_when_empty(bool skip=true);
     metric_definition_impl& set_type(const sstring& type_name);
     metric_definition_impl(
         metric_name_type name,
         metric_type type,
         metric_function f,
         description d,
-        std::vector<label_instance> labels);
+        std::vector<label_instance> labels,
+        std::vector<label> aggregate_labels = {});
 };
 
 class metric_groups_def {
diff --git a/include/seastar/core/metrics_api.hh b/include/seastar/core/metrics_api.hh
index 6843df47ab4..c2e3e8e2fdd 100644
--- a/include/seastar/core/metrics_api.hh
+++ b/include/seastar/core/metrics_api.hh
@@ -152,6 +152,7 @@ struct metric_family_info {
     metric_type_def inherit_type;
     description d;
     sstring name;
+    std::vector<std::string> aggregate_labels;
 };
 
 
@@ -161,6 +162,7 @@ struct metric_family_info {
 struct metric_info {
     metric_id id;
     bool enabled;
+    bool skip_when_empty;
 };
 
 
@@ -185,7 +187,7 @@ class registered_metric {
     metric_function _f;
     shared_ptr<impl> _impl;
 public:
-    registered_metric(metric_id id, metric_function f, bool enabled=true);
+    registered_metric(metric_id id, metric_function f, bool enabled=true, bool skip_when_empty=false);
     virtual ~registered_metric() {}
     virtual metric_value operator()() const {
         return _f();
@@ -198,7 +200,9 @@ public:
     void set_enabled(bool b) {
         _info.enabled = b;
     }
-
+    void set_skip_when_empty(bool skip) {
+        _info.skip_when_empty = skip;
+    }
     const metric_id& get_id() const {
         return _info.id;
     }
@@ -330,7 +334,7 @@ public:
         return _value_map;
     }
 
-    void add_registration(const metric_id& id, const metric_type& type, metric_function f, const description& d, bool enabled);
+    void add_registration(const metric_id& id, const metric_type& type, metric_function f, const description& d, bool enabled, bool skip_when_empty, const std::vector<std::string>& aggregate_labels);
     void remove_registration(const metric_id& id);
     future<> stop() {
         return make_ready_future<>();
diff --git a/src/core/metrics.cc b/src/core/metrics.cc
index ea39b3e72d0..cf8ef1a5af6 100644
--- a/src/core/metrics.cc
+++ b/src/core/metrics.cc
@@ -118,9 +118,10 @@ bool label_instance::operator!=(const label_instance& id2) const {
 label shard_label("shard");
 namespace impl {
 
-registered_metric::registered_metric(metric_id id, metric_function f, bool enabled) :
+registered_metric::registered_metric(metric_id id, metric_function f, bool enabled, bool skip_when_empty) :
         _f(f), _impl(get_local_impl()) {
     _info.enabled = enabled;
+    set_skip_when_empty(skip_when_empty);
     _info.id = id;
 }
 
@@ -146,7 +147,8 @@ metric_definition_impl::metric_definition_impl(
         metric_type type,
         metric_function f,
         description d,
-        std::vector<label_instance> _labels)
+        std::vector<label_instance> _labels,
+        std::vector<label> _aggregate_labels)
         : name(name), type(type), f(f)
         , d(d), enabled(true) {
     for (auto i: _labels) {
@@ -155,6 +157,7 @@ metric_definition_impl::metric_definition_impl(
     if (labels.find(shard_label.name()) == labels.end()) {
         labels[shard_label.name()] = shard();
     }
+    aggregate(_aggregate_labels);
 }
 
 metric_definition_impl& metric_definition_impl::operator ()(bool _enabled) {
@@ -172,6 +175,18 @@ metric_definition_impl& metric_definition_impl::set_type(const sstring& type_nam
     return *this;
 }
 
+metric_definition_impl& metric_definition_impl::aggregate(const std::vector<label>& _labels) {
+    aggregate_labels.reserve(_labels.size());
+    std::transform(_labels.begin(), _labels.end(),std::back_inserter(aggregate_labels),
+            [](const label& l) { return l.name(); });
+    return *this;
+}
+
+metric_definition_impl& metric_definition_impl::skip_when_empty(bool skip) {
+    _skip_when_empty = skip;
+    return *this;
+}
+
 std::unique_ptr<metric_groups_def> create_metric_groups() {
     return  std::make_unique<metric_groups_impl>();
 }
@@ -186,7 +201,7 @@ metric_groups_impl& metric_groups_impl::add_metric(group_name_type name, const m
 
     metric_id id(name, md._impl->name, md._impl->labels);
 
-    get_local_impl()->add_registration(id, md._impl->type, md._impl->f, md._impl->d, md._impl->enabled);
+    get_local_impl()->add_registration(id, md._impl->type, md._impl->f, md._impl->d, md._impl->enabled, md._impl->_skip_when_empty, md._impl->aggregate_labels);
 
     _registration.push_back(id);
     return *this;
@@ -327,8 +342,8 @@ std::vector<std::vector<metric_function>>& impl::functions() {
     return _current_metrics;
 }
 
-void impl::add_registration(const metric_id& id, const metric_type& type, metric_function f, const description& d, bool enabled) {
-    auto rm = ::seastar::make_shared<registered_metric>(id, f, enabled);
+void impl::add_registration(const metric_id& id, const metric_type& type, metric_function f, const description& d, bool enabled, bool skip_when_empty, const std::vector<std::string>& aggregate_labels) {
+    auto rm = ::seastar::make_shared<registered_metric>(id, f, enabled, skip_when_empty);
     sstring name = id.full_name();
     if (_value_map.find(name) != _value_map.end()) {
         auto& metric = _value_map[name];
@@ -344,6 +359,7 @@ void impl::add_registration(const metric_id& id, const metric_type& type, metric
         _value_map[name].info().d = d;
         _value_map[name].info().inherit_type = type.type_name;
         _value_map[name].info().name = id.full_name();
+        _value_map[name].info().aggregate_labels = aggregate_labels;
         _value_map[name][id.labels()] = rm;
     }
     dirty();

From 3f3582035d3d62924e936f71e059dcd53564de7d Mon Sep 17 00:00:00 2001
From: Amnon Heiman <amnon@scylladb.com>
Date: Thu, 4 Nov 2021 08:35:08 +0200
Subject: [PATCH 4/4] prometheus.cc: Optimize reporting

This patch adds multiple functionality to Prometheus reporting:
1. Add summary reporting. Summaries are used in seastar to report aggregated
   percentile information (for example p95 and p99)
   The main usage is to report per-shard summary of a latency
   histograms.
2. Support aggregated metrics. With an aggregated metrics,
   Prometheus would aggregate multiple metrics based on labels and
   would report the result. Usually this would be for reporting a single
   latency histogram per node instead of per shard. But it could be used
   for counters and gauge as well.
3. Skip empty counters, histograms and summaries. It's a common practice to
   register lots of metrics even if they are not being used.
   Histograms have a huge effect on performance, so not reporting an empty
   histogram is a great performance boost both for the application and
   for the Prometheus server.
   This is true for Summaries and Counters as well, marking a metrics
   with skip_when_empty would mean Prometheus will not report those
   metrics.
4. As an optimization, the stringstream that is used per metric is
   reused and clear insted of recreated.

Signed-off-by: Amnon Heiman <amnon@scylladb.com>
---
 src/core/prometheus.cc | 173 ++++++++++++++++++++++++++++++++---------
 1 file changed, 136 insertions(+), 37 deletions(-)

diff --git a/src/core/prometheus.cc b/src/core/prometheus.cc
index 937eec75724..20352c84597 100644
--- a/src/core/prometheus.cc
+++ b/src/core/prometheus.cc
@@ -51,6 +51,8 @@ static std::string to_str(seastar::metrics::impl::data_type dt) {
         return "counter";
     case seastar::metrics::impl::data_type::HISTOGRAM:
         return "histogram";
+    case seastar::metrics::impl::data_type::SUMMARY:
+        return "summary";
     }
     return "untyped";
 }
@@ -63,6 +65,7 @@ static std::string to_str(const seastar::metrics::impl::metric_value& v) {
     case seastar::metrics::impl::data_type::COUNTER:
         return std::to_string(v.i());
     case seastar::metrics::impl::data_type::HISTOGRAM:
+    case seastar::metrics::impl::data_type::SUMMARY:
         break;
     }
     return ""; // we should never get here but it makes the compiler happy
@@ -478,14 +481,123 @@ metric_family_range get_range(const metrics_families_per_shard& mf, const sstrin
 
 }
 
+void write_histogram(std::stringstream& s, const config& ctx, const sstring& name, const seastar::metrics::histogram& h, std::map<sstring, sstring> labels) {
+    add_name(s, name + "_sum", labels, ctx);
+    s << h.sample_sum;
+    s << "\n";
+    add_name(s, name + "_count", labels, ctx);
+    s << h.sample_count;
+    s << "\n";
+
+    auto& le = labels["le"];
+    auto bucket = name + "_bucket";
+    for (auto  i : h.buckets) {
+         le = std::to_string(i.upper_bound);
+        add_name(s, bucket, labels, ctx);
+        s << i.count;
+        s << "\n";
+    }
+    labels["le"] = "+Inf";
+    add_name(s, bucket, labels, ctx);
+    s << h.sample_count;
+    s << "\n";
+}
+
+void write_summary(std::stringstream& s, const config& ctx, const sstring& name, const seastar::metrics::histogram& h, std::map<sstring, sstring> labels) {
+    if (h.sample_sum) {
+        add_name(s, name + "_sum", labels, ctx);
+        s << h.sample_sum;
+        s << "\n";
+    }
+    if (h.sample_count) {
+        add_name(s, name + "_count", labels, ctx);
+        s << h.sample_count;
+        s << "\n";
+    }
+    auto& le = labels["quantile"];
+    for (auto  i : h.buckets) {
+        le = std::to_string(i.upper_bound);
+        add_name(s, name, labels, ctx);
+        s << i.count;
+        s << "\n";
+    }
+}
+/*!
+ * \brief a helper class to aggregate metrics over labels
+ *
+ * This class sum multiple metrics based on a list of labels.
+ * It returns one or more metrics each aggregated by the aggregate_by labels.
+ *
+ * To use it, you define what labels it should aggregate by and then pass to
+ * it metrics with their labels.
+ * For example if a metrics has a 'shard' and 'name' labels and you aggregate by 'shard'
+ * it would return a map of metrics each with only the 'name' label
+ *
+ */
+class metric_aggregate_by_labels {
+    std::vector<std::string> _labels_to_aggregate_by;
+    std::unordered_map<std::map<sstring, sstring>, seastar::metrics::impl::metric_value> _values;
+public:
+    metric_aggregate_by_labels(std::vector<std::string>  labels) : _labels_to_aggregate_by(std::move(labels)) {
+    }
+    /*!
+     * \brief add a metric
+     *
+     * This method gets a metric and its labels and add it to the aggregated metric.
+     * For example, if a metric has the labels {'shard':'0', 'name':'myhist'} and we are aggregating
+     * over 'shard'
+     * The metric would be added to the aggregated metric with labels {'name':'myhist'}.
+     *
+     */
+    void add(const seastar::metrics::impl::metric_value& m, std::map<sstring, sstring> labels) {
+        for (auto&& l : _labels_to_aggregate_by) {
+            labels.erase(l);
+        }
+        if (_values.find(labels) == _values.end()) {
+            _values.emplace(labels, m);
+        } else {
+            _values[labels] += m;
+        }
+    }
+    const std::unordered_map<std::map<sstring, sstring>, seastar::metrics::impl::metric_value>& get_values() const {
+        return _values;
+    }
+    bool empty() const {
+        return _values.empty();
+    }
+};
+
+std::string get_value_as_string(std::stringstream& s, const mi::metric_value& value) {
+    std::string value_str;
+    try {
+        value_str = to_str(value);
+    } catch (const std::range_error& e) {
+        seastar_logger.debug("prometheus: write_text_representation: {}: {}", s.str(), e.what());
+        value_str = "NaN";
+    } catch (...) {
+        auto ex = std::current_exception();
+        // print this error as it's ignored later on by `connection::start_response`
+        seastar_logger.error("prometheus: write_text_representation: {}: {}", s.str(), ex);
+        std::rethrow_exception(std::move(ex));
+    }
+    return value_str;
+}
+
 future<> write_text_representation(output_stream<char>& out, const config& ctx, const metric_family_range& m) {
     return seastar::async([&ctx, &out, &m] () mutable {
         bool found = false;
+        std::stringstream s;
         for (metric_family& metric_family : m) {
             auto name = ctx.prefix + "_" + metric_family.name();
             found = false;
-            metric_family.foreach_metric([&out, &ctx, &found, &name, &metric_family](auto value, auto value_info) mutable {
-                std::stringstream s;
+            metric_aggregate_by_labels aggregated_values(metric_family.metadata().aggregate_labels);
+            bool should_aggregate = !metric_family.metadata().aggregate_labels.empty();
+            metric_family.foreach_metric([&s, &out, &ctx, &found, &name, &metric_family, &aggregated_values, should_aggregate](auto value, auto value_info) mutable {
+                s.clear();
+                s.str("");
+                if (value_info.skip_when_empty && value.is_empty()) {
+                    return;
+                }
                 if (!found) {
                     if (metric_family.metadata().d.str() != "") {
                         s << "# HELP " << name << " " <<  metric_family.metadata().d.str() << "\n";
@@ -493,48 +605,35 @@ future<> write_text_representation(output_stream<char>& out, const config& ctx,
                     s << "# TYPE " << name << " " << to_str(metric_family.metadata().type) << "\n";
                     found = true;
                 }
-                if (value.type() == mi::data_type::HISTOGRAM) {
-                    auto&& h = value.get_histogram();
-                    std::map<sstring, sstring> labels = value_info.id.labels();
-                    add_name(s, name + "_sum", labels, ctx);
-                    s << h.sample_sum;
-                    s << "\n";
-                    add_name(s, name + "_count", labels, ctx);
-                    s << h.sample_count;
-                    s << "\n";
-
-                    auto& le = labels["le"];
-                    auto bucket = name + "_bucket";
-                    for (auto  i : h.buckets) {
-                         le = std::to_string(i.upper_bound);
-                        add_name(s, bucket, labels, ctx);
-                        s << i.count;
-                        s << "\n";
-                    }
-                    labels["le"] = "+Inf";
-                    add_name(s, bucket, labels, ctx);
-                    s << h.sample_count;
-                    s << "\n";
+                if (should_aggregate) {
+                    aggregated_values.add(value, value_info.id.labels());
+                } else if (value.type() == mi::data_type::SUMMARY) {
+                    write_summary(s, ctx, name, value.get_histogram(), value_info.id.labels());
+                } else if (value.type() == mi::data_type::HISTOGRAM) {
+                    write_histogram(s, ctx, name, value.get_histogram(), value_info.id.labels());
                 } else {
                     add_name(s, name, value_info.id.labels(), ctx);
-                    std::string value_str;
-                    try {
-                        value_str = to_str(value);
-                    } catch (const std::range_error& e) {
-                        seastar_logger.debug("prometheus: write_text_representation: {}: {}", s.str(), e.what());
-                        value_str = "NaN";
-                    } catch (...) {
-                        auto ex = std::current_exception();
-                        // print this error as it's ignored later on by `connection::start_response`
-                        seastar_logger.error("prometheus: write_text_representation: {}: {}", s.str(), ex);
-                        std::rethrow_exception(std::move(ex));
-                    }
-                    s << value_str;
+                    s << get_value_as_string(s, value);
                     s << "\n";
                 }
                 out.write(s.str()).get();
                 thread::maybe_yield();
             });
+            if (!aggregated_values.empty()) {
+                for (auto&& h : aggregated_values.get_values()) {
+                    s.clear();
+                    s.str("");
+                    if (h.second.type() == mi::data_type::HISTOGRAM) {
+                        write_histogram(s, ctx, name, h.second.get_histogram(), h.first);
+                    } else {
+                        add_name(s, name, h.first, ctx);
+                        s << get_value_as_string(s, h.second);
+                        s << "\n";
+                    }
+                    out.write(s.str()).get();
+                    thread::maybe_yield();
+                }
+            }
         }
     });
 }