Merge pull request #15326 from vespa-engine/balder/add-featureflag-for-feed.concurrency

Add a feature flag for controlling the default feed.concurrency setting.

7 files changed, 48 insertions, 5 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 30e1126ca43..7c1f92a0a93 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
@@ -102,6 +102,7 @@ public interface ModelContext {
         boolean useAsyncMessageHandlingOnSchedule();
         int contentNodeBucketDBStripeBits();
         int mergeChunkSize();
+        double feedConcurrency();
 
         // TODO(balder) Last used on 7.306
         default boolean useContentNodeBtreeDb() { 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 664ed896ad6..0e4d0db484e 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
@@ -51,6 +51,7 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
     private boolean useAsyncMessageHandlingOnSchedule = false;
     private int contentNodeBucketDBStripeBits = 0;
     private int mergeChunkSize = 0x400000 - 0x1000; // 4M -4k
+    private double feedConcurrency = 0.5;
 
     @Override public ModelContext.FeatureFlags featureFlags() { return this; }
     @Override public boolean multitenant() { return multitenant; }
@@ -84,6 +85,12 @@ public class TestProperties implements ModelContext.Properties, ModelContext.Fea
     @Override public boolean useAsyncMessageHandlingOnSchedule() { return useAsyncMessageHandlingOnSchedule; }
     @Override public int contentNodeBucketDBStripeBits() { return contentNodeBucketDBStripeBits; }
     @Override public int mergeChunkSize() { return mergeChunkSize; }
+    @Override public double feedConcurrency() { return feedConcurrency; }
+
+    public TestProperties setFeedConcurrency(double feedConcurrency) {
+        this.feedConcurrency = feedConcurrency;
+        return this;
+    }
 
     public TestProperties setMergeChunkSize(int size) {
         mergeChunkSize = size;
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 1f5646ebabb..948093f6c77 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
@@ -64,6 +64,7 @@ public class ContentSearchCluster extends AbstractConfigProducer implements Prot
     private final Map<StorageGroup, NodeSpec> groupToSpecMap = new LinkedHashMap<>();
     private Optional<ResourceLimits> resourceLimits = Optional.empty();
     private final ProtonConfig.Indexing.Optimize.Enum feedSequencerType;
+    private double defaultFeedConcurrency;
 
     /** Whether the nodes of this cluster also hosts a container cluster in a hosted system */
     private final boolean combined;
@@ -203,6 +204,7 @@ public class ContentSearchCluster extends AbstractConfigProducer implements Prot
         this.flushOnShutdown = flushOnShutdown;
         this.combined = combined;
         feedSequencerType = convertFeedSequencerType(featureFlags.feedSequencerType());
+        defaultFeedConcurrency = featureFlags.feedConcurrency();
     }
 
     public void setVisibilityDelay(double delay) {
@@ -364,7 +366,7 @@ public class ContentSearchCluster extends AbstractConfigProducer implements Prot
 
     @Override
     public void getConfig(ProtonConfig.Builder builder) {
-        builder.feeding.concurrency(0.50); // As if specified 1.0 in services.xml
+        builder.feeding.concurrency(defaultFeedConcurrency); // As if specified 1.0 in services.xml
         boolean hasAnyNonIndexedCluster = false;
         for (NewDocumentType type : TopologicalDocumentTypeSorter.sort(documentDefinitions.values())) {
             ProtonConfig.Documentdb.Builder ddbB = new ProtonConfig.Documentdb.Builder();
diff --git a/config-model/src/test/java/com/yahoo/vespa/model/search/test/DocumentDatabaseTestCase.java b/config-model/src/test/java/com/yahoo/vespa/model/search/test/DocumentDatabaseTestCase.java
index 4005418f8dc..a7131b6b3a4 100644
--- a/config-model/src/test/java/com/yahoo/vespa/model/search/test/DocumentDatabaseTestCase.java
+++ b/config-model/src/test/java/com/yahoo/vespa/model/search/test/DocumentDatabaseTestCase.java
@@ -2,6 +2,9 @@
 package com.yahoo.vespa.model.search.test;
 
 import com.google.common.collect.ImmutableMap;
+import com.yahoo.config.model.api.ModelContext;
+import com.yahoo.config.model.deploy.DeployState;
+import com.yahoo.config.model.deploy.TestProperties;
 import com.yahoo.vespa.config.search.IndexschemaConfig;
 import com.yahoo.vespa.config.search.core.ProtonConfig;
 import com.yahoo.vespa.config.search.RankProfilesConfig;
@@ -106,12 +109,16 @@ public class DocumentDatabaseTestCase {
     }
 
     private VespaModel createModel(List<DocType> nameAndModes, String xmlTuning) {
+        return createModel(nameAndModes, xmlTuning, null);
+    }
+    private VespaModel createModel(List<DocType> nameAndModes, String xmlTuning, DeployState.Builder builder) {
         List<String> sds = new ArrayList<>(nameAndModes.size());
         for (DocType nameAndMode : nameAndModes) {
             sds.add(nameAndMode.getType());
         }
-        return new VespaModelCreatorWithMockPkg(vespaHosts, createVespaServicesXml(nameAndModes, xmlTuning),
-                ApplicationPackageUtils.generateSchemas(sds)).create();
+        var creator = new VespaModelCreatorWithMockPkg(vespaHosts, createVespaServicesXml(nameAndModes, xmlTuning),
+                ApplicationPackageUtils.generateSchemas(sds));
+        return builder != null ? creator.create(builder) : creator.create();
     }
 
     @Test
@@ -121,6 +128,12 @@ public class DocumentDatabaseTestCase {
         verifyConcurrency("store-only", "", 1.0, 0.0);
     }
     @Test
+    public void requireThatFeatureFlagConcurrencyIsReflectedCorrectlyForDefault() {
+        verifyConcurrency("index", "", 0.30, 0.30, 0.3);
+        verifyConcurrency("streaming", "", 0.6, 0.0, 0.3);
+        verifyConcurrency("store-only", "", 0.8, 0.0, 0.4);
+    }
+    @Test
     public void requireThatMixedModeConcurrencyIsReflectedCorrectlyForDefault() {
         verifyConcurrency(Arrays.asList(DocType.create("a", "index"), DocType.create("b", "streaming")), "", 1.0, Arrays.asList(0.50, 0.0));
     }
@@ -140,12 +153,22 @@ public class DocumentDatabaseTestCase {
         verifyConcurrency("streaming", feedTuning, 0.7, 0.0);
         verifyConcurrency("store-only", feedTuning, 0.7, 0.0);
     }
+    private void verifyConcurrency(String mode, String xmlTuning, double global, double local, double featureFlagConcurrency) {
+        verifyConcurrency(Arrays.asList(DocType.create("a", mode)), xmlTuning, global, Arrays.asList(local), featureFlagConcurrency);
+    }
     private void verifyConcurrency(String mode, String xmlTuning, double global, double local) {
-        verifyConcurrency(Arrays.asList(DocType.create("a", mode)), xmlTuning, global, Arrays.asList(local));
+        verifyConcurrency(Arrays.asList(DocType.create("a", mode)), xmlTuning, global, Arrays.asList(local), null);
     }
     private void verifyConcurrency(List<DocType> nameAndModes, String xmlTuning, double global, List<Double> local) {
+        verifyConcurrency(nameAndModes, xmlTuning, global, local, null);
+    }
+    private void verifyConcurrency(List<DocType> nameAndModes, String xmlTuning, double global, List<Double> local, Double featureFlagConcurrency) {
         assertEquals(nameAndModes.size(), local.size());
-        VespaModel model = createModel(nameAndModes, xmlTuning);
+        TestProperties properties = new TestProperties();
+        if (featureFlagConcurrency != null) {
+            properties.setFeedConcurrency(featureFlagConcurrency);
+        }
+        VespaModel model = createModel(nameAndModes, xmlTuning, new DeployState.Builder().properties(properties));
         ContentSearchCluster contentSearchCluster = model.getContentClusters().get("test").getSearch();
         ProtonConfig proton = getProtonCfg(contentSearchCluster);
         assertEquals(global, proton.feeding().concurrency(), SMALL);
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 1d5f9ef8b79..720da3d40d3 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
@@ -196,6 +196,7 @@ public class ModelContextImpl implements ModelContext {
         private final boolean useAsyncMessageHandlingOnSchedule;
         private final int contentNodeBucketDBStripeBits;
         private final int mergeChunkSize;
+        private final double feedConcurrency;
 
         public Properties(ApplicationId applicationId,
                           ConfigserverConfig configserverConfig,
@@ -256,6 +257,8 @@ public class ModelContextImpl implements ModelContext {
                     .with(FetchVector.Dimension.APPLICATION_ID, applicationId.serializedForm()).value();
             mergeChunkSize = Flags.MERGE_CHUNK_SIZE.bindTo(flagSource)
                     .with(FetchVector.Dimension.APPLICATION_ID, applicationId.serializedForm()).value();
+            feedConcurrency = Flags.FEED_CONCURRENCY.bindTo(flagSource)
+                    .with(FetchVector.Dimension.APPLICATION_ID, applicationId.serializedForm()).value();
         }
 
         @Override public ModelContext.FeatureFlags featureFlags() { return featureFlags; }
@@ -338,6 +341,7 @@ public class ModelContextImpl implements ModelContext {
         @Override public boolean useAsyncMessageHandlingOnSchedule() { return useAsyncMessageHandlingOnSchedule; }
         @Override public int contentNodeBucketDBStripeBits() { return contentNodeBucketDBStripeBits; }
         @Override public int mergeChunkSize() { return mergeChunkSize; }
+        @Override public double feedConcurrency() { return feedConcurrency; }
     }
 
 }
diff --git a/configserver/src/test/java/com/yahoo/vespa/config/server/ModelContextImplTest.java b/configserver/src/test/java/com/yahoo/vespa/config/server/ModelContextImplTest.java
index a86a6cb2a96..6694ac8c694 100644
--- a/configserver/src/test/java/com/yahoo/vespa/config/server/ModelContextImplTest.java
+++ b/configserver/src/test/java/com/yahoo/vespa/config/server/ModelContextImplTest.java
@@ -92,6 +92,7 @@ public class ModelContextImplTest {
         assertFalse(context.properties().useAsyncMessageHandlingOnSchedule());
         assertEquals(0, context.properties().contentNodeBucketDBStripeBits());
         assertEquals(0x400000, context.properties().mergeChunkSize());
+        assertEquals(0.5, context.properties().feedConcurrency(), 0.0);
     }
 
 }
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 ace2fc363b4..6a831cb67d5 100644
--- a/flags/src/main/java/com/yahoo/vespa/flags/Flags.java
+++ b/flags/src/main/java/com/yahoo/vespa/flags/Flags.java
@@ -324,6 +324,11 @@ public class Flags {
             "Size of merge buffer in service layer",
             "Takes effect at redeployment",
             ZONE_ID, APPLICATION_ID);
+    public static final UnboundDoubleFlag FEED_CONCURRENCY = defineDoubleFlag(
+            "feed-concurrency", 0.5,
+            "How much concurrency should be allowed for feed",
+            "Takes effect at redeployment",
+            ZONE_ID, APPLICATION_ID);
 
     public static final UnboundBooleanFlag REGIONAL_CONTAINER_REGISTRY = defineFeatureFlag(
             "regional-container-registry",