From 57c8d7a639b2548a55c20017b3e6f6855389961e Mon Sep 17 00:00:00 2001
From: Michal Maslanka <michal@redpanda.com>
Date: Thu, 5 Oct 2023 16:13:11 +0200
Subject: [PATCH] c/config_manager: populate configuration status on members
 notification

When member is added to the cluster `cluster::configuration_manager`
should be immediately aware of it instead of waiting for the first
status update.

Wired up cluster member added notification to update cluster
configuration reconciliation status with initial state for joining
member.

Fixes: #13497
Fixes: #13503

Signed-off-by: Michal Maslanka <michal@redpanda.com>
---
 src/v/cluster/config_manager.cc | 19 ++++++++++++++-----
 src/v/cluster/config_manager.h  |  2 +-
 2 files changed, 15 insertions(+), 6 deletions(-)

diff --git a/src/v/cluster/config_manager.cc b/src/v/cluster/config_manager.cc
index e24915be744f6..d0b23804b0f0d 100644
--- a/src/v/cluster/config_manager.cc
+++ b/src/v/cluster/config_manager.cc
@@ -18,9 +18,11 @@
 #include "cluster/logger.h"
 #include "cluster/members_table.h"
 #include "cluster/partition_leaders_table.h"
+#include "cluster/types.h"
 #include "config/configuration.h"
 #include "config/node_config.h"
 #include "features/feature_table.h"
+#include "model/metadata.h"
 #include "resource_mgmt/io_priority.h"
 #include "rpc/connection_cache.h"
 #include "utils/file_io.h"
@@ -81,7 +83,7 @@ config_manager::config_manager(
         /**
          * Register notification immediately not to lose status updates.
          */
-        _member_removed_notification
+        _member_update_notification
           = _members.local().register_members_updated_notification(
             [this](model::node_id id, model::membership_state new_state) {
                 handle_cluster_members_update(id, new_state);
@@ -229,17 +231,24 @@ ss::future<> config_manager::start() {
 }
 void config_manager::handle_cluster_members_update(
   model::node_id id, model::membership_state new_state) {
-    if (new_state != model::membership_state::removed) {
-        return;
+    vlog(
+      clusterlog.debug,
+      "Processing membership notification: {{id: {} state: {}}}",
+      id,
+      new_state);
+    if (new_state == model::membership_state::active) {
+        // add an empty status placeholder if node is not yet known
+        status.try_emplace(id, config_status{.node = id});
+    } else if (new_state == model::membership_state::removed) {
+        status.erase(id);
     }
-    status.erase(id);
 }
 
 ss::future<> config_manager::stop() {
     vlog(clusterlog.info, "Stopping Config Manager...");
     _reconcile_wait.broken();
     _members.local().unregister_members_updated_notification(
-      _member_removed_notification);
+      _member_update_notification);
     _leaders.local().unregister_leadership_change_notification(
       _raft0_leader_changed_notification);
     co_await _gate.close();
diff --git a/src/v/cluster/config_manager.h b/src/v/cluster/config_manager.h
index bd3e53e1f445e..3554a4b35f2e3 100644
--- a/src/v/cluster/config_manager.h
+++ b/src/v/cluster/config_manager.h
@@ -135,7 +135,7 @@ class config_manager final {
     ss::sharded<partition_leaders_table>& _leaders;
     ss::sharded<features::feature_table>& _feature_table;
     ss::sharded<cluster::members_table>& _members;
-    notification_id_type _member_removed_notification;
+    notification_id_type _member_update_notification;
     notification_id_type _raft0_leader_changed_notification;
 
     ss::condition_variable _reconcile_wait;