Merge pull request #11704 from vespa-engine/vekterli/support-config-disabling-of-merges

Support config disabling of merges

7 files changed, 57 insertions, 7 deletions

diff --git a/storage/src/tests/distributor/distributortest.cpp b/storage/src/tests/distributor/distributortest.cpp
index d456401876e..83ef7891630 100644
--- a/storage/src/tests/distributor/distributortest.cpp
+++ b/storage/src/tests/distributor/distributortest.cpp
@@ -181,6 +181,12 @@ struct DistributorTest : Test, DistributorTestUtil {
         configureDistributor(builder);
     }
 
+    void configure_merge_operations_disabled(bool disabled) {
+        ConfigBuilder builder;
+        builder.mergeOperationsDisabled = disabled;
+        configureDistributor(builder);
+    }
+
     void configureMaxClusterClockSkew(int seconds);
     void sendDownClusterStateCommand();
     void replyToSingleRequestBucketInfoCommandWith1Bucket();
@@ -1021,6 +1027,17 @@ TEST_F(DistributorTest, fast_path_on_consistent_gets_config_is_propagated_to_int
     EXPECT_FALSE(getConfig().update_fast_path_restart_enabled());
 }
 
+TEST_F(DistributorTest, merge_disabling_config_is_propagated_to_internal_config) {
+    createLinks(true);
+    setupDistributor(Redundancy(1), NodeCount(1), "distributor:1 storage:1");
+
+    configure_merge_operations_disabled(true);
+    EXPECT_TRUE(getConfig().merge_operations_disabled());
+
+    configure_merge_operations_disabled(false);
+    EXPECT_FALSE(getConfig().merge_operations_disabled());
+}
+
 TEST_F(DistributorTest, concurrent_reads_not_enabled_if_btree_db_is_not_enabled) {
     createLinks(false);
     setupDistributor(Redundancy(1), NodeCount(1), "distributor:1 storage:1");
diff --git a/storage/src/tests/distributor/statecheckerstest.cpp b/storage/src/tests/distributor/statecheckerstest.cpp
index 01c4ad1cf6a..fa7afd39cf3 100644
--- a/storage/src/tests/distributor/statecheckerstest.cpp
+++ b/storage/src/tests/distributor/statecheckerstest.cpp
@@ -168,6 +168,7 @@ struct StateCheckersTest : Test, DistributorTestUtil {
         uint32_t _minSplitBits {0};
         bool _includeMessagePriority {false};
         bool _includeSchedulingPriority {false};
+        bool _merge_operations_disabled {false};
         CheckerParams();
         ~CheckerParams();
 
@@ -203,6 +204,10 @@ struct StateCheckersTest : Test, DistributorTestUtil {
             _includeSchedulingPriority = includePri;
             return *this;
         }
+        CheckerParams& merge_operations_disabled(bool disabled) noexcept {
+            _merge_operations_disabled = disabled;
+            return *this;
+        }
     };
 
     template <typename CheckerImpl>
@@ -213,6 +218,7 @@ struct StateCheckersTest : Test, DistributorTestUtil {
         addNodesToBucketDB(bid, params._bucketInfo);
         setRedundancy(params._redundancy);
         enableDistributorClusterState(params._clusterState);
+        getConfig().set_merge_operations_disabled(params._merge_operations_disabled);
         if (!params._pending_cluster_state.empty()) {
             auto cmd = std::make_shared<api::SetSystemStateCommand>(lib::ClusterState(params._pending_cluster_state));
             _distributor->onDown(cmd);
@@ -817,6 +823,15 @@ TEST_F(StateCheckersTest, retired_nodes_out_of_sync_are_merged) {
                           ".0.s:r .1.s:r .2.s:r .3.s:r"));
 }
 
+TEST_F(StateCheckersTest, no_merge_operation_generated_if_merges_explicitly_config_disabled) {
+    runAndVerify<SynchronizeAndMoveStateChecker>(
+        CheckerParams()
+            .expect("NO OPERATIONS GENERATED") // Would normally generate a merge op
+            .bucketInfo("0=1,2=2")
+            .clusterState("distributor:1 storage:3")
+            .merge_operations_disabled(true));
+}
+
 std::string
 StateCheckersTest::testDeleteExtraCopies(
         const std::string& bucketInfo, uint32_t redundancy,
diff --git a/storage/src/vespa/storage/config/distributorconfiguration.cpp b/storage/src/vespa/storage/config/distributorconfiguration.cpp
index 0b8564e561a..ed14e227fc1 100644
--- a/storage/src/vespa/storage/config/distributorconfiguration.cpp
+++ b/storage/src/vespa/storage/config/distributorconfiguration.cpp
@@ -40,6 +40,7 @@ DistributorConfiguration::DistributorConfiguration(StorageComponent& component)
       _sequenceMutatingOperations(true),
       _allowStaleReadsDuringClusterStateTransitions(false),
       _update_fast_path_restart_enabled(false),
+      _merge_operations_disabled(false),
       _minimumReplicaCountingMode(ReplicaCountingMode::TRUSTED)
 { }
 
@@ -151,6 +152,7 @@ DistributorConfiguration::configure(const vespa::config::content::core::StorDist
     _sequenceMutatingOperations = config.sequenceMutatingOperations;
     _allowStaleReadsDuringClusterStateTransitions = config.allowStaleReadsDuringClusterStateTransitions;
     _update_fast_path_restart_enabled = config.restartWithFastUpdatePathIfAllGetTimestampsAreConsistent;
+    _merge_operations_disabled = config.mergeOperationsDisabled;
 
     _minimumReplicaCountingMode = config.minimumReplicaCountingMode;
 
diff --git a/storage/src/vespa/storage/config/distributorconfiguration.h b/storage/src/vespa/storage/config/distributorconfiguration.h
index 51ac7f8dae0..5b01e37992b 100644
--- a/storage/src/vespa/storage/config/distributorconfiguration.h
+++ b/storage/src/vespa/storage/config/distributorconfiguration.h
@@ -221,6 +221,13 @@ public:
         _update_fast_path_restart_enabled = enabled;
     }
 
+    bool merge_operations_disabled() const noexcept {
+        return _merge_operations_disabled;
+    }
+    void set_merge_operations_disabled(bool disabled) noexcept {
+        _merge_operations_disabled = disabled;
+    }
+
     bool containsTimeStatement(const std::string& documentSelection) const;
     
 private:
@@ -265,6 +272,7 @@ private:
     bool _sequenceMutatingOperations;
     bool _allowStaleReadsDuringClusterStateTransitions;
     bool _update_fast_path_restart_enabled;
+    bool _merge_operations_disabled;
 
     DistrConfig::MinimumReplicaCountingMode _minimumReplicaCountingMode;
     
diff --git a/storage/src/vespa/storage/config/stor-distributormanager.def b/storage/src/vespa/storage/config/stor-distributormanager.def
index e4a002ae81f..4587e7b3ebe 100644
--- a/storage/src/vespa/storage/config/stor-distributormanager.def
+++ b/storage/src/vespa/storage/config/stor-distributormanager.def
@@ -214,3 +214,8 @@ use_btree_database bool default=false restart
 ## Since all replicas of the document were in sync, applying the update in-place
 ## shall be considered safe.
 restart_with_fast_update_path_if_all_get_timestamps_are_consistent bool default=false
+
+## If set, no merge operations may be generated for any reason by a distributor.
+## This is ONLY intended for system testing of certain transient edge cases and
+## MUST NOT be set to true in a production environment.
+merge_operations_disabled bool default=false
diff --git a/storage/src/vespa/storage/distributor/operations/external/updateoperation.cpp b/storage/src/vespa/storage/distributor/operations/external/updateoperation.cpp
index 6f82ceaab92..771bd90b247 100644
--- a/storage/src/vespa/storage/distributor/operations/external/updateoperation.cpp
+++ b/storage/src/vespa/storage/distributor/operations/external/updateoperation.cpp
@@ -149,13 +149,13 @@ UpdateOperation::onReceive(DistributorMessageSender& sender,
 
                 for (uint32_t i = 0; i < _results.size(); i++) {
                     if (_results[i].oldTs < oldTs) {
-                        LOG(warning, "Update operation for '%s' in bucket %s updated documents with different timestamps. "
-                                     "This should not happen and may indicate undetected replica divergence. "
-                                     "Found ts=%" PRIu64 " on node %u, ts=%" PRIu64 " on node %u",
-                                     reply.getDocumentId().toString().c_str(),
-                                     reply.getBucket().toString().c_str(),
-                                     _results[i].oldTs, _results[i].nodeId,
-                                     _results[goodNode].oldTs, _results[goodNode].nodeId);
+                        LOG(error, "Update operation for '%s' in bucket %s updated documents with different timestamps. "
+                                   "This should not happen and may indicate undetected replica divergence. "
+                                   "Found ts=%" PRIu64 " on node %u, ts=%" PRIu64 " on node %u",
+                                   reply.getDocumentId().toString().c_str(),
+                                   reply.getBucket().toString().c_str(),
+                                   _results[i].oldTs, _results[i].nodeId,
+                                   _results[goodNode].oldTs, _results[goodNode].nodeId);
                         _metrics.diverging_timestamp_updates.inc();
 
                         replyToSend.setNodeWithNewestTimestamp(_results[goodNode].nodeId);
diff --git a/storage/src/vespa/storage/distributor/statecheckers.cpp b/storage/src/vespa/storage/distributor/statecheckers.cpp
index b1fa3056cb1..2da4dd529c5 100644
--- a/storage/src/vespa/storage/distributor/statecheckers.cpp
+++ b/storage/src/vespa/storage/distributor/statecheckers.cpp
@@ -835,6 +835,9 @@ allCopiesAreInvalid(const StateChecker::Context& c)
 StateChecker::Result
 SynchronizeAndMoveStateChecker::check(StateChecker::Context& c)
 {
+    if (c.distributorConfig.merge_operations_disabled()) {
+        return Result::noMaintenanceNeeded();
+    }
     if (isInconsistentlySplit(c)) {
         return Result::noMaintenanceNeeded();
     }