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

@@ -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 {

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<>();

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();