config: add shard_local_cfg_unsafe()

And swap any uses of `shard_local_cfg()` which occur in the bootstrap
process (before it is marked as `ready`) over to the unsafe accessor.
These uses of potentially stale values are all audited and deemed safe,
due to minimal or no impact on the behavior of the system post bootstrap.

Two other relevant changes in this commit:
* Pulling out `feature_table::setup_metrics()` into its own function,
  called after the bootstrap process (in order to properly respect
  `shard_local_cfg().disable_{public}_metrics`)
* Updating `storage_resources` to _not_ invoke the `internal::chunks()`
  singleton constructor, which would in turn invoke `shard_local_cfg()`.
  We use `storage_resources for an early replay of the `kvstore` segments,
  so it is fine to use stale configuration values here. However, after
  bootstrapping is complete, it is *probably* important to set the
  `_append_chunk_size` back to the same value as what `internal::chunks()`
  is operating with. We now do so when initializing the storage system in
  `wire_up_bootstrap_services()`.

Author: WillemKauf

src/v/cluster/cluster_discovery.cc

@@ -37,7 +37,7 @@ cluster_discovery::cluster_discovery(
   ss::abort_source& as)
   : _node_uuid(node_uuid)
   , _cluster_uuid(std::move(cluster_uuid))
-  , _join_retry_jitter(config::shard_local_cfg().join_retry_timeout_ms())
+  , _join_retry_jitter(config::shard_local_cfg_unsafe().join_retry_timeout_ms())
   , _join_timeout(std::chrono::seconds(2))
   , _as(as) {}
 

src/v/cluster/config_manager.cc

@@ -145,7 +145,7 @@ ss::future<> config_manager::wait_for_bootstrap() {
 ss::future<> config_manager::do_bootstrap() {
     config_update_request update;
 
-    config::shard_local_cfg().for_each(
+    config::shard_local_cfg_unsafe().for_each(
       [&update](const config::base_property& p) {
           if (!p.is_default()) {
               json::StringBuffer buf;
@@ -292,7 +292,7 @@ static void preload_local(
   const ss::sstring& raw_value,
   std::optional<std::reference_wrapper<config_manager::preload_result>>
     result) {
-    auto& cfg = config::shard_local_cfg();
+    auto& cfg = config::shard_local_cfg_unsafe();
     if (cfg.contains(key)) {
         auto& property = cfg.get(key);
         try {
@@ -339,7 +339,7 @@ static void preload_local(
   const YAML::Node& value,
   std::optional<std::reference_wrapper<config_manager::preload_result>>
     result) {
-    auto& cfg = config::shard_local_cfg();
+    auto& cfg = config::shard_local_cfg_unsafe();
     if (cfg.contains(key)) {
         auto& property = cfg.get(key);
         std::string raw_value;
@@ -408,7 +408,7 @@ config_manager::preload(const YAML::Node& legacy_config) {
         // to set something in redpanda.yaml and it's not working.
         if (legacy_config["redpanda"]) {
             const auto nag_properties
-              = config::shard_local_cfg().property_names_and_aliases();
+              = config::shard_local_cfg_unsafe().property_names_and_aliases();
             for (const auto& node : legacy_config["redpanda"]) {
                 auto name = node.first.as<ss::sstring>();
                 if (nag_properties.contains(name)) {
@@ -447,7 +447,7 @@ ss::future<bool> config_manager::load_bootstrap() {
 
     // This node has never seen a cluster configuration message.
     // Bootstrap configuration from local yaml file.
-    auto errors = config::shard_local_cfg().read_yaml(config);
+    auto errors = config::shard_local_cfg_unsafe().read_yaml(config);
     for (const auto& i : errors) {
         vlog(
           clusterlog.warn,
@@ -457,18 +457,19 @@ ss::future<bool> config_manager::load_bootstrap() {
     }
 
     co_await ss::smp::invoke_on_all(
-      [&config] { config::shard_local_cfg().read_yaml(config); });
+      [&config] { config::shard_local_cfg_unsafe().read_yaml(config); });
 
     co_return true;
 }
 
 ss::future<> config_manager::load_legacy(const YAML::Node& legacy_config) {
-    co_await ss::smp::invoke_on_all(
-      [&legacy_config] { config::shard_local_cfg().load(legacy_config); });
+    co_await ss::smp::invoke_on_all([&legacy_config] {
+        config::shard_local_cfg_unsafe().load(legacy_config);
+    });
 
     // This node has never seen a cluster configuration message.
     // Bootstrap configuration from local yaml file.
-    auto errors = config::shard_local_cfg().load(legacy_config);
+    auto errors = config::shard_local_cfg_unsafe().load(legacy_config);
 
     // Report any invalid properties.  Do not refuse to start redpanda,
     // as the properties will have been either ignored or clamped
@@ -625,7 +626,7 @@ ss::future<> config_manager::reconcile_status() {
  */
 config_manager::apply_result
 apply_local(const cluster_config_delta_cmd_data& data, bool silent) {
-    auto& cfg = config::shard_local_cfg();
+    auto& cfg = config::shard_local_cfg_unsafe();
     auto result = config_manager::apply_result{};
     for (const auto& u : data.upsert) {
         if (!cfg.contains(u.key)) {
@@ -813,7 +814,7 @@ void config_manager::merge_apply_result(
  */
 ss::future<>
 config_manager::store_delta(const cluster_config_delta_cmd_data& data) {
-    auto& cfg = config::shard_local_cfg();
+    auto& cfg = config::shard_local_cfg_unsafe();
 
     for (const auto& u : data.upsert) {
         /// skip section

src/v/config/configuration.cc

@@ -4981,8 +4981,18 @@ std::unique_ptr<configuration> make_config() {
     }
 }
 
-configuration& shard_local_cfg() {
+configuration& shard_local_cfg_unsafe() {
     static thread_local std::unique_ptr<configuration> cfg = make_config();
     return *cfg;
 }
+
+configuration& shard_local_cfg() {
+    auto& cfg = shard_local_cfg_unsafe();
+    vassert(
+      cfg.is_ready(),
+      "Used shard_local_cfg() before it was marked as ready by the startup "
+      "process");
+    return cfg;
+}
+
 } // namespace config

src/v/config/configuration.h

@@ -846,6 +846,9 @@ struct configuration final : public config_store {
 
     development_feature_property<int> development_feature_property_testing_only;
 
+    void mark_ready(bool ready) { _ready = ready; }
+    bool is_ready() const { return _ready; }
+
 private:
     // to query if experimental features are enabled in order to log a nag. it
     // does not use the query to control any experimental feature.
@@ -874,6 +877,8 @@ struct configuration final : public config_store {
     }
 
     ss::sstring store_name() const override { return "cluster"; }
+
+    bool _ready{true};
 };
 
 template<typename T>
@@ -884,6 +889,21 @@ bool development_feature_property<T>::development_features_enabled(
 
 std::unique_ptr<configuration> make_config();
 
+// Accessors for the shard local cluster configuration.
+// shard_local_cfg() contains a vassert which strictly enforces that the
+// configuration has been marked as ready by the bootstrap process after we have
+// successfully preloaded from our node local state, and furthermore, recieved
+// an up to date snapshot from the controller leader of the cluster wide view of
+// the configuration.
+// However, some configurations are somewhat required to be accessed in the
+// bootstrapping process before this can happen. For that reason, we provide a
+// shard_local_cfg_unsafe() accessor, which does not perform this same check.
+// It is highly recommended that use of this is kept to a minimum, and if you do
+// need to use it, that you are confident that using a potentially stale
+// configuration value during the bootstrapping process will not have any
+// ramifications on the behavior of the system after its value is updated during
+// bootstrapping.
 configuration& shard_local_cfg();
+configuration& shard_local_cfg_unsafe();
 
 } // namespace config

src/v/config/node_config.cc

@@ -341,7 +341,7 @@ node_config::error_map_t node_config::load(const YAML::Node& root_node) {
         throw std::invalid_argument("'redpanda' root is required");
     }
 
-    auto ignore = shard_local_cfg().property_names_and_aliases();
+    auto ignore = shard_local_cfg_unsafe().property_names_and_aliases();
 
     auto errors = config_store::read_yaml(root_node["redpanda"]);
     validate_multi_node_property_config(errors);

src/v/features/feature_table.cc

@@ -828,7 +828,7 @@ feature_table::decode_version_fence(model::record_batch batch) {
 void feature_table::set_original_version(cluster::cluster_version v) {
     _original_version = v;
     if (v != cluster::invalid_version) {
-        config::shard_local_cfg().notify_original_version(
+        config::shard_local_cfg_unsafe().notify_original_version(
           config::legacy_version{v});
     }
 }

src/v/redpanda/application.cc

@@ -31,6 +31,7 @@
 #include "datalake/credential_manager.h"
 #include "datalake/datalake_manager.h"
 #include "datalake/datalake_usage_aggregator.h"
+#include "features/feature_table.h"
 #include "kafka/client/configuration.h"
 #include "kafka/server/rm_group_frontend.h"
 #include "metrics/prometheus_sanitize.h"
@@ -340,6 +341,7 @@ int application::run(int ac, char** av) {
                 // checks), so crypto must be initialized first.
                 wire_up_and_start_crypto_services();
                 wire_up_pre_bootstrap_services();
+                mark_config_ready(false).get();
                 hydrate_cluster_config(node_cfg_yaml);
                 wire_up_and_start_rpc_service();
                 establish_cluster_view(app_signal);
@@ -802,6 +804,15 @@ void application::establish_cluster_view(::stop_signal& app_signal) {
       app_signal.abort_source());
 
     bootstrap_controller_view().get();
+
+    // The shard_local_cfg() is now safe to use as we have a view consistent
+    // with the rest of the cluster.
+    mark_config_ready(true).get();
+}
+
+ss::future<> application::mark_config_ready(bool ready) {
+    co_await ss::smp::invoke_on_all(
+      [ready] { config::shard_local_cfg_unsafe().mark_ready(ready); });
 }
 
 void application::hydrate_cluster_config(const YAML::Node& config) {

src/v/redpanda/application.h

@@ -264,6 +264,8 @@ class application : public ssx::sharded_service_container {
       std::optional<cloud_storage_clients::bucket_name>& bucket_name,
       cloud_topics::test_fixture_cfg ct_test_cfg);
 
+    // Marks the shard_local_cfg as ready per the provided flag.
+    ss::future<> mark_config_ready(bool ready);
     // Performs recovery on the local kvstore, applies a local feature table
     // snapshot, and sets in-memory node/cluster UUIDs.
     ss::future<> bootstrap_from_kvstore();

src/v/redpanda/application_bootstrap.cc

@@ -92,6 +92,12 @@ void application::wire_up_and_start_crypto_services() {
 
 void application::wire_up_bootstrap_services() {
     // Wire up local storage.
+    storage
+      .invoke_on_all([](storage::api& s) {
+          s.resources().update_append_chunk_size(
+            storage::internal::chunks().chunk_size());
+      })
+      .get();
     ss::smp::invoke_on_all([] {
         return storage::internal::chunks().start();
     }).get();

src/v/redpanda/application_config.cc

@@ -175,8 +175,8 @@ storage::kvstore_config kvstore_config_from_global_config(
      *           - ... #cores
      */
     return storage::kvstore_config(
-      config::shard_local_cfg().kvstore_max_segment_size(),
-      config::shard_local_cfg().kvstore_flush_interval.bind(),
+      config::shard_local_cfg_unsafe().kvstore_max_segment_size(),
+      config::shard_local_cfg_unsafe().kvstore_flush_interval.bind(),
       config::node().data_directory().as_sstring(),
       sanitizer_config);
 }