Initialize blockstore after loading pool configuration to pre-shard the DB correctly on start

src/blockstore/blockstore.h

@@ -183,6 +183,9 @@ public:
     // Update configuration
     virtual void parse_config(blockstore_config_t & config) = 0;
 
+    // Reshard database for a pool
+    virtual void reshard(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size) = 0;
+
     // Event loop
     virtual void loop() = 0;
 

src/blockstore/blockstore_impl.cpp

@@ -23,6 +23,7 @@ blockstore_impl_t::blockstore_impl_t(blockstore_config_t & config, ring_loop_i *
         dsk.open_meta();
         dsk.open_journal();
         dsk.calc_lengths();
+        dsk.check_lengths();
     }
     catch (std::exception & e)
     {
@@ -31,16 +32,13 @@ blockstore_impl_t::blockstore_impl_t(blockstore_config_t & config, ring_loop_i *
     }
     meta_superblock = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.meta_block_size);
     memset(meta_superblock, 0, dsk.meta_block_size);
-}
-
-void blockstore_impl_t::init()
-{
     flusher = new journal_flusher_t(this);
     if (dsk.inmemory_journal)
     {
         buffer_area = (uint8_t*)memalign_or_die(MEM_ALIGNMENT, dsk.journal_len);
     }
     heap = new blockstore_heap_t(&dsk, buffer_area, log_level);
+    ringloop->wakeup();
 }
 
 blockstore_impl_t::~blockstore_impl_t()
@@ -394,3 +392,8 @@ std::string blockstore_impl_t::get_op_diag(blockstore_op_t *op)
         snprintf(buf, sizeof(buf), "state=%d", priv->op_state);
     return std::string(buf);
 }
+
+void blockstore_impl_t::reshard(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size)
+{
+    heap->reshard(pool, pg_count, pg_stripe_size);
+}

src/blockstore/blockstore_impl.h

@@ -142,7 +142,6 @@ public:
     int metadata_buf_size;
     blockstore_init_meta* metadata_init_reader;
 
-    void init();
     void check_wait(blockstore_op_t *op);
     void init_op(blockstore_op_t *op);
 
@@ -190,6 +189,8 @@ public:
     void parse_config(blockstore_config_t & config);
     void parse_config(blockstore_config_t & config, bool init);
 
+    void reshard(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size);
+
     // Event loop
     void loop();
 

src/blockstore/blockstore_init.cpp

@@ -72,7 +72,6 @@ resume_1:
     }
     if (is_zero((uint64_t*)bs->meta_superblock, bs->dsk.meta_block_size))
     {
-        bs->dsk.check_lengths();
         {
             blockstore_meta_header_v3_t *hdr = (blockstore_meta_header_v3_t *)bs->meta_superblock;
             hdr->zero = 0;
@@ -141,7 +140,7 @@ resume_1:
             hdr->bitmap_granularity != bs->dsk.bitmap_granularity ||
             hdr->data_csum_type != bs->dsk.data_csum_type ||
             hdr->csum_block_size != bs->dsk.csum_block_size ||
-            hdr->meta_area_size > bs->dsk.meta_area_size)
+            hdr->meta_area_size != bs->dsk.meta_area_size)
         {
             printf(
                 "Configuration stored in metadata superblock"
@@ -154,15 +153,7 @@ resume_1:
             );
             exit(1);
         }
-        bs->dsk.meta_area_size = hdr->meta_area_size;
-        if (bs->dsk.meta_format != hdr->version)
-        {
-            bs->dsk.meta_format = hdr->version;
-            bs->dsk.calc_lengths();
-        }
-        bs->dsk.check_lengths();
     }
-    bs->init();
     bs->heap->start_load(((blockstore_meta_header_v3_t *)bs->meta_superblock)->completed_lsn);
     if (bs->dsk.inmemory_journal)
     {

src/blockstore/v1/impl.cpp

@@ -807,4 +807,9 @@ std::string blockstore_impl_t::get_op_diag(blockstore_op_t *op)
     return std::string(buf);
 }
 
+void blockstore_impl_t::reshard(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size)
+{
+    reshard_clean_db(pool, pg_count, pg_stripe_size);
+}
+
 } // namespace v1

src/blockstore/v1/impl.h

@@ -288,6 +288,9 @@ public:
     void parse_config(blockstore_config_t & config);
     void parse_config(blockstore_config_t & config, bool init);
 
+    // Reshard database for a pool
+    void reshard(pool_id_t pool, uint32_t pg_count, uint32_t pg_stripe_size);
+
     // Event loop
     void loop();
 

src/client/etcd_state_client.cpp

@@ -568,9 +568,19 @@ void etcd_state_client_t::start_ws_keepalive()
 
 void etcd_state_client_t::load_global_config()
 {
-    etcd_call("/kv/range", json11::Json::object {
+    json11::Json::object req = { { "success", json11::Json::array {
-        { "key", base64_encode(etcd_prefix+"/config/global") }
+        json11::Json::object {
-    }, etcd_quick_timeout, max_etcd_attempts, 0, [this](std::string err, json11::Json data)
+            { "request_range", json11::Json::object {
+                { "key", base64_encode(etcd_prefix+"/config/global") },
+            } }
+        },
+        json11::Json::object {
+            { "request_range", json11::Json::object {
+                { "key", base64_encode(etcd_prefix+"/config/pools") },
+            } }
+        },
+    } } };
+    etcd_txn(req, etcd_quick_timeout, max_etcd_attempts, 0, [this](std::string err, json11::Json data)
     {
         if (err != "")
         {
@@ -588,10 +598,12 @@ void etcd_state_client_t::load_global_config()
             }
             return;
         }
+        json11::Json config_kv = data["responses"][0]["response_range"]["kvs"][0];
+        json11::Json pools_kv = data["responses"][1]["response_range"]["kvs"][0];
         json11::Json::object global_config;
-        if (data["kvs"].array_items().size() > 0)
+        if (!config_kv.is_null())
         {
-            auto kv = parse_etcd_kv(data["kvs"][0]);
+            auto kv = parse_etcd_kv(config_kv);
             if (kv.value.is_object())
             {
                 global_config = kv.value.object_items();
@@ -608,6 +620,11 @@ void etcd_state_client_t::load_global_config()
             global_bitmap_granularity = DEFAULT_BITMAP_GRANULARITY;
         }
         global_immediate_commit = parse_immediate_commit(global_config["immediate_commit"].string_value(), IMMEDIATE_ALL);
+        if (!pools_kv.is_null())
+        {
+            auto kv = parse_etcd_kv(pools_kv);
+            parse_state(kv);
+        }
         on_load_config_hook(global_config);
     });
 }

src/osd/osd.cpp

@@ -31,25 +31,6 @@ osd_t::osd_t(const json11::Json & config, ring_loop_t *ringloop)
     // FIXME: Use timerfd_interval based directly on io_uring
     this->tfd = epmgr->tfd;
 
-    if (!json_is_true(this->config["disable_blockstore"]))
-    {
-        auto bs_cfg = json_to_string_map(this->config);
-        this->bs = blockstore_i::create(bs_cfg, ringloop, tfd);
-        // Wait for blockstore initialisation before actually starting OSD logic
-        // to prevent peering timeouts during restart with filled databases
-        while (!bs->is_started())
-        {
-            ringloop->loop();
-            if (bs->is_started())
-                break;
-            ringloop->wait();
-        }
-        // Autosync based on the number of unstable writes to prevent stalls due to insufficient journal space
-        uint64_t max_autosync = bs->get_journal_size() / bs->get_block_size() / 2;
-        if (autosync_writes > max_autosync)
-            autosync_writes = max_autosync;
-    }
-
     if (json_is_true(this->config["osd_memlock"]))
     {
         // Lock all OSD memory if requested
@@ -117,6 +98,7 @@ osd_t::~osd_t()
         autosync_timer_id = -1;
     }
     ringloop->unregister_consumer(&consumer);
+    ringloop->unregister_consumer(&init_consumer);
     delete epmgr;
     if (bs)
         delete bs;
@@ -131,6 +113,42 @@ osd_t::~osd_t()
     free(zero_buffer);
 }
 
+void osd_t::init_blockstore(std::function<void()> on_init)
+{
+    if (!json_is_true(this->config["disable_blockstore"]))
+    {
+        auto bs_cfg = json_to_string_map(this->config);
+        this->bs = blockstore_i::create(bs_cfg, ringloop, tfd);
+        // Pre-configure pool PG shards
+        for (auto & pool_item: st_cli.pool_config)
+        {
+            bs->reshard(pool_item.first, pool_item.second.pg_count, pool_item.second.pg_stripe_size);
+        }
+        // Autosync based on the number of unstable writes to prevent stalls due to insufficient journal space
+        uint64_t max_autosync = bs->get_journal_size() / bs->get_block_size() / 2;
+        if (autosync_writes > max_autosync)
+            autosync_writes = max_autosync;
+        if (on_init)
+        {
+            init_consumer.loop = [this, on_init]()
+            {
+                // Wait for blockstore initialisation before actually starting OSD logic
+                // to prevent peering timeouts during restart with filled databases
+                if (bs->is_started())
+                {
+                    ringloop->set_immediate([this, on_init] { init_consumer.loop = NULL; on_init(); });
+                    ringloop->unregister_consumer(&init_consumer);
+                }
+            };
+            ringloop->register_consumer(&init_consumer);
+        }
+    }
+    else if (on_init)
+    {
+        on_init();
+    }
+}
+
 void osd_t::parse_config(bool init)
 {
     config = msgr.merge_configs(cli_config, file_config, etcd_global_config, etcd_osd_config);

src/osd/osd.h

@@ -158,6 +158,7 @@ class osd_t
     json11::Json self_state;
     bool loading_peer_config = false;
     std::set<pool_pg_num_t> pg_state_dirty;
+    bool etcd_global_config_loaded = false;
     bool pg_config_applied = false;
     bool etcd_reporting_pg_state = false;
     bool etcd_reporting_stats = false;
@@ -206,7 +207,7 @@ class osd_t
     void *zero_buffer = NULL;
     uint64_t zero_buffer_size = 0;
     uint32_t bs_block_size, bs_bitmap_granularity, clean_entry_bitmap_size;
-    ring_loop_t *ringloop;
+    ring_loop_t *ringloop = NULL;
     timerfd_manager_t *tfd = NULL;
     epoll_manager_t *epmgr = NULL;
 
@@ -218,6 +219,7 @@ class osd_t
     int rdmacm_port = 0;
 #endif
     ring_consumer_t consumer;
+    ring_consumer_t init_consumer;
 
     // op statistics
     osd_op_stats_t prev_stats, prev_report_stats;
@@ -241,6 +243,7 @@ class osd_t
 
     // cluster connection
     void parse_config(bool init);
+    void init_blockstore(std::function<void()> on_init);
     void init_cluster();
     void on_change_osd_state_hook(osd_num_t peer_osd);
     void on_change_backfillfull_hook(pool_id_t pool_id);

src/osd/osd_cluster.cpp

@@ -18,6 +18,7 @@ void osd_t::init_cluster()
 {
     if (!st_cli.address_count())
     {
+        init_blockstore(NULL);
         if (run_primary)
         {
             // Test version of clustering code with 1 pool, 1 PG and 2 peers
@@ -422,7 +423,10 @@ void osd_t::on_change_osd_state_hook(osd_num_t peer_osd)
 
 void osd_t::on_change_pool_config_hook()
 {
-    apply_pg_locks_localize_only();
+    if (etcd_global_config_loaded)
+    {
+        apply_pg_locks_localize_only();
+    }
 }
 
 void osd_t::apply_pg_locks_localize_only()
@@ -484,9 +488,13 @@ void osd_t::on_load_config_hook(json11::Json::object & global_config)
 {
     etcd_global_config = global_config;
     parse_config(true);
-    bind_socket();
-    acquire_lease();
     st_cli.on_load_config_hook = [this](json11::Json::object & cfg) { on_reload_config_hook(cfg); };
+    etcd_global_config_loaded = true;
+    init_blockstore([this]()
+    {
+        bind_socket();
+        acquire_lease();
+    });
 }
 
 void osd_t::on_reload_config_hook(json11::Json::object & global_config)