MB-64256: Add dynamic config for fusionSyncRateLimit

mohammadzaeem · mohammadzaeem · commit ee157d3b455b · 2025-03-07T14:19:54.000Z
Change-Id: I01c895bca0260b3a8bfffe306c3199d95202f416 Reviewed-on: https://review.couchbase.org/c/kv_engine/+/224401 Reviewed-by: Paolo Cocchi <paolo.cocchi@couchbase.com> Reviewed-by: Trond Norbye <trond.norbye@couchbase.com> Tested-by: Build Bot <build@couchbase.com>
diff --git a/engines/ep/configuration.json b/engines/ep/configuration.json
@@ -1675,6 +1675,17 @@
                 }
             }
         },
+        "magma_fusion_sync_rate_limit": {
+            "default": "(1024 * 1024 * 75)",
+            "descr": "The rate limit for Fusion sync uploads, in bytes per second.",
+            "dynamic": true,
+            "type": "size_t",
+            "validator": {
+                "range": {
+                    "min": 1
+                }
+            }
+        },
         "continuous_backup_enabled": {
             "default": "false",
             "descr": "True if continouous backup is enabled.",
diff --git a/engines/ep/management/cbepctl b/engines/ep/management/cbepctl
@@ -299,6 +299,7 @@ Available params for "set":
     history_retention_bytes      - Max bytes of history a bucket should aim to retain on disk.
     fusion_metadata_auth_token   - Authorization token for the Chronical service
     magma_fusion_migration_rate_limit - The rate limit for Fusion extent migration, in bytes per second.
+    magma_fusion_sync_rate_limit - The rate limit for Fusion sync uploads, in bytes per second.
 
   Available params for "set replication_param":
     <empty>
diff --git a/engines/ep/src/ep_engine.cc b/engines/ep/src/ep_engine.cc
@@ -966,10 +966,10 @@ cb::engine_errc EventuallyPersistentEngine::setFlushParam(
             configuration.setContinuousBackupEnabled(cb_stob(val));
         } else if (key == "continuous_backup_interval") {
             configuration.setContinuousBackupInterval(std::stoull(val));
-        } else if (key == "magma_fusion_migration_rate_limit") {
-            configuration.setMagmaFusionMigrationRateLimit(std::stoull(val));
         } else if (key == "fusion_metadata_auth_token") {
             setFusionMetadataAuthToken(val);
+        } else if (key.starts_with("magma_fusion")) {
+            return setFusionFlushParam(key, val, msg);
         } else if (key == "workload_monitor_enabled") {
             configuration.setWorkloadMonitorEnabled(cb_stob(val));
         } else {
@@ -1006,6 +1006,23 @@ cb::engine_errc EventuallyPersistentEngine::setFlushParam(
     return rv;
 }
 
+cb::engine_errc EventuallyPersistentEngine::setFusionFlushParam(
+        std::string_view key, const std::string& val, std::string& msg) {
+    auto rv = cb::engine_errc::success;
+    if (key == "magma_fusion_migration_rate_limit") {
+        configuration.setMagmaFusionMigrationRateLimit(std::stoull(val));
+    } else if (key == "magma_fusion_sync_rate_limit") {
+        configuration.setMagmaFusionSyncRateLimit(std::stoull(val));
+    } else {
+        EP_LOG_WARN_CTX("Rejecting setFusionFlushParam request",
+                        {"key", key},
+                        {"value", val});
+        msg = "Unknown config param " + std::string{key};
+        rv = cb::engine_errc::invalid_arguments;
+    }
+    return rv;
+}
+
 cb::engine_errc EventuallyPersistentEngine::setDcpParam(std::string_view key,
                                                         const std::string& val,
                                                         std::string& msg) {
diff --git a/engines/ep/src/ep_engine.h b/engines/ep/src/ep_engine.h
@@ -725,6 +725,9 @@ class EventuallyPersistentEngine : public EngineIface, public DcpIface {
     cb::engine_errc setFlushParam(std::string_view key,
                                   const std::string& val,
                                   std::string& msg);
+    cb::engine_errc setFusionFlushParam(std::string_view key,
+                                        const std::string& val,
+                                        std::string& msg);
 
     cb::engine_errc setReplicationParam(std::string_view key,
                                         const std::string& val,
diff --git a/engines/ep/src/kvstore/magma-kvstore/magma-kvstore.cc b/engines/ep/src/kvstore/magma-kvstore/magma-kvstore.cc
@@ -900,6 +900,7 @@ MagmaKVStore::MagmaKVStore(MagmaKVStoreConfig& configuration,
                 configuration.getFusionLogCheckpointInterval());
         magma->SetFusionMigrationRateLimit(
                 configuration.getFusionMigrationRateLimit());
+        magma->SetFusionSyncRateLimit(configuration.getFusionSyncRateLimit());
     }
 }
 
@@ -4650,6 +4651,14 @@ size_t MagmaKVStore::getMagmaFusionMigrationRateLimit() const {
     return magma->GetFusionMigrationRateLimit();
 }
 
+void MagmaKVStore::setMagmaFusionSyncRateLimit(size_t value) {
+    magma->SetFusionSyncRateLimit(value);
+}
+
+size_t MagmaKVStore::getMagmaFusionSyncRateLimit() const {
+    return magma->GetFusionSyncRateLimit();
+}
+
 cb::engine_errc MagmaKVStore::startFusionUploader(Vbid vbid, uint64_t term) {
     const auto status =
             magma->StartFusionUploader(Magma::KVStoreID(vbid.get()), term);
diff --git a/engines/ep/src/kvstore/magma-kvstore/magma-kvstore.h b/engines/ep/src/kvstore/magma-kvstore/magma-kvstore.h
@@ -680,6 +680,9 @@ class MagmaKVStore : public KVStore {
     size_t getMagmaFusionMigrationRateLimit() const;
     void setMagmaFusionMigrationRateLimit(size_t value);
 
+    size_t getMagmaFusionSyncRateLimit() const;
+    void setMagmaFusionSyncRateLimit(size_t value);
+
     // Magma uses a unique logger with a prefix of magma so that all logging
     // calls from the wrapper thru magma will be prefixed with magma.
     std::shared_ptr<BucketLogger> logger;
diff --git a/engines/ep/src/kvstore/magma-kvstore/magma-kvstore_config.cc b/engines/ep/src/kvstore/magma-kvstore/magma-kvstore_config.cc
@@ -48,6 +48,8 @@ class MagmaKVStoreConfig::ConfigChangeListener : public ValueChangedListener {
             config.setContinousBackupInterval(std::chrono::seconds(value));
         } else if (key == "magma_fusion_migration_rate_limit") {
             config.setFusionMigrationRateLimit(value);
+        } else if (key == "magma_fusion_sync_rate_limit") {
+            config.setFusionSyncRateLimit(value);
         }
     }
 
@@ -231,6 +233,10 @@ MagmaKVStoreConfig::MagmaKVStoreConfig(Configuration& config,
     config.addValueChangedListener(
             "magma_fusion_migration_rate_limit",
             std::make_unique<ConfigChangeListener>(*this));
+    fusionSyncRateLimit = config.getMagmaFusionSyncRateLimit();
+    config.addValueChangedListener(
+            "magma_fusion_sync_rate_limit",
+            std::make_unique<ConfigChangeListener>(*this));
 }
 
 void MagmaKVStoreConfig::setStore(MagmaKVStore* store) {
@@ -244,6 +250,13 @@ void MagmaKVStoreConfig::setFusionMigrationRateLimit(size_t value) {
     }
 }
 
+void MagmaKVStoreConfig::setFusionSyncRateLimit(size_t value) {
+    fusionSyncRateLimit.store(value);
+    if (store) {
+        store->setMagmaFusionSyncRateLimit(value);
+    }
+}
+
 void MagmaKVStoreConfig::setMagmaFragmentationPercentage(size_t value) {
     magmaFragmentationPercentage.store(value);
     if (store) {
diff --git a/engines/ep/src/kvstore/magma-kvstore/magma-kvstore_config.h b/engines/ep/src/kvstore/magma-kvstore/magma-kvstore_config.h
@@ -305,6 +305,12 @@ class MagmaKVStoreConfig : public KVStoreConfig {
 
     void setFusionMigrationRateLimit(size_t value);
 
+    size_t getFusionSyncRateLimit() const {
+        return fusionSyncRateLimit;
+    }
+
+    void setFusionSyncRateLimit(size_t value);
+
     magma::Magma::Config magmaCfg;
 
     /**
@@ -587,4 +593,6 @@ class MagmaKVStoreConfig : public KVStoreConfig {
     std::chrono::seconds fusionLogCheckpointInterval;
     // The rate limit for Fusion extent migration, in bytes per second
     std::atomic<size_t> fusionMigrationRateLimit;
+    // The rate limit for Fusion sync uploads, in bytes per second
+    std::atomic<size_t> fusionSyncRateLimit;
 };
diff --git a/engines/ep/src/kvstore/magma-kvstore/magma-memory-tracking-proxy.cc b/engines/ep/src/kvstore/magma-kvstore/magma-memory-tracking-proxy.cc
@@ -774,6 +774,16 @@ size_t MagmaMemoryTrackingProxy::GetFusionMigrationRateLimit() const {
     return magma->GetFusionMigrationRateLimit();
 }
 
+void MagmaMemoryTrackingProxy::SetFusionSyncRateLimit(size_t limit) {
+    cb::UseArenaMallocSecondaryDomain domainGuard;
+    magma->SetFusionSyncRateLimit(limit);
+}
+
+size_t MagmaMemoryTrackingProxy::GetFusionSyncRateLimit() const {
+    cb::UseArenaMallocSecondaryDomain sGuard;
+    return magma->GetFusionSyncRateLimit();
+}
+
 magma::Status MagmaMemoryTrackingProxy::StartFusionUploader(
         magma::Magma::KVStoreID kvId, uint64_t term) {
     cb::UseArenaMallocSecondaryDomain d;
diff --git a/engines/ep/src/kvstore/magma-kvstore/magma-memory-tracking-proxy.h b/engines/ep/src/kvstore/magma-kvstore/magma-memory-tracking-proxy.h
@@ -305,6 +305,10 @@ class MagmaMemoryTrackingProxy {
     void SetFusionMigrationRateLimit(size_t limit);
     size_t GetFusionMigrationRateLimit() const;
 
+    // The rate limit for Fusion sync uploads, in bytes per second.
+    void SetFusionSyncRateLimit(size_t limit);
+    size_t GetFusionSyncRateLimit() const;
+
     /**
      * Start uploading data to FusionLogStore for the latest revision of the
      * given kvstore.
diff --git a/engines/ep/tests/ep_testsuite.cc b/engines/ep/tests/ep_testsuite.cc
@@ -6520,6 +6520,7 @@ static enum test_result test_mb19687_fixed(EngineIface* h) {
               "ep_magma_fusion_upload_interval",
               "ep_magma_fusion_log_checkpoint_interval",
               "ep_magma_fusion_migration_rate_limit",
+              "ep_magma_fusion_sync_rate_limit",
               "ep_max_num_bgfetchers",
               "ep_max_num_flushers",
               "ep_max_checkpoints",
@@ -6872,6 +6873,7 @@ static enum test_result test_mb19687_fixed(EngineIface* h) {
               "ep_magma_fusion_upload_interval",
               "ep_magma_fusion_log_checkpoint_interval",
               "ep_magma_fusion_migration_rate_limit",
+              "ep_magma_fusion_sync_rate_limit",
               "ep_oom_errors",
               "ep_overhead",
               "ep_pager_sleep_time_ms",
diff --git a/engines/ep/tests/module_tests/magma_bucket_tests.cc b/engines/ep/tests/module_tests/magma_bucket_tests.cc
@@ -1170,6 +1170,8 @@ class STMagmaFusionTest : public STParamMagmaBucketTest {
                          std::to_string(fusionLogCheckpointInterval);
         config_string += ";magma_fusion_migration_rate_limit=" +
                          std::to_string(fusionMigrationRateLimit);
+        config_string += ";magma_fusion_sync_rate_limit=" +
+                         std::to_string(fusionSyncRateLimit);
         STParameterizedBucketTest::SetUp();
     }
 
@@ -1181,6 +1183,7 @@ class STMagmaFusionTest : public STParamMagmaBucketTest {
     const size_t fusionUploadInterval = 1234;
     const size_t fusionLogCheckpointInterval = 5678;
     const size_t fusionMigrationRateLimit = 9101112;
+    const size_t fusionSyncRateLimit = 13141516;
 };
 
 TEST_P(STMagmaFusionTest, Config) {
@@ -1204,6 +1207,7 @@ TEST_P(STMagmaFusionTest, Config) {
               kvstore.getFusionLogCheckpointInterval().count());
     EXPECT_EQ(fusionMigrationRateLimit,
               kvstore.getMagmaFusionMigrationRateLimit());
+    EXPECT_EQ(fusionSyncRateLimit, kvstore.getMagmaFusionSyncRateLimit());
 }
 
 TEST_P(STMagmaFusionTest, MagmaFusionMigrationRateLimit) {
@@ -1231,6 +1235,29 @@ TEST_P(STMagmaFusionTest, MagmaFusionMigrationRateLimit) {
             << "value should not have been updated";
 }
 
+TEST_P(STMagmaFusionTest, MagmaFusionSyncRateLimit) {
+    std::string msg;
+    ASSERT_EQ(cb::engine_errc::success,
+              engine->setFlushParam(
+                      "magma_fusion_sync_rate_limit", "777777", msg));
+
+    auto& kvstore = dynamic_cast<MagmaKVStore&>(*store->getRWUnderlying(vbid));
+    auto& config = dynamic_cast<const MagmaKVStoreConfig&>(kvstore.getConfig());
+    EXPECT_EQ(777777, config.getFusionSyncRateLimit()) << "config not updated";
+    EXPECT_EQ(777777, kvstore.getMagmaFusionSyncRateLimit())
+            << "value not passed down to Magma";
+
+    // Test with an invalid value: 0
+    ASSERT_EQ(cb::engine_errc::invalid_arguments,
+              engine->setFlushParam("magma_fusion_sync_rate_limit", "0", msg));
+
+    // Verify the values are not updated
+    EXPECT_EQ(777777, config.getFusionSyncRateLimit())
+            << "config should not have been updated";
+    EXPECT_EQ(777777, kvstore.getMagmaFusionSyncRateLimit())
+            << "value should not have been updated";
+}
+
 TEST_P(STMagmaFusionTest, MetadataAuthToken) {
     auto& kvstore = dynamic_cast<MagmaKVStore&>(*store->getRWUnderlying(vbid));
     ASSERT_TRUE(kvstore.getFusionMetadataAuthToken().empty());

Original file line number	Diff line number	Diff line change
`@@ -48,6 +48,8 @@ class MagmaKVStoreConfig::ConfigChangeListener : public ValueChangedListener {`
`48`	`48`	`config.setContinousBackupInterval(std::chrono::seconds(value));`
`49`	`49`	`} else if (key == "magma_fusion_migration_rate_limit") {`
`50`	`50`	`config.setFusionMigrationRateLimit(value);`
	`51`	`+ } else if (key == "magma_fusion_sync_rate_limit") {`
	`52`	`+ config.setFusionSyncRateLimit(value);`
`51`	`53`	`}`
`52`	`54`	`}`
`53`	`55`
`@@ -231,6 +233,10 @@ MagmaKVStoreConfig::MagmaKVStoreConfig(Configuration& config,`
`231`	`233`	`config.addValueChangedListener(`
`232`	`234`	`"magma_fusion_migration_rate_limit",`
`233`	`235`	`std::make_unique<ConfigChangeListener>(*this));`
	`236`	`+ fusionSyncRateLimit = config.getMagmaFusionSyncRateLimit();`
	`237`	`+ config.addValueChangedListener(`
	`238`	`+ "magma_fusion_sync_rate_limit",`
	`239`	`+ std::make_unique<ConfigChangeListener>(*this));`
`234`	`240`	`}`
`235`	`241`
`236`	`242`	`void MagmaKVStoreConfig::setStore(MagmaKVStore* store) {`
`@@ -244,6 +250,13 @@ void MagmaKVStoreConfig::setFusionMigrationRateLimit(size_t value) {`
`244`	`250`	`}`
`245`	`251`	`}`
`246`	`252`
	`253`	`+void MagmaKVStoreConfig::setFusionSyncRateLimit(size_t value) {`
	`254`	`+ fusionSyncRateLimit.store(value);`
	`255`	`+ if (store) {`
	`256`	`+ store->setMagmaFusionSyncRateLimit(value);`
	`257`	`+ }`
	`258`	`+}`
	`259`	`+`
`247`	`260`	`void MagmaKVStoreConfig::setMagmaFragmentationPercentage(size_t value) {`
`248`	`261`	`magmaFragmentationPercentage.store(value);`
`249`	`262`	`if (store) {`