Merge pull request #16508 from vespa-engine/balder/control-bucket-move-job

Balder/control bucket move job

6 files changed, 75 insertions, 1 deletions

diff --git a/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java b/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java
index 50676577d3d..42ecf7e6bf3 100644
--- a/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java
+++ b/config-model-api/src/main/java/com/yahoo/config/model/api/ModelContext.java
@@ -72,6 +72,7 @@ public interface ModelContext {
         @ModelFeatureFlag(owners = {"baldersheim"}, comment = "Select sequencer type use while feeding") default String feedSequencerType() { throw new UnsupportedOperationException("TODO specify default value"); }
         @ModelFeatureFlag(owners = {"baldersheim"}) default String responseSequencerType() { throw new UnsupportedOperationException("TODO specify default value"); }
         @ModelFeatureFlag(owners = {"baldersheim"}) default int defaultNumResponseThreads() { return 2; }
+        @ModelFeatureFlag(owners = {"baldersheim"}) default int maxPendingMoveOps() { throw new UnsupportedOperationException("TODO specify default value"); }
         @ModelFeatureFlag(owners = {"baldersheim"}) default boolean skipCommunicationManagerThread() { throw new UnsupportedOperationException("TODO specify default value"); }
         @ModelFeatureFlag(owners = {"baldersheim"}) default boolean skipMbusRequestThread() { throw new UnsupportedOperationException("TODO specify default value"); }
         @ModelFeatureFlag(owners = {"baldersheim"}) default boolean skipMbusReplyThread() { throw new UnsupportedOperationException("TODO specify default value"); }
@@ -79,6 +80,7 @@ public interface ModelContext {
         @ModelFeatureFlag(owners = {"baldersheim"}) default boolean useAsyncMessageHandlingOnSchedule() { throw new UnsupportedOperationException("TODO specify default value"); }
         @ModelFeatureFlag(owners = {"baldersheim"}) default double feedConcurrency() { throw new UnsupportedOperationException("TODO specify default value"); }
         @ModelFeatureFlag(owners = {"baldersheim"}) default boolean useBucketExecutorForLidSpaceCompact() { throw new UnsupportedOperationException("TODO specify default value"); }
+        @ModelFeatureFlag(owners = {"baldersheim"}) default boolean useBucketExecutorForBucketMove() { throw new UnsupportedOperationException("TODO specify default value"); }
         @ModelFeatureFlag(owners = {"musum", "mpolden"}, comment = "Revisit in February 2021") default boolean reconfigurableZookeeperServer() { return false; }
         @ModelFeatureFlag(owners = {"bjorncs", "tokle"}, removeAfter = "7.357") default boolean enableJdiscConnectionLog() { return true; }
         @ModelFeatureFlag(owners = {"bjorncs", "tokle", "baldersheim"}, removeAfter = "7.357") default boolean enableZstdCompressionAccessLog() { return true; }
diff --git a/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java b/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java
index 397b11661e5..d0e8afa5616 100644
--- a/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java
+++ b/config-model/src/main/java/com/yahoo/config/model/deploy/TestProperties.java
@@ -41,6 +41,7 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
     private String sequencerType = "LATENCY";
     private String responseSequencerType = "ADAPTIVE";
     private int responseNumThreads = 2;
+    private int maxPendingMoveOps = 10;
     private Optional<EndpointCertificateSecrets> endpointCertificateSecrets = Optional.empty();
     private AthenzDomain athenzDomain;
     private ApplicationRoles applicationRoles;
@@ -51,6 +52,7 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
     private boolean enableAutomaticReindexing = false;
     private boolean reconfigurableZookeeperServer = false;
     private boolean useBucketExecutorForLidSpaceCompact;
+    private boolean useBucketExecutorForBucketMove;
     private boolean enableFeedBlockInDistributor = false;
     private double maxDeadBytesRatio = 0.2;
     private int clusterControllerMaxHeapSizeInMb = 512;
@@ -77,6 +79,7 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
     @Override public Optional<ApplicationRoles> applicationRoles() { return Optional.ofNullable(applicationRoles); }
     @Override public String responseSequencerType() { return responseSequencerType; }
     @Override public int defaultNumResponseThreads() { return responseNumThreads; }
+    @Override public int maxPendingMoveOps() { return maxPendingMoveOps; }
     @Override public boolean skipCommunicationManagerThread() { return false; }
     @Override public boolean skipMbusRequestThread() { return false; }
     @Override public boolean skipMbusReplyThread() { return false; }
@@ -87,6 +90,7 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
     @Override public boolean enableAutomaticReindexing() { return enableAutomaticReindexing; }
     @Override public boolean reconfigurableZookeeperServer() { return reconfigurableZookeeperServer; }
     @Override public boolean useBucketExecutorForLidSpaceCompact() { return useBucketExecutorForLidSpaceCompact; }
+    @Override public boolean useBucketExecutorForBucketMove() { return useBucketExecutorForBucketMove; }
     @Override public boolean enableFeedBlockInDistributor() { return enableFeedBlockInDistributor; }
     @Override public double maxDeadBytesRatio() { return maxDeadBytesRatio; }
     @Override public int clusterControllerMaxHeapSizeInMb() { return clusterControllerMaxHeapSizeInMb; }
@@ -117,6 +121,10 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
         responseNumThreads = numThreads;
         return this;
     }
+    public TestProperties setMaxPendingMoveOps(int moveOps) {
+        maxPendingMoveOps = moveOps;
+        return this;
+    }
     public TestProperties setDefaultTermwiseLimit(double limit) {
         defaultTermwiseLimit = limit;
         return this;
@@ -194,6 +202,11 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
         return this;
     }
 
+    public TestProperties useBucketExecutorForBucketMove(boolean enabled) {
+        useBucketExecutorForBucketMove = enabled;
+        return this;
+    }
+
     public TestProperties enableFeedBlockInDistributor(boolean enabled) {
         enableFeedBlockInDistributor = enabled;
         return this;
diff --git a/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java b/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
index 0fdd39e70cf..43f0f0f450d 100644
--- a/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
+++ b/config-model/src/main/java/com/yahoo/vespa/model/content/ContentSearchCluster.java
@@ -12,7 +12,6 @@ import com.yahoo.vespa.model.builder.xml.dom.DomSearchTuningBuilder;
 import com.yahoo.vespa.model.builder.xml.dom.ModelElement;
 import com.yahoo.vespa.model.builder.xml.dom.VespaDomBuilder;
 import com.yahoo.vespa.model.content.cluster.ContentCluster;
-import com.yahoo.vespa.model.content.cluster.DomResourceLimitsBuilder;
 import com.yahoo.vespa.model.search.AbstractSearchCluster;
 import com.yahoo.vespa.model.search.IndexedSearchCluster;
 import com.yahoo.vespa.model.search.NamedSchema;
@@ -64,8 +63,10 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
     private final Map<StorageGroup, NodeSpec> groupToSpecMap = new LinkedHashMap<>();
     private Optional<ResourceLimits> resourceLimits = Optional.empty();
     private final ProtonConfig.Indexing.Optimize.Enum feedSequencerType;
+    private final int maxPendingMoveOps;
     private final double defaultFeedConcurrency;
     private final boolean useBucketExecutorForLidSpaceCompact;
+    private final boolean useBucketExecutorForBucketMove;
     private final double defaultMaxDeadBytesRatio;
 
     /** Whether the nodes of this cluster also hosts a container cluster in a hosted system */
@@ -204,9 +205,11 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
         this.globallyDistributedDocuments = globallyDistributedDocuments;
         this.flushOnShutdown = flushOnShutdown;
         this.combined = combined;
+        maxPendingMoveOps = featureFlags.maxPendingMoveOps();
         feedSequencerType = convertFeedSequencerType(featureFlags.feedSequencerType());
         defaultFeedConcurrency = featureFlags.feedConcurrency();
         useBucketExecutorForLidSpaceCompact = featureFlags.useBucketExecutorForLidSpaceCompact();
+        useBucketExecutorForBucketMove = featureFlags.useBucketExecutorForBucketMove();
         defaultMaxDeadBytesRatio = featureFlags.maxDeadBytesRatio();
     }
 
@@ -429,7 +432,9 @@ public class ContentSearchCluster extends AbstractConfigProducer<SearchCluster>
         } else {
             builder.indexing.optimize(feedSequencerType);
         }
+        builder.maintenancejobs.maxoutstandingmoveops(maxPendingMoveOps);
         builder.lidspacecompaction.usebucketexecutor(useBucketExecutorForLidSpaceCompact);
+        builder.bucketmove.usebucketexecutor(useBucketExecutorForBucketMove);
     }
 
     private boolean isGloballyDistributed(NewDocumentType docType) {
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/builder/xml/dom/ContentBuilderTest.java b/config-model/src/test/java/com/yahoo/vespa/model/builder/xml/dom/ContentBuilderTest.java
index 55234a3a96e..3a3dde0cf87 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/builder/xml/dom/ContentBuilderTest.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/builder/xml/dom/ContentBuilderTest.java
@@ -835,6 +835,34 @@ public class ContentBuilderTest extends DomBuilderTest {
         assertEquals(flag, config.lidspacecompaction().usebucketexecutor());
     }
 
+    private void verifyThatFeatureFlagControlsUseBucketExecutorForBucketMove(boolean flag) {
+        DeployState.Builder deployStateBuilder = new DeployState.Builder().properties(new TestProperties().useBucketExecutorForBucketMove(flag));
+        VespaModel model = new VespaModelCreatorWithMockPkg(new MockApplicationPackage.Builder()
+                .withServices(singleNodeContentXml())
+                .withSearchDefinition(MockApplicationPackage.MUSIC_SEARCHDEFINITION)
+                .build())
+                .create(deployStateBuilder);
+        ProtonConfig config = getProtonConfig(model.getContentClusters().values().iterator().next());
+        assertEquals(flag, config.bucketmove().usebucketexecutor());
+    }
+
+    private void verifyThatFeatureFlagControlsMaxpendingMoveOps(int moveOps) {
+        DeployState.Builder deployStateBuilder = new DeployState.Builder().properties(new TestProperties().setMaxPendingMoveOps(moveOps));
+        VespaModel model = new VespaModelCreatorWithMockPkg(new MockApplicationPackage.Builder()
+                .withServices(singleNodeContentXml())
+                .withSearchDefinition(MockApplicationPackage.MUSIC_SEARCHDEFINITION)
+                .build())
+                .create(deployStateBuilder);
+        ProtonConfig config = getProtonConfig(model.getContentClusters().values().iterator().next());
+        assertEquals(moveOps, config.maintenancejobs().maxoutstandingmoveops());
+    }
+
+    @Test
+    public void verifyMaxPendingMoveOps() {
+        verifyThatFeatureFlagControlsMaxpendingMoveOps(13);
+        verifyThatFeatureFlagControlsMaxpendingMoveOps(107);
+    }
+
     @Test
     public void verifyUseBucketExecutorForLidSpaceCompact() {
         verifyThatFeatureFlagControlsUseBucketExecutorForLidSpaceCompact(true);
@@ -842,6 +870,12 @@ public class ContentBuilderTest extends DomBuilderTest {
     }
 
     @Test
+    public void verifyUseBucketExecutorForBucketMove() {
+        verifyThatFeatureFlagControlsUseBucketExecutorForBucketMove(true);
+        verifyThatFeatureFlagControlsUseBucketExecutorForBucketMove(false);
+    }
+
+    @Test
     public void failWhenNoDocumentsElementSpecified() {
         expectedException.expect(IllegalArgumentException.class);
         expectedException.expectMessage("The specified content engine requires the <documents> element to be specified.");
diff --git a/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java b/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java
index 490ceb48c08..4dd384bf649 100644
--- a/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java
+++ b/configserver/src/main/java/com/yahoo/vespa/config/server/deploy/ModelContextImpl.java
@@ -152,6 +152,7 @@ public class ModelContextImpl implements ModelContext {
         private final String feedSequencer;
         private final String responseSequencer;
         private final int numResponseThreads;
+        private final int maxPendingMoveOps;
         private final boolean skipCommunicationManagerThread;
         private final boolean skipMbusRequestThread;
         private final boolean skipMbusReplyThread;
@@ -160,6 +161,7 @@ public class ModelContextImpl implements ModelContext {
         private final double feedConcurrency;
         private final boolean reconfigurableZookeeperServer;
         private final boolean useBucketExecutorForLidSpaceCompact;
+        private final boolean useBucketExecutorForBucketMove;
         private final boolean enableFeedBlockInDistributor;
         private final double maxDeadBytesRatio;
         private final int clusterControllerMaxHeapSizeInMb;
@@ -171,6 +173,7 @@ public class ModelContextImpl implements ModelContext {
             this.feedSequencer = flagValue(source, appId, Flags.FEED_SEQUENCER_TYPE);
             this.responseSequencer = flagValue(source, appId, Flags.RESPONSE_SEQUENCER_TYPE);
             this.numResponseThreads = flagValue(source, appId, Flags.RESPONSE_NUM_THREADS);
+            this.maxPendingMoveOps = flagValue(source, appId, Flags.MAX_PENDING_MOVE_OPS);
             this.skipCommunicationManagerThread = flagValue(source, appId, Flags.SKIP_COMMUNICATIONMANAGER_THREAD);
             this.skipMbusRequestThread = flagValue(source, appId, Flags.SKIP_MBUS_REQUEST_THREAD);
             this.skipMbusReplyThread = flagValue(source, appId, Flags.SKIP_MBUS_REPLY_THREAD);
@@ -179,6 +182,7 @@ public class ModelContextImpl implements ModelContext {
             this.feedConcurrency = flagValue(source, appId, Flags.FEED_CONCURRENCY);
             this.reconfigurableZookeeperServer = flagValue(source, appId, Flags.RECONFIGURABLE_ZOOKEEPER_SERVER_FOR_CLUSTER_CONTROLLER);
             this.useBucketExecutorForLidSpaceCompact = flagValue(source, appId, Flags.USE_BUCKET_EXECUTOR_FOR_LID_SPACE_COMPACT);
+            this.useBucketExecutorForBucketMove = flagValue(source, appId, Flags.USE_BUCKET_EXECUTOR_FOR_BUCKET_MOVE);
             this.enableFeedBlockInDistributor = flagValue(source, appId, Flags.ENABLE_FEED_BLOCK_IN_DISTRIBUTOR);
             this.maxDeadBytesRatio = flagValue(source, appId, Flags.MAX_DEAD_BYTES_RATIO);
             this.clusterControllerMaxHeapSizeInMb = flagValue(source, appId, Flags.CLUSTER_CONTROLLER_MAX_HEAP_SIZE_IN_MB);
@@ -190,6 +194,7 @@ public class ModelContextImpl implements ModelContext {
         @Override public String feedSequencerType() { return feedSequencer; }
         @Override public String responseSequencerType() { return responseSequencer; }
         @Override public int defaultNumResponseThreads() { return numResponseThreads; }
+        @Override public int maxPendingMoveOps() { return maxPendingMoveOps; }
         @Override public boolean skipCommunicationManagerThread() { return skipCommunicationManagerThread; }
         @Override public boolean skipMbusRequestThread() { return skipMbusRequestThread; }
         @Override public boolean skipMbusReplyThread() { return skipMbusReplyThread; }
@@ -198,6 +203,7 @@ public class ModelContextImpl implements ModelContext {
         @Override public double feedConcurrency() { return feedConcurrency; }
         @Override public boolean reconfigurableZookeeperServer() { return reconfigurableZookeeperServer; }
         @Override public boolean useBucketExecutorForLidSpaceCompact() { return useBucketExecutorForLidSpaceCompact; }
+        @Override public boolean useBucketExecutorForBucketMove() { return useBucketExecutorForBucketMove; }
         @Override public boolean enableFeedBlockInDistributor() { return enableFeedBlockInDistributor; }
         @Override public double maxDeadBytesRatio() { return maxDeadBytesRatio; }
         @Override public int clusterControllerMaxHeapSizeInMb() { return clusterControllerMaxHeapSizeInMb; }
diff --git a/flags/src/main/java/com/yahoo/vespa/flags/Flags.java b/flags/src/main/java/com/yahoo/vespa/flags/Flags.java
index 8d7fc494dcd..7d3cb11718a 100644
--- a/flags/src/main/java/com/yahoo/vespa/flags/Flags.java
+++ b/flags/src/main/java/com/yahoo/vespa/flags/Flags.java
@@ -165,6 +165,13 @@ public class Flags {
             "Takes effect at redeployment",
             ZONE_ID, APPLICATION_ID);
 
+    public static final UnboundIntFlag MAX_PENDING_MOVE_OPS = defineIntFlag(
+            "max-pending-move-ops", 10,
+            List.of("baldersheim"), "2021-02-15", "2021-04-01",
+            "Max number of move operations inflight",
+            "Takes effect at redeployment",
+            ZONE_ID, APPLICATION_ID);
+
     public static final UnboundDoubleFlag FEED_CONCURRENCY = defineDoubleFlag(
             "feed-concurrency", 0.5,
             List.of("baldersheim"), "2020-12-02", "2022-01-01",
@@ -179,6 +186,13 @@ public class Flags {
             "Takes effect on next internal redeployment",
             APPLICATION_ID);
 
+    public static final UnboundBooleanFlag USE_BUCKET_EXECUTOR_FOR_BUCKET_MOVE = defineFeatureFlag(
+            "use-bucket-executor-for-bucket-move", false,
+            List.of("baldersheim"), "2021-02-15", "2021-04-01",
+            "Wheter to use content-level bucket executor or legacy frozen buckets",
+            "Takes effect on next internal redeployment",
+            APPLICATION_ID);
+
     public static final UnboundBooleanFlag USE_POWER_OF_TWO_CHOICES_LOAD_BALANCING = defineFeatureFlag(
             "use-power-of-two-choices-load-balancing", false,
             List.of("tokle"), "2020-12-02", "2021-02-15",