Merge 'Minor metrics memory optimizations' from Stephan Dollberg

The metrics subsystem is quite a heavy memory user at high metric counts.

This PR features two smaller improvements reducing memory usage by removing some redundant storage. I have a higher impact change following that internalizes labels.

Two commits (details in commit):
 - Don't store full metric_id in the metric_groups_impl
 - Store impl SP in the metric_groups_impl and not each registered_metric

Closes #2554

* github.com:scylladb/seastar:
  metrics: Don't store a metrics impl sp per registered metric
  metrics: Store register_ref in metric_groups_impl registration list

include/seastar/core/metrics_api.hh

@@ -197,26 +197,11 @@ struct metric_info {
 };
 
 
-using metrics_registration = std::vector<metric_id>;
-
-class metric_groups_impl : public metric_groups_def {
-    metrics_registration _registration;
-public:
-    metric_groups_impl() = default;
-    ~metric_groups_impl();
-    metric_groups_impl(const metric_groups_impl&) = delete;
-    metric_groups_impl(metric_groups_impl&&) = default;
-    metric_groups_impl& add_metric(group_name_type name, const metric_definition& md);
-    metric_groups_impl& add_group(group_name_type name, const std::initializer_list<metric_definition>& l);
-    metric_groups_impl& add_group(group_name_type name, const std::vector<metric_definition>& l);
-};
-
 class impl;
 
 class registered_metric final {
     metric_info _info;
     metric_function _f;
-    shared_ptr<impl> _impl;
 public:
     registered_metric(metric_id id, metric_function f, bool enabled=true, skip_when_empty skip=skip_when_empty::no);
     metric_value operator()() const {
@@ -250,6 +235,20 @@ public:
 
 using register_ref = shared_ptr<registered_metric>;
 using metric_instances = std::map<labels_type, register_ref>;
+using metrics_registration = std::vector<register_ref>;
+
+class metric_groups_impl : public metric_groups_def {
+    metrics_registration _registration;
+    shared_ptr<impl> _impl; // keep impl alive while metrics are registered
+public:
+    metric_groups_impl();
+    ~metric_groups_impl();
+    metric_groups_impl(const metric_groups_impl&) = delete;
+    metric_groups_impl(metric_groups_impl&&) = default;
+    metric_groups_impl& add_metric(group_name_type name, const metric_definition& md);
+    metric_groups_impl& add_group(group_name_type name, const std::initializer_list<metric_definition>& l);
+    metric_groups_impl& add_group(group_name_type name, const std::vector<metric_definition>& l);
+};
 
 class metric_family {
     metric_instances _instances;
@@ -375,7 +374,7 @@ public:
         return _value_map;
     }
 
-    void add_registration(const metric_id& id, const metric_type& type, metric_function f, const description& d, bool enabled, skip_when_empty skip, const std::vector<std::string>& aggregate_labels);
+    register_ref add_registration(const metric_id& id, const metric_type& type, metric_function f, const description& d, bool enabled, skip_when_empty skip, const std::vector<std::string>& aggregate_labels);
     void remove_registration(const metric_id& id);
     future<> stop() {
         return make_ready_future<>();

src/core/metrics.cc

@@ -215,7 +215,7 @@ label shard_label("shard");
 namespace impl {
 
 registered_metric::registered_metric(metric_id id, metric_function f, bool enabled, skip_when_empty skip) :
-        _f(f), _impl(get_local_impl()) {
+        _f(f) {
     _info.enabled = enabled;
     _info.should_skip_when_empty = skip;
     _info.id = id;
@@ -294,19 +294,28 @@ std::unique_ptr<metric_groups_def> create_metric_groups() {
     return  std::make_unique<metric_groups_impl>();
 }
 
+metric_groups_impl::metric_groups_impl() {}
+
 metric_groups_impl::~metric_groups_impl() {
     for (const auto& i : _registration) {
-        unregister_metric(i);
+        unregister_metric(i->info().id);
     }
 }
 
 metric_groups_impl& metric_groups_impl::add_metric(group_name_type name, const metric_definition& md)  {
+    // We could just do this in the constructor but some metric groups (like the
+    // smp queue ones) are actually constructed on a different shard originally
+    // than where the actual metrics are added.
+    // Hence, the shared_ptr owning shard check would fail so we do it only here.
+    if (_impl == nullptr) {
+        _impl = get_local_impl();
+    }
 
     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, md._impl->_skip_when_empty, md._impl->aggregate_labels);
+    auto reg = 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);
+    _registration.push_back(std::move(reg));
     return *this;
 }
 
@@ -439,7 +448,7 @@ std::vector<std::deque<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, skip_when_empty skip, const std::vector<std::string>& aggregate_labels) {
+register_ref impl::add_registration(const metric_id& id, const metric_type& type, metric_function f, const description& d, bool enabled, skip_when_empty skip, const std::vector<std::string>& aggregate_labels) {
     auto rm = ::seastar::make_shared<registered_metric>(id, f, enabled, skip);
     for (auto&& rl : _relabel_configs) {
         apply_relabeling(rl, rm->info());
@@ -468,6 +477,8 @@ void impl::add_registration(const metric_id& id, const metric_type& type, metric
         _value_map[name][rm->info().id.labels()] = rm;
     }
     dirty();
+
+    return rm;
 }
 
 future<metric_relabeling_result> impl::set_relabel_configs(const std::vector<relabel_config>& relabel_configs) {