Change downsampling thread pool settings (#96821)

Here we change the settings for the rollup thread pool like follows:

* we use a fixed number of threads (>= 1)
* we use a limited queue for tasks waiting execution
One one side we parallelise rollup operations, on the other side, if too many rollup requests are
received, rollup tasks will be rejected to avoid overloading the node.

Resolves #96758

Author: salvatore-campagna

docs/changelog/96821.yaml

@@ -0,0 +1,6 @@
+pr: 96821
+summary: Change rollup thread pool settings
+area: TSDB
+type: enhancement
+issues:
+ - 96758

x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/downsample/TransportDownsampleIndexerAction.java

@@ -30,13 +30,12 @@
 import org.elasticsearch.xpack.core.ClientHelper;
 import org.elasticsearch.xpack.core.downsample.DownsampleIndexerAction;
 import org.elasticsearch.xpack.core.rollup.action.RollupShardTask;
+import org.elasticsearch.xpack.rollup.Rollup;
 
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicReferenceArray;
 
-import static org.elasticsearch.xpack.rollup.Rollup.TASK_THREAD_POOL_NAME;
-
 /**
  * A {@link TransportBroadcastAction} that rollups all the shards of a source index into a new rollup index.
  *
@@ -69,7 +68,7 @@ public TransportDownsampleIndexerAction(
             indexNameExpressionResolver,
             DownsampleIndexerAction.Request::new,
             DownsampleIndexerAction.ShardDownsampleRequest::new,
-            TASK_THREAD_POOL_NAME
+            Rollup.DOWSAMPLE_TASK_THREAD_POOL_NAME
         );
         this.client = new OriginSettingClient(client, ClientHelper.ROLLUP_ORIGIN);
         this.clusterService = clusterService;

x-pack/plugin/rollup/src/main/java/org/elasticsearch/xpack/rollup/Rollup.java

@@ -21,6 +21,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsFilter;
 import org.elasticsearch.common.settings.SettingsModule;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.persistent.PersistentTasksExecutor;
@@ -90,6 +91,8 @@ public class Rollup extends Plugin implements ActionPlugin, PersistentTaskPlugin
     public static final int CURRENT_ROLLUP_VERSION = ROLLUP_VERSION_V2;
 
     public static final String TASK_THREAD_POOL_NAME = RollupField.NAME + "_indexing";
+    public static final String DOWSAMPLE_TASK_THREAD_POOL_NAME = "downsample_indexing";
+    public static final int DOWNSAMPLE_TASK_THREAD_POOL_QUEUE_SIZE = 256;
 
     public static final String ROLLUP_TEMPLATE_VERSION_FIELD = "rollup-version";
 
@@ -162,7 +165,7 @@ public List<RestHandler> getRestHandlers(
 
     @Override
     public List<ExecutorBuilder<?>> getExecutorBuilders(Settings settingsToUse) {
-        FixedExecutorBuilder indexing = new FixedExecutorBuilder(
+        final FixedExecutorBuilder rollup = new FixedExecutorBuilder(
             settingsToUse,
             Rollup.TASK_THREAD_POOL_NAME,
             1,
@@ -171,7 +174,16 @@ public List<ExecutorBuilder<?>> getExecutorBuilders(Settings settingsToUse) {
             false
         );
 
-        return Collections.singletonList(indexing);
+        final FixedExecutorBuilder downsample = new FixedExecutorBuilder(
+            settingsToUse,
+            Rollup.DOWSAMPLE_TASK_THREAD_POOL_NAME,
+            ThreadPool.searchOrGetThreadPoolSize(EsExecutors.allocatedProcessors(settingsToUse)),
+            Rollup.DOWNSAMPLE_TASK_THREAD_POOL_QUEUE_SIZE,
+            "xpack.downsample.thread_pool",
+            false
+        );
+
+        return List.of(rollup, downsample);
     }
 
     @Override

x-pack/plugin/rollup/src/test/java/org/elasticsearch/xpack/downsample/DownsampleActionSingleNodeTests.java

@@ -99,6 +99,7 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -1132,4 +1133,84 @@ private String createDataStream() throws Exception {
         assertAcked(client().execute(CreateDataStreamAction.INSTANCE, new CreateDataStreamAction.Request(dataStreamName)).get());
         return dataStreamName;
     }
+
+    public void testConcurrentRollup() throws IOException, InterruptedException {
+        final DownsampleConfig config = new DownsampleConfig(randomInterval());
+        SourceSupplier sourceSupplier = () -> {
+            String ts = randomDateForInterval(config.getInterval());
+            double labelDoubleValue = DATE_FORMATTER.parseMillis(ts);
+            int labelIntegerValue = randomInt();
+            long labelLongValue = randomLong();
+            String labelIpv4Address = NetworkAddress.format(randomIp(true));
+            String labelIpv6Address = NetworkAddress.format(randomIp(false));
+            Date labelDateValue = randomDate();
+            int keywordArraySize = randomIntBetween(2, 5);
+            String[] keywordArray = new String[keywordArraySize];
+            for (int i = 0; i < keywordArraySize; ++i) {
+                keywordArray[i] = randomAlphaOfLength(10);
+            }
+            int doubleArraySize = randomIntBetween(3, 10);
+            double[] doubleArray = new double[doubleArraySize];
+            for (int i = 0; i < doubleArraySize; ++i) {
+                doubleArray[i] = randomDouble();
+            }
+            return XContentFactory.jsonBuilder()
+                .startObject()
+                .field(FIELD_TIMESTAMP, ts)
+                .field(FIELD_DIMENSION_1, randomFrom(dimensionValues))
+                .field(FIELD_DIMENSION_2, randomIntBetween(1, 10))
+                .field(FIELD_NUMERIC_1, randomInt())
+                .field(FIELD_NUMERIC_2, DATE_FORMATTER.parseMillis(ts))
+                .startObject(FIELD_AGG_METRIC)
+                .field("min", randomDoubleBetween(-2000, -1001, true))
+                .field("max", randomDoubleBetween(-1000, 1000, true))
+                .field("sum", randomIntBetween(100, 10000))
+                .field("value_count", randomIntBetween(100, 1000))
+                .endObject()
+                .field(FIELD_LABEL_DOUBLE, labelDoubleValue)
+                .field(FIELD_METRIC_LABEL_DOUBLE, labelDoubleValue)
+                .field(FIELD_LABEL_INTEGER, labelIntegerValue)
+                .field(FIELD_LABEL_KEYWORD, ts)
+                .field(FIELD_LABEL_UNMAPPED, randomBoolean() ? labelLongValue : labelDoubleValue)
+                .field(FIELD_LABEL_TEXT, ts)
+                .field(FIELD_LABEL_BOOLEAN, randomBoolean())
+                .field(FIELD_LABEL_IPv4_ADDRESS, labelIpv4Address)
+                .field(FIELD_LABEL_IPv6_ADDRESS, labelIpv6Address)
+                .field(FIELD_LABEL_DATE, labelDateValue)
+                .field(FIELD_LABEL_KEYWORD_ARRAY, keywordArray)
+                .field(FIELD_LABEL_DOUBLE_ARRAY, doubleArray)
+                .startObject(FIELD_LABEL_AGG_METRIC)
+                .field("min", randomDoubleBetween(-2000, -1001, true))
+                .field("max", randomDoubleBetween(-1000, 1000, true))
+                .field("sum", Double.valueOf(randomIntBetween(100, 10000)))
+                .field("value_count", randomIntBetween(100, 1000))
+                .endObject()
+                .endObject();
+        };
+        docCount = 512; // Hard code to have 512 documents in the source index, otherwise running this test take too long.
+        bulkIndex(sourceIndex, sourceSupplier);
+        prepareSourceIndex(sourceIndex);
+
+        int n = randomIntBetween(3, 6);
+        final CountDownLatch rollupComplete = new CountDownLatch(n);
+        final List<String> targets = new ArrayList<>();
+        final List<Thread> threads = new ArrayList<>();
+        for (int i = 0; i < n; i++) {
+            final String targetIndex = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+            targets.add(targetIndex);
+            threads.add(new Thread(() -> {
+                rollup(sourceIndex, targetIndex, config);
+                rollupComplete.countDown();
+            }));
+        }
+        for (int i = 0; i < n; i++) {
+            threads.get(i).start();
+        }
+
+        assertTrue(rollupComplete.await(30, TimeUnit.SECONDS));
+
+        for (int i = 0; i < n; i++) {
+            assertRollupIndex(sourceIndex, targets.get(i), config);
+        }
+    }
 }