support manual trigger snapshot creation function

Author: yawzhang

conanfile.py

@@ -9,7 +9,7 @@
 
 class HomestoreConan(ConanFile):
     name = "homestore"
-    version = "7.1.0"
+    version = "7.1.1"
 
     homepage = "https://github.com/eBay/Homestore"
     description = "HomeStore Storage Engine"

src/include/homestore/replication/repl_dev.h

@@ -621,6 +621,9 @@ class ReplDev {
     // clear reqs that has allocated blks on the given chunk.
     virtual void clear_chunk_req(chunk_num_t chunk_id) = 0;
 
+    // create a snapshot manually and try to compact logs upto compact_lsn
+    virtual void trigger_snapshot_creation(repl_lsn_t compact_lsn, bool wait_for_commit) = 0;
+
 protected:
     shared< ReplDevListener > m_listener;
 

src/include/homestore/replication_service.hpp

@@ -110,6 +110,14 @@ class ReplicationService {
     /// @param group_id Group where the replace member happens
     virtual ReplServiceError destroy_repl_dev(group_id_t group_id, uint64_t trace_id = 0) = 0;
 
+    /// @brief Trigger a snapshot creation manually for a given group id
+    /// @param group_id Group id for which snapshot creation is requested
+    /// @param compact_lsn LSN upto which the snapshot expect to compact, -1 means no compaction upper needed
+    /// @param wait_for_commit default is true, means it will wait until the local committed lsn reach the compact_lsn,
+    /// if set to false, the snapshot and compaction will be triggered based on the latest committed lsn
+    virtual void trigger_snapshot_creation(group_id_t group_id, repl_lsn_t compact_lsn = -1,
+                                           bool wait_for_commit = true) = 0;
+
     /// @brief Get the repl dev for a given group id if it is already created or opened
     /// @param group_id Group id interested in
     /// @return ReplDev is opened or ReplServiceError::SERVER_NOT_FOUND if it doesn't exist

src/lib/replication/repl_dev/raft_repl_dev.cpp

@@ -46,8 +46,9 @@ RaftReplDev::RaftReplDev(RaftReplService& svc, superblk< raft_repl_dev_superblk
             });
         m_next_dsn = m_rd_sb->last_applied_dsn + 1;
         m_commit_upto_lsn = m_rd_sb->durable_commit_lsn;
-        m_last_flushed_commit_lsn = m_commit_upto_lsn;
+        m_last_flushed_cp_lsn = m_rd_sb->checkpoint_lsn;
         m_compact_lsn = m_rd_sb->compact_lsn;
+        m_last_flushed_compact_lsn = m_compact_lsn;
 
         m_rdev_name = m_rd_sb->rdev_name;
         // Its ok not to do compare exchange, because loading is always single threaded as of now
@@ -818,6 +819,54 @@ void RaftReplDev::on_create_snapshot(nuraft::snapshot& s, nuraft::async_result<
     if (when_done) { when_done(ret_val, null_except); }
 }
 
+void RaftReplDev::trigger_snapshot_creation(repl_lsn_t compact_lsn, bool wait_for_commit) {
+    // Step 1. Update truncation boundary if compact_lsn is specified
+    if (compact_lsn >= 0) {
+        // Step 1.1 Wait for commit upto compact_lsn if needed
+        if (wait_for_commit) {
+            RD_LOGI(NO_TRACE_ID, "Waiting for commit upto compact_lsn={}, current_commit_lsn={}", compact_lsn,
+                    m_commit_upto_lsn.load());
+            while (compact_lsn > m_commit_upto_lsn.load()) {
+                std::this_thread::sleep_for(std::chrono::milliseconds(100));
+            }
+        }
+        // Step 1.2 trigger cp_flush to make sure all changes are flushed to disk before updating truncation boundary
+        hs()->cp_mgr().trigger_cp_flush(true /*force*/).get();
+        RD_LOGI(NO_TRACE_ID, "cp_flush completed before updating truncation boundary to lsn={}", compact_lsn);
+        // Step 1.3 Update truncation boundary
+        RD_LOGI(NO_TRACE_ID, "Updating truncation boundary to lsn={}, current_truncation_boundary={}", compact_lsn,
+                m_truncation_upper_limit.load());
+        update_truncation_boundary(compact_lsn);
+    }
+
+    // Step 2. Create snapshot manually
+    auto result = raft_server()->create_snapshot();
+    if (result == 0) {
+        RD_LOGE(NO_TRACE_ID, "Failed to create snapshot - another snapshot may be in progress");
+        return;
+    }
+    RD_LOGI(NO_TRACE_ID, "Successfully created snapshot, result log_idx={}", result);
+
+    // Step 3. Make sure logs are compacted upto compact_lsn after snapshot created
+    // Since we want to compact logs upto compact_lsn after the snapshot created, here we check the compact_lsn and
+    // trigger a compact manually if needed (since logs will only to truncated upto min(m_truncation_upper_limit,
+    // snp_idx-snp_distance), there is a that real_compact_lsn < compact_lsn after snapshot created).
+    // Note that manual compact may cause concurrency issue with the automatic log compaction in raft server. Currently,
+    // it doesn't matter because this API is used in op scenario when members syncup and no new write are handled. If we
+    // want to use it in more phase, we need to add lock to avoid concurrency issue.
+    if (compact_lsn > 0 && m_compact_lsn.load() < compact_lsn) {
+        RD_LOGI(NO_TRACE_ID, "Manually compacting logs upto lsn={} after snapshot, current_compact_lsn={}", compact_lsn,
+                m_compact_lsn.load());
+        m_data_journal->compact(compact_lsn);
+        RD_LOGI(NO_TRACE_ID, "Compacted logs upto lsn={} after snapshot", compact_lsn);
+    }
+
+    // Step 4. trigger cp_flush to make sure all changes are flushed to disk after snapshot creation and log compaction
+    hs()->cp_mgr().trigger_cp_flush(true /*force*/).get();
+    RD_LOGI(NO_TRACE_ID, "cp_flush completed after snapshot creation and log compaction");
+    RD_LOGI(NO_TRACE_ID, "snapshot creation and compaction completed");
+}
+
 void RaftReplDev::propose_truncate_boundary() {
     init_req_counter counter(pending_request_num);
     auto repl_status = get_replication_status();
@@ -1591,7 +1640,7 @@ void RaftReplDev::handle_commit(repl_req_ptr_t rreq, bool recovery) {
         complete_replace_member(rreq);
         break;
     case journal_type_t::HS_CTRL_UPDATE_TRUNCATION_BOUNDARY:
-        update_truncation_boundary(rreq);
+        update_truncation_boundary(r_cast< const truncate_ctx* >(rreq->header().cbytes())->truncation_upper_limit);
         break;
     case journal_type_t::HS_CTRL_REMOVE_MEMBER:
         remove_member(rreq);
@@ -1755,7 +1804,7 @@ void RaftReplDev::clean_replace_member_task(repl_req_ptr_t rreq) {
     RD_LOGI(rreq->traceID(), "Raft repl replace_member_task has been cleared, task_id={}", task_id);
 }
 
-void RaftReplDev::update_truncation_boundary(repl_req_ptr_t rreq) {
+void RaftReplDev::update_truncation_boundary(repl_lsn_t truncation_upper_limit) {
     repl_lsn_t cur_checkpoint_lsn = 0;
     {
         std::unique_lock lg{m_sb_mtx};
@@ -1764,8 +1813,7 @@ void RaftReplDev::update_truncation_boundary(repl_req_ptr_t rreq) {
     // expected truncation_upper_limit should not larger than the current checkpoint_lsn, this is to ensure that
     // when a crash happens before index flushed to disk, all the logs larger than checkpoint_lsn are still available
     // to replay.
-    auto ctx = r_cast< const truncate_ctx* >(rreq->header().cbytes());
-    auto exp_truncation_upper_limit = std::min(ctx->truncation_upper_limit, cur_checkpoint_lsn);
+    auto exp_truncation_upper_limit = std::min(truncation_upper_limit, cur_checkpoint_lsn);
     auto cur_truncation_upper_limit = m_truncation_upper_limit.load();
     // exp_truncation_upper_limit might be less or equal to cur_truncation_upper_limit after Baseline Re-sync,
     // we should skip update to ensure the truncation_upper_limit is always increasing.
@@ -2389,8 +2437,9 @@ void RaftReplDev::cp_flush(CP* cp, cshared< ReplDevCPContext > ctx) {
     auto const clsn = ctx->compacted_to_lsn;
     auto const dsn = ctx->last_applied_dsn;
 
-    if (lsn == m_last_flushed_commit_lsn) {
-        // Not dirtied since last flush ignore
+    // compact can be triggered manually while no new logs are committed, so both lsn and clsn need to be checked
+    if (lsn == m_last_flushed_cp_lsn && clsn == m_last_flushed_compact_lsn) {
+        // Not dirtied since last cp flush ignore
         return;
     }
 
@@ -2402,7 +2451,8 @@ void RaftReplDev::cp_flush(CP* cp, cshared< ReplDevCPContext > ctx) {
     m_rd_sb->checkpoint_lsn = lsn;
     m_rd_sb->last_applied_dsn = dsn;
     m_rd_sb.write();
-    m_last_flushed_commit_lsn = lsn;
+    m_last_flushed_cp_lsn = lsn;
+    m_last_flushed_compact_lsn = clsn;
     RD_LOGD(NO_TRACE_ID, "cp flush in raft repl dev, lsn={}, clsn={}, next_dsn={}, cp string:{}", lsn, clsn,
             m_next_dsn.load(), cp->to_string());
 }

src/lib/replication/repl_dev/raft_repl_dev.h

@@ -203,7 +203,8 @@ class RaftReplDev : public ReplDev,
 
     std::mutex m_sb_mtx; // Lock to protect the repl dev superblock
 
-    repl_lsn_t m_last_flushed_commit_lsn{0}; // LSN upto which it was flushed to persistent store
+    repl_lsn_t m_last_flushed_cp_lsn{0};      // LSN upto which it was flushed to persistent store at the checkpoint
+    repl_lsn_t m_last_flushed_compact_lsn{0}; // LSN upto which it was compacted at the checkpoint
     iomgr::timer_handle_t m_sb_flush_timer_hdl;
 
     std::atomic< uint64_t > m_next_dsn{0}; // Data Sequence Number that will keep incrementing for each data entry
@@ -373,6 +374,24 @@ class RaftReplDev : public ReplDev,
      */
     void on_create_snapshot(nuraft::snapshot& s, nuraft::async_result< bool >::handler_type& when_done);
 
+    /**
+     * \brief  Manually create a snapshot, the snapshot will be created directly based on the latest committed log
+     * index.
+     *
+     * This function is called when we want to create a snapshot and compact the log store manually.
+     * it provides an optional compact lsn for the log store, which is useful in scenarios like
+     * the truncation upper limit is reset due to restart (it is a in-memory param) and the previous auto background
+     * compact was skipped because of the truncation upper limit is zero.
+     *
+     * \param compact_lsn The specific truncation upper limit for the log store.
+     * \param wait_for_commit Whether to wait for the committed lsn to reach the compact_lsn before creating the
+     * snapshot.
+     *
+     * \note This function will trigger cp_flush before creating snapshot and after compaction to make sure the logs was
+     * compacted and change has been persisted, which might take some time.
+     */
+    void trigger_snapshot_creation(repl_lsn_t compact_lsn, bool wait_for_commit) override;
+
 #if 0
     /**
      * Truncates the replication log by providing a specified number of reserved entries.
@@ -475,7 +494,7 @@ class RaftReplDev : public ReplDev,
     void create_snp_resync_data(raft_buf_ptr_t& data_out);
     bool save_snp_resync_data(nuraft::buffer& data, nuraft::snapshot& s);
 
-    void update_truncation_boundary(repl_req_ptr_t rreq);
+    void update_truncation_boundary(repl_lsn_t truncation_upper_limit);
     void propose_truncate_boundary();
 
     void report_blk_metrics_if_needed(repl_req_ptr_t rreq);

src/lib/replication/repl_dev/solo_repl_dev.h

@@ -121,6 +121,8 @@ class SoloReplDev : public ReplDev {
     // solo repl device truncation is triggerred by CP after flushing all the data in its cp_cleanup routine;
     void truncate();
 
+    void trigger_snapshot_creation(repl_lsn_t compact_lsn, bool wait_for_commit) override { return; }
+
 private:
     void write_journal(repl_req_ptr_t rreq);
     void on_log_found(logstore_seq_num_t lsn, log_buffer buf, void* ctx);

src/lib/replication/service/generic_repl_svc.cpp

@@ -233,6 +233,8 @@ ReplServiceError SoloReplService::destroy_repl_dev(group_id_t group_id, uint64_t
     return remove_repl_dev(group_id).get();
 }
 
+void SoloReplService::trigger_snapshot_creation(group_id_t group_id, repl_lsn_t compact_lsn, bool wait_for_commit) {}
+
 std::unique_ptr< CPContext > SoloReplServiceCPHandler::on_switchover_cp(CP* cur_cp, CP* new_cp) {
     return std::make_unique< CPContext >(new_cp);
 }

src/lib/replication/service/generic_repl_svc.h

@@ -112,6 +112,7 @@ class SoloReplService : public GenericReplService {
                                                   const std::vector< replica_member_info >& others,
                                                   uint64_t trace_id = 0) const override;
     ReplServiceError destroy_repl_dev(group_id_t group_id, uint64_t trace_id = 0) override;
+    void trigger_snapshot_creation(group_id_t group_id, repl_lsn_t compact_lsn, bool wait_for_commit) override;
 };
 
 class SoloReplServiceCPHandler : public CPCallbacks {

src/lib/replication/service/raft_repl_service.cpp

@@ -610,6 +610,18 @@ ReplServiceError RaftReplService::destroy_repl_dev(group_id_t group_id, uint64_t
     return ReplServiceError::OK;
 }
 
+void RaftReplService::trigger_snapshot_creation(group_id_t group_id, repl_lsn_t compact_lsn, bool wait_for_commit) {
+    auto rdev_result = get_repl_dev(group_id);
+    if (!rdev_result) {
+        LOGWARNMOD(replication, "ReplDev group_id={} not found while scheduling snapshot creation",
+                   boost::uuids::to_string(group_id));
+        return;
+    }
+
+    std::dynamic_pointer_cast< RaftReplDev >(rdev_result.value())
+        ->trigger_snapshot_creation(compact_lsn, wait_for_commit);
+}
+
 ////////////////////// Reaper Thread related //////////////////////////////////
 void RaftReplService::start_repl_service_timers() {
     // we need to explictly cancel the timers before we stop the repl_devs, but we cannot cancel a thread timer

src/lib/replication/service/raft_repl_service.h

@@ -106,6 +106,8 @@ class RaftReplService : public GenericReplService,
                                                   uint64_t trace_id = 0) const override;
     ReplServiceError destroy_repl_dev(group_id_t group_id, uint64_t trace_id = 0) override;
 
+    void trigger_snapshot_creation(group_id_t group_id, repl_lsn_t compact_lsn, bool wait_for_commit) override;
+
 private:
     RaftReplDev* raft_group_config_found(sisl::byte_view const& buf, void* meta_cookie);
     void start_repl_service_timers();