From 862e549741b0c196b9a5e0e577cb6cf2137ee677 Mon Sep 17 00:00:00 2001
From: Michael Boquard <michael@redpanda.com>
Date: Tue, 12 Nov 2024 16:26:41 -0500
Subject: [PATCH] auditing: construct metrics probe during service construction

Move the construction of the audit probe from service start to service
construction. Previously, the audit probe was created as a unique pointer
when the service was started and destroyed when the service stopped.

Since the audit service has the same lifespan as the application, creating
the probe during service construction reduces the risk of dereferencing a
null pointer if the probe is accessed before the service has started.

Signed-off-by: Michael Boquard <michael@redpanda.com>
---
 src/v/security/audit/audit_log_manager.cc | 12 +++++-------
 src/v/security/audit/audit_log_manager.h  |  7 +++++--
 2 files changed, 10 insertions(+), 9 deletions(-)

diff --git a/src/v/security/audit/audit_log_manager.cc b/src/v/security/audit/audit_log_manager.cc
index 9124eaf232cb..1f78548edcf8 100644
--- a/src/v/security/audit/audit_log_manager.cc
+++ b/src/v/security/audit/audit_log_manager.cc
@@ -739,6 +739,10 @@ audit_log_manager::audit_log_manager(
   , _controller(controller)
   , _config(client_config)
   , _metadata_cache(metadata_cache) {
+    _probe.setup_metrics([this] {
+        return 1.0
+               - (static_cast<double>(_queue_bytes_sem.available_units()) / static_cast<double>(_max_queue_size_bytes));
+    });
     if (ss::this_shard_id() == client_shard_id) {
         _sink = std::make_unique<audit_sink>(this, controller, client_config);
     }
@@ -808,11 +812,6 @@ ss::future<> audit_log_manager::start() {
           "Redpanda is operating in recovery mode.  Auditing is disabled!");
         co_return;
     }
-    _probe = std::make_unique<audit_probe>();
-    _probe->setup_metrics([this] {
-        return 1.0
-               - (static_cast<double>(_queue_bytes_sem.available_units()) / static_cast<double>(_max_queue_size_bytes));
-    });
     if (ss::this_shard_id() != client_shard_id) {
         co_return;
     }
@@ -846,7 +845,6 @@ ss::future<> audit_log_manager::stop() {
         /// Gate may already be closed if ::pause() had been called
         co_await _gate.close();
     }
-    _probe.reset(nullptr);
     if (_queue.size() > 0) {
         vlog(
           adtlog.debug,
@@ -1006,7 +1004,7 @@ audit_log_manager::should_enqueue_audit_event() const {
           adtlog.warn,
           "Audit message passthrough active until cluster has been fully "
           "upgraded to the min supported version for audit_logging");
-        _probe->audit_error();
+        _probe.audit_error();
         return std::make_optional(audit_event_passthrough::yes);
     }
     if (_auth_misconfigured) {
diff --git a/src/v/security/audit/audit_log_manager.h b/src/v/security/audit/audit_log_manager.h
index d9caa8224e54..06be3a1a0098 100644
--- a/src/v/security/audit/audit_log_manager.h
+++ b/src/v/security/audit/audit_log_manager.h
@@ -283,7 +283,7 @@ class audit_log_manager
         return true;
     }
 
-    audit_probe& probe() { return *_probe; }
+    audit_probe& probe() { return _probe; }
 
     template<security::audit::returns_auditable_resource_vector Func>
     auto restrict_topics(Func&& func) const noexcept {
@@ -401,6 +401,10 @@ class audit_log_manager
     underlying_t _queue;
     ssx::semaphore _active_drain{1, "audit-drain"};
 
+    // Probe is mutable so it can be modified in const methods when they need to
+    // report auditing failures
+    mutable audit_probe _probe;
+
     /// Single instance contains a kafka::client::client instance.
     friend class audit_sink;
     std::unique_ptr<audit_sink> _sink;
@@ -409,7 +413,6 @@ class audit_log_manager
     model::node_id _self;
     cluster::controller* _controller;
     kafka::client::configuration& _config;
-    std::unique_ptr<audit_probe> _probe;
 
     ss::sharded<cluster::metadata_cache>* _metadata_cache;
 };