redpanda: the great bootstrap refactor

The start-up process for `redpanda` now looks like this:

1. Cluster configs are hydrated from the local `config_cache` and legacy
   `node_cfg_yaml`.
2. `storage::api` is constructed (but not started), and the local `kvstore`
   is recovered.
3. We bootstrap from the `kvstore`:
   1. Apply the local feature table snapshot (potentially stale) from the `kvstore`.
   2. We attempt to load any existing cluster/node UUIDs from the `kvstore`.
4. We perform cluster discovery - if we are a joiner, we will perform our RPC
   to register with the cluster, and collect the `controller_join_snapshot`
   this way. If we are simply a restarter, this is mostly a no-op.
5. We bootstrap our view of the controller - if we were a joiner, we already
   have our required snapshot. If we are a restarter, we utilize the new RPC
   to fetch the controller snapshot from the leader. Unlike the first time
   registration RPC, we do permit this to fail, and will operate with stale
   configs/feature tables during bootstrap. If we have a controller snapshot
   (obtained in either aforementioned way), it is applied to the feature table,
   and the configuration is preloaded.
6. We now have a consistent view with the rest of the cluster of the feature
   table and cluster configuration for the rest of the bootstrapping process.

raft0 replay still happens _after_ the rest of the system is started.

A big caveat with cluster recovery - `needs_restart` properties fetched can no
longer be applied until the recovered node has restarted. There is no clear way
to make this _not_ the behavior.

Author: WillemKauf

src/v/cluster/cloud_metadata/tests/cluster_recovery_backend_test.cc

@@ -126,6 +126,7 @@ TEST_P(ClusterRecoveryBackendLeadershipParamTest, TestRecoveryControllerState) {
     // Update the cluster config (via the controller, rather than shard local).
     cluster::config_update_request req;
     req.upsert.emplace_back("log_segment_size_jitter_percent", "1");
+    req.upsert.emplace_back("log_segment_size", "2147483649");
     app.controller->get_config_frontend()
       .local()
       .patch(std::move(req), model::timeout_clock::now() + 30s)
@@ -228,6 +229,7 @@ TEST_P(ClusterRecoveryBackendLeadershipParamTest, TestRecoveryControllerState) {
     raft0 = nullptr;
     restart(should_wipe::yes);
     task_local_cfg.get("log_segment_size_jitter_percent").reset();
+    task_local_cfg.get("log_segment_size").reset();
     RPTEST_REQUIRE_EVENTUALLY(5s, [this] {
         return app.storage.local().get_cluster_uuid().has_value();
     });
@@ -239,6 +241,7 @@ TEST_P(ClusterRecoveryBackendLeadershipParamTest, TestRecoveryControllerState) {
                    .has_value());
     ASSERT_NE(
       1, config::shard_local_cfg().log_segment_size_jitter_percent.value());
+    ASSERT_NE(2147483649, config::shard_local_cfg().log_segment_size.value());
     ASSERT_TRUE(!app.controller->get_credential_store().local().contains(
       security::credential_user{"userguy"}));
     ASSERT_EQ(
@@ -274,14 +277,26 @@ TEST_P(ClusterRecoveryBackendLeadershipParamTest, TestRecoveryControllerState) {
                   .is_recovery_active();
     });
 
+    bool has_restarted = false;
     // Validate the controller state is restored.
     auto validate_post_recovery = [&] {
         ASSERT_TRUE(app.controller->get_feature_table()
                       .local()
                       .get_configured_license()
                       .has_value());
+        // log_segment_size_jitter_percent is marked as needs_restart::yes. We
+        // won't see its recovered value reflected until the node is restarted.
+        auto log_segment_size_jitter_expected
+          = has_restarted ? 1
+                          : config::shard_local_cfg()
+                              .log_segment_size_jitter_percent.default_value();
         ASSERT_EQ(
-          1, config::shard_local_cfg().log_segment_size_jitter_percent.value());
+          log_segment_size_jitter_expected,
+          config::shard_local_cfg().log_segment_size_jitter_percent.value());
+        // On the other hand, log_segment_size is marked as needs_restart::no,
+        // so we will see its value reflected immediately.
+        ASSERT_EQ(
+          2147483649, config::shard_local_cfg().log_segment_size.value());
 
         // Validate User restoration.
         ASSERT_TRUE(app.controller->get_credential_store().local().contains(
@@ -339,6 +354,7 @@ TEST_P(ClusterRecoveryBackendLeadershipParamTest, TestRecoveryControllerState) {
 
     // Sanity check that the above invariants still hold after restarting.
     restart(should_wipe::no);
+    has_restarted = true;
     RPTEST_REQUIRE_EVENTUALLY(5s, [this] {
         auto latest_recovery = app.controller->get_cluster_recovery_table()
                                  .local()

src/v/kafka/client/test/fixture.h

@@ -32,6 +32,10 @@ class kafka_client_fixture : public redpanda_thread_fixture {
             config.get("disable_metrics").set_value(false);
         }).get();
         app.initialize(proxy_config(), proxy_client_config());
+        app.wire_up_pre_bootstrap_services();
+        app.hydrate_cluster_config(make_minimal_cfg());
+        app.wire_up_and_start_rpc_service();
+        app.establish_cluster_view(*app_signal);
         app.check_environment();
         app.wire_up_and_start_crypto_services();
         app.wire_up_and_start(*app_signal, test_mode);

src/v/kafka/server/tests/topic_recreate_test.cc

@@ -138,6 +138,10 @@ class recreate_test_fixture : public redpanda_thread_fixture {
             config.get("disable_metrics").set_value(false);
         }).get();
         app.initialize(proxy_config(), proxy_client_config());
+        app.wire_up_pre_bootstrap_services();
+        app.hydrate_cluster_config(make_minimal_cfg());
+        app.wire_up_and_start_rpc_service();
+        app.establish_cluster_view(*app_signal);
         app.check_environment();
         app.wire_up_and_start_crypto_services();
         app.wire_up_and_start(*app_signal, true);

src/v/redpanda/BUILD

@@ -39,6 +39,8 @@ redpanda_cc_library(
     deps = [
         ":cli_parser",
         "//src/v/base",
+        "//src/v/bytes:iobuf_parser",
+        "//src/v/bytes:iostream",
         "//src/v/cloud_io:cache",
         "//src/v/cloud_io:remote",
         "//src/v/cloud_storage",
@@ -66,6 +68,7 @@ redpanda_cc_library(
         "//src/v/cluster:offsets_lookup",
         "//src/v/cluster:partition_properties_stm",
         "//src/v/cluster:tx_manager_migrator_rpc",
+        "//src/v/cluster:types",
         "//src/v/cluster/utils:partition_change_notifier_impl",
         "//src/v/cluster_link:fwd",
         "//src/v/cluster_link:rpc_service",
@@ -95,6 +98,7 @@ redpanda_cc_library(
         "//src/v/kafka/server:write_at_offset_stm",
         "//src/v/metrics",
         "//src/v/migrations",
+        "//src/v/model",
         "//src/v/net",
         "//src/v/net:tls",
         "//src/v/pandaproxy:core",

src/v/redpanda/application.cc

@@ -16,6 +16,7 @@
 #include "cloud_storage_clients/client_pool.h"
 #include "cluster/cloud_metadata/offsets_upload_router.h"
 #include "cluster/cloud_metadata/offsets_uploader.h"
+#include "cluster/cluster_discovery.h"
 #include "cluster/config_manager.h"
 #include "cluster/controller.h"
 #include "cluster/node_isolation_watcher.h"
@@ -338,7 +339,11 @@ int application::run(int ac, char** av) {
                 // Cluster config validation uses OpenSSL (e.g. TLS cipher
                 // checks), so crypto must be initialized first.
                 wire_up_and_start_crypto_services();
+                wire_up_pre_bootstrap_services();
                 hydrate_cluster_config(node_cfg_yaml);
+                wire_up_and_start_rpc_service();
+                establish_cluster_view(app_signal);
+                log_cluster_config();
                 init_crashtracker(app_signal);
                 initialize();
                 check_environment();
@@ -387,9 +392,20 @@ void application::initialize(
   std::optional<YAML::Node> schema_reg_client_cfg,
   std::optional<YAML::Node> audit_log_client_cfg) {
     ss::smp::invoke_on_all([] {
-        // initialize memory groups now that our configuration is loaded
+        // re-initialize memory groups now that our configuration is loaded
+        memory_groups_holder().reset();
         memory_groups();
     }).get();
+
+    // With memory groups re-initialized, we can now set a proper memory
+    // capacity in the _rpc server (which was constructed before a consistent
+    // cluster view was established).
+    _rpc
+      .invoke_on_all([](rpc::rpc_server& r) {
+          r.set_memory_capacity(memory_groups().rpc_total_memory());
+      })
+      .get();
+
     construct_service(
       _memory_sampling, std::ref(_log), ss::sharded_parameter([]() {
           return config::shard_local_cfg().sampled_memory_profile.bind();
@@ -460,21 +476,6 @@ void application::initialize(
           "data directory", config::node().data_directory().path);
         syschecks::pidfile_create(config::node().pidfile_path());
     }
-    smp_groups::config smp_groups_cfg{
-      .raft_group_max_non_local_requests
-      = config::shard_local_cfg().raft_smp_max_non_local_requests().value_or(
-        smp_groups::default_raft_non_local_requests(
-          config::shard_local_cfg().topic_partitions_per_shard())),
-      .proxy_group_max_non_local_requests
-      = config::shard_local_cfg().pp_sr_smp_max_non_local_requests().value_or(
-        smp_groups::default_max_nonlocal_requests)};
-
-    smp_service_groups.create_groups(smp_groups_cfg).get();
-    _deferred.emplace_back(
-      [this] { smp_service_groups.destroy_groups().get(); });
-
-    // Ensure the scheduling groups singleton is initialized early
-    std::ignore = scheduling_groups::instance();
 
     construct_service(_scheduling_groups_probe).get();
     _scheduling_groups_probe
@@ -514,6 +515,10 @@ void application::initialize(
 }
 
 void application::setup_metrics() {
+    // Two systems that were created in the pre-bootstrapping process that may
+    // now need their metrics enabled.
+    feature_table.invoke_on_all(&features::feature_table::setup_metrics).get();
+    _rpc.invoke_on_all(&rpc::rpc_server::setup_metrics).get();
     setup_internal_metrics();
     setup_public_metrics();
 }
@@ -724,31 +729,78 @@ YAML::Node application::hydrate_node_config(const po::variables_map& cfg) {
     return config;
 }
 
-void application::hydrate_cluster_config(const YAML::Node& config) {
-    auto config_printer = [this](std::string_view service, const auto& cfg) {
-        std::vector<ss::sstring> items;
-        cfg.for_each([&items, &service](const auto& item) {
-            items.push_back(
-              ssx::sformat("{}.{}\t- {}", service, item, item.desc()));
-        });
-        std::sort(items.begin(), items.end());
-        for (const auto& item : items) {
-            vlog(_log.info, "{}", item);
-        }
-    };
+// Forward declarations of helper functions defined in application_config.cc
+std::optional<storage::file_sanitize_config> read_file_sanitizer_config();
+
+storage::kvstore_config kvstore_config_from_global_config(
+  std::optional<storage::file_sanitize_config> sanitizer_config);
+
+storage::log_config manager_config_from_global_config(
+  scheduling_groups& sgs,
+  std::optional<storage::file_sanitize_config> sanitizer_config);
+
+void application::wire_up_pre_bootstrap_services() {
+    // Ensure the scheduling groups singleton is initialized early
+    std::ignore = scheduling_groups::instance();
+
+    // Construct the feature table
+    syschecks::systemd_message("Creating feature table").get();
+    construct_service(feature_table).get();
+
+    // Construct local storage
+    const auto sanitizer_config = read_file_sanitizer_config();
+    syschecks::systemd_message("Creating storage").get();
+    construct_service(
+      storage,
+      [c = sanitizer_config]() mutable {
+          return kvstore_config_from_global_config(std::move(c));
+      },
+      [c = sanitizer_config]() mutable {
+          auto log_cfg = manager_config_from_global_config(
+            scheduling_groups::instance(), std::move(c));
+          log_cfg.reclaim_opts.background_reclaimer_sg
+            = scheduling_groups::instance().cache_background_reclaim_sg();
+          return log_cfg;
+      },
+      std::ref(feature_table))
+      .get();
+
+    // Construct smp groups using potentially stale cluster config.
+    auto& cfg = config::shard_local_cfg_unsafe();
+    smp_groups::config smp_groups_cfg{
+      .raft_group_max_non_local_requests
+      = cfg.raft_smp_max_non_local_requests().value_or(
+        smp_groups::default_raft_non_local_requests(
+          cfg.topic_partitions_per_shard())),
+      .proxy_group_max_non_local_requests
+      = cfg.pp_sr_smp_max_non_local_requests().value_or(
+        smp_groups::default_max_nonlocal_requests)};
+
+    smp_service_groups.create_groups(smp_groups_cfg).get();
+    _deferred.emplace_back(
+      [this] { smp_service_groups.destroy_groups().get(); });
+}
 
+void application::establish_cluster_view(::stop_signal& app_signal) {
+    bootstrap_from_kvstore().get();
+
+    // Begin the cluster discovery manager so we can confirm our initial node
+    // ID. A valid node ID is required before we can initialize the rest of our
+    // subsystems. The local node and cluster UUIDs would have been set in
+    // bootstrap_from_kvstore().
+    _cluster_discovery = std::make_unique<cluster::cluster_discovery>(
+      storage.local().node_uuid(),
+      storage.local().get_cluster_uuid(),
+      app_signal.abort_source());
+
+    bootstrap_controller_view().get();
+}
+
+void application::hydrate_cluster_config(const YAML::Node& config) {
     // This includes loading from local bootstrap file or legacy
     // config file on first-start or upgrade cases.
     _config_preload = cluster::config_manager::preload(config).get();
 
-    vlog(_log.info, "Cluster configuration properties:");
-    vlog(_log.info, "(use `rpk cluster config edit` to change)");
-    config_printer("redpanda", config::shard_local_cfg());
-
-    vlog(_log.info, "Node configuration properties:");
-    vlog(_log.info, "(use `rpk redpanda config set <cfg> <value>` to change)");
-    config_printer("redpanda", config::node());
-
     if (config["pandaproxy"]) {
         _proxy_config.emplace(config["pandaproxy"]);
         for (const auto& e : _proxy_config->errors()) {
@@ -768,8 +820,6 @@ void application::hydrate_cluster_config(const YAML::Node& config) {
             set_local_kafka_client_config(_proxy_client_config, config::node());
         }
         set_pp_kafka_client_defaults(*_proxy_config, *_proxy_client_config);
-        config_printer("pandaproxy", *_proxy_config);
-        config_printer("pandaproxy_client", *_proxy_client_config);
     }
     if (config["schema_registry"]) {
         _schema_reg_config.emplace(config["schema_registry"]);
@@ -780,8 +830,6 @@ void application::hydrate_cluster_config(const YAML::Node& config) {
               _schema_reg_client_config, config::node());
         }
         set_sr_kafka_client_defaults(*_schema_reg_client_config);
-        config_printer("schema_registry", *_schema_reg_config);
-        config_printer("schema_registry_client", *_schema_reg_client_config);
     }
     /// Auditing will be toggled via cluster config settings, internal audit
     /// client options can be configured via local config properties
@@ -791,7 +839,44 @@ void application::hydrate_cluster_config(const YAML::Node& config) {
         set_local_kafka_client_config(_audit_log_client_config, config::node());
     }
     set_auditing_kafka_client_defaults(*_audit_log_client_config);
-    config_printer("audit_log_client", *_audit_log_client_config);
+}
+
+void application::log_cluster_config() {
+    auto config_printer = [this](std::string_view service, const auto& cfg) {
+        std::vector<ss::sstring> items;
+        cfg.for_each([&items, &service](const auto& item) {
+            items.push_back(
+              ssx::sformat("{}.{}\t- {}", service, item, item.desc()));
+        });
+        std::sort(items.begin(), items.end());
+        for (const auto& item : items) {
+            vlog(_log.info, "{}", item);
+        }
+    };
+
+    vlog(_log.info, "Cluster configuration properties:");
+    vlog(_log.info, "(use `rpk cluster config edit` to change)");
+    config_printer("redpanda", config::shard_local_cfg());
+
+    vlog(_log.info, "Node configuration properties:");
+    vlog(_log.info, "(use `rpk redpanda config set <cfg> <value>` to change)");
+    config_printer("redpanda", config::node());
+
+    if (_proxy_config) {
+        config_printer("pandaproxy", *_proxy_config);
+    }
+    if (_proxy_client_config) {
+        config_printer("pandaproxy_client", *_proxy_client_config);
+    }
+    if (_schema_reg_config) {
+        config_printer("schema_registry", *_schema_reg_config);
+    }
+    if (_schema_reg_client_config) {
+        config_printer("schema_registry_client", *_schema_reg_client_config);
+    }
+    if (_audit_log_client_config) {
+        config_printer("audit_log_client", *_audit_log_client_config);
+    }
 }
 
 void application::check_environment() {