controller/probe: Reduce lifecycle

The probe requires access to members, topics, and partition_leaders.

* Start the probe after all dependencies have been started
* Stop the probe before its dependencies are stopped

Fix redpanda-data#5466

Signed-off-by: Ben Pope <ben@redpanda.com>

src/v/cluster/controller.cc

@@ -92,7 +92,8 @@ ss::future<> controller::wire_up() {
           return _authorizer.start(
             []() { return config::shard_local_cfg().superusers.bind(); });
       })
-      .then([this] { return _tp_state.start(); });
+      .then([this] { return _tp_state.start(); })
+      .then([this] { _probe.start(); });
 }
 
 ss::future<> controller::start() {
@@ -377,6 +378,7 @@ ss::future<> controller::stop() {
         f = shutdown_input();
     }
 
+    _probe.stop();
     return f.then([this] {
         auto stop_leader_balancer = _leader_balancer ? _leader_balancer->stop()
                                                      : ss::now();

src/v/cluster/controller_probe.cc

@@ -24,24 +24,33 @@
 namespace cluster {
 
 controller_probe::controller_probe(controller& c) noexcept
-  : _controller(c) {
-    _controller._raft_manager.local().register_leadership_notification(
-      [this](
-        raft::group_id group,
-        model::term_id /*term*/,
-        std::optional<model::node_id> leader_id) {
-          // We are only interested in notifications regarding the controller
-          // group.
-          if (_controller._raft0->group() != group) {
-              return;
-          }
-
-          if (leader_id != _controller.self()) {
-              _public_metrics.reset();
-          } else {
-              setup_metrics();
-          }
-      });
+  : _controller(c)
+  , _leadership_notification_handle{} {}
+
+void controller_probe::start() {
+    _leadership_notification_handle
+      = _controller._raft_manager.local().register_leadership_notification(
+        [this](
+          raft::group_id group,
+          model::term_id /*term*/,
+          std::optional<model::node_id> leader_id) {
+            // We are only interested in notifications regarding the controller
+            // group.
+            if (_controller._raft0->group() != group) {
+                return;
+            }
+
+            if (leader_id != _controller.self()) {
+                _public_metrics.reset();
+            } else {
+                setup_metrics();
+            }
+        });
+}
+
+void controller_probe::stop() {
+    _controller._raft_manager.local().unregister_leadership_notification(
+      _leadership_notification_handle);
 }
 
 void controller_probe::setup_metrics() {

src/v/cluster/controller_probe.h

@@ -12,6 +12,7 @@
 #pragma once
 
 #include "cluster/fwd.h"
+#include "cluster/types.h"
 #include "seastarx.h"
 
 #include <seastar/core/metrics_registration.hh>
@@ -22,11 +23,15 @@ class controller_probe {
 public:
     explicit controller_probe(cluster::controller&) noexcept;
 
+    void start();
+    void stop();
+
     void setup_metrics();
 
 private:
     cluster::controller& _controller;
     std::unique_ptr<ss::metrics::metric_groups> _public_metrics;
+    cluster::notification_id_type _leadership_notification_handle;
 };
 
 } // namespace cluster