add http api for reconcile leader in case of imbalanced leadership distribution

Author: yawzhang

conanfile.py

@@ -10,7 +10,7 @@
 
 class HomeObjectConan(ConanFile):
     name = "homeobject"
-    version = "2.7.3"
+    version = "2.7.4"
 
     homepage = "https://github.com/eBay/HomeObject"
     description = "Blob Store built on HomeReplication"

src/lib/homestore_backend/hs_homeobject.cpp

@@ -478,4 +478,30 @@ void HSHomeObject::trigger_immediate_gc() {
     }
 }
 
+void HSHomeObject::reconcile_pg_leader(int32_t pg_id) {
+    if (pg_id == -1) {
+        LOGI("PG id not set, start reconciling leaders for all PGs");
+        std::shared_lock lock_guard(_pg_lock);
+        std::vector< std::future< void > > futures;
+        for (const auto& [id, pg] : _pg_map) {
+            auto hs_pg = static_cast< HS_PG* >(pg.get());
+            futures.emplace_back(std::async(std::launch::async, [hs_pg, id]() {
+                hs_pg->reconcile_leader();
+                LOGI("Triggered reconcile leader for PG {}", id);
+            }));
+        }
+        for (auto& future : futures) {
+            future.get();
+        }
+    } else {
+        LOGI("Reconciling leader for PG {}", pg_id);
+        auto hs_pg = get_hs_pg(pg_id);
+        if (hs_pg) {
+            hs_pg->reconcile_leader();
+        } else {
+            LOGE("PG {} not found", pg_id);
+        }
+    }
+}
+
 } // namespace homeobject

src/lib/homestore_backend/hs_homeobject.hpp

@@ -361,6 +361,8 @@ class HSHomeObject : public HomeObjectImpl {
          * Returns all replication info of all peers.
          */
         void get_peer_info(std::vector< peer_info >& members) const;
+
+        void reconcile_leader() const;
     };
 
     struct HS_Shard : public Shard {
@@ -881,6 +883,17 @@ class HSHomeObject : public HomeObjectImpl {
     cshared< HeapChunkSelector > chunk_selector() const { return chunk_selector_; }
     cshared< GCManager > gc_manager() const { return gc_mgr_; }
 
+    /**
+     * @brief Reconciles the leaders for all PGs or a specific PG identified by pg_id.
+     *
+     * This function ensures that the leadership status of PGs is consistent across the system.
+     * If a specific pg_id is provided, it reconciles the leader for that particular PG.
+     * If pg_id is -1 (default), it reconciles leaders for all PGs.
+     *
+     * @param pg_id The ID of the PG to reconcile. Default is -1, which indicates all PGs.
+     */
+    void reconcile_pg_leader(int32_t pg_id = -1);
+
     // Blob manager related.
     void on_blob_message_rollback(int64_t lsn, sisl::blob const& header, sisl::blob const& key,
                                   cintrusive< homestore::repl_req_ctx >& hs_ctx);

src/lib/homestore_backend/hs_http_manager.cpp

@@ -33,6 +33,8 @@ HttpManager::HttpManager(HSHomeObject& ho) : ho_(ho) {
          Pistache::Rest::Routes::bind(&HttpManager::get_obj_life, this)},
         {Pistache::Http::Method::Get, "/api/v1/mallocStats",
          Pistache::Rest::Routes::bind(&HttpManager::get_malloc_stats, this)},
+        {Pistache::Http::Method::Post, "/api/v1/reconcile_leader",
+         Pistache::Rest::Routes::bind(&HttpManager::reconcile_leader, this)},
 #ifdef _PRERELEASE
         {Pistache::Http::Method::Post, "/api/v1/crashSystem",
          Pistache::Rest::Routes::bind(&HttpManager::crash_system, this)},
@@ -63,6 +65,14 @@ void HttpManager::get_malloc_stats(const Pistache::Rest::Request& request, Pista
     response.send(Pistache::Http::Code::Ok, sisl::get_malloc_stats_detailed().dump(2));
 }
 
+void HttpManager::reconcile_leader(const Pistache::Rest::Request& request, Pistache::Http::ResponseWriter response) {
+    const auto pg_id_param = request.query().get("pg_id");
+    int32_t pg_id = std::stoi(pg_id_param.value_or("-1"));
+    LOGINFO("Received reconcile leader request for pg_id {}", pg_id);
+    ho_.reconcile_pg_leader(pg_id);
+    response.send(Pistache::Http::Code::Ok, "Reconcile leader request submitted");
+}
+
 #ifdef _PRERELEASE
 void HttpManager::crash_system(const Pistache::Rest::Request& request, Pistache::Http::ResponseWriter response) {
     std::string crash_type;

src/lib/homestore_backend/hs_http_manager.hpp

@@ -26,6 +26,7 @@ class HttpManager {
 private:
     void get_obj_life(const Pistache::Rest::Request& request, Pistache::Http::ResponseWriter response);
     void get_malloc_stats(const Pistache::Rest::Request& request, Pistache::Http::ResponseWriter response);
+    void reconcile_leader(const Pistache::Rest::Request& request, Pistache::Http::ResponseWriter response);
 
 #ifdef _PRERELEASE
     void crash_system(const Pistache::Rest::Request& request, Pistache::Http::ResponseWriter response);

src/lib/homestore_backend/hs_pg_manager.cpp

@@ -765,6 +765,8 @@ void HSHomeObject::HS_PG::get_peer_info(std::vector< peer_info >& members) const
     }
 }
 
+void HSHomeObject::HS_PG::reconcile_leader() const { repl_dev_->reconcile_leader(); }
+
 // NOTE: caller should hold the _pg_lock
 const HSHomeObject::HS_PG* HSHomeObject::_get_hs_pg_unlocked(pg_id_t pg_id) const {
     auto iter = _pg_map.find(pg_id);