Threadpool merge scheduler (#120869)

This adds a new merge scheduler implementation that uses a (new)
dedicated thread pool to run the merges. This way the number of
concurrent merges is limited to the number of threads in the pool
(i.e. the number of allocated processors to the ES JVM).

It implements dynamic IO throttling (the same target IO rate for all
merges, roughly, with caveats) that's adjusted based on the number
of currently active (queued + running) merges.
Smaller merges are always preferred to larger ones, irrespective of
the index shard that they're coming from.
The implementation also supports the per-shard "max thread count"
and "max merge count" settings, the later being used today for indexing throttling.
Note that IO throttling, max merge count, and max thread count work similarly,
but not identical, to their siblings in the ConcurrentMergeScheduler.

The per-shard merge statistics are not affected, and the thread-pool statistics should
reflect the merge ones (i.e. the completed thread pool stats reflects the total
number of merges, across shards, per node).

Author: albertzaharovits

docs/changelog/120869.yaml

@@ -0,0 +1,5 @@
+pr: 120869
+summary: Threadpool merge scheduler
+area: Engine
+type: feature
+issues: []

server/src/internalClusterTest/java/org/elasticsearch/index/engine/InternalEngineMergeIT.java

@@ -8,24 +8,40 @@
  */
 package org.elasticsearch.index.engine;
 
+import org.elasticsearch.action.admin.cluster.node.stats.NodeStats;
+import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsResponse;
 import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.bulk.BulkResponse;
 import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
+import org.elasticsearch.threadpool.ThreadPool;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
 import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.lessThan;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 
-@ClusterScope(supportsDedicatedMasters = false, numDataNodes = 1, scope = Scope.SUITE)
+@ClusterScope(supportsDedicatedMasters = false, numDataNodes = 1, numClientNodes = 0, scope = Scope.TEST)
 public class InternalEngineMergeIT extends ESIntegTestCase {
 
+    private boolean useThreadPoolMerging;
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
+        useThreadPoolMerging = randomBoolean();
+        Settings.Builder settings = Settings.builder().put(super.nodeSettings(nodeOrdinal, otherSettings));
+        settings.put(ThreadPoolMergeScheduler.USE_THREAD_POOL_MERGE_SCHEDULER_SETTING.getKey(), useThreadPoolMerging);
+        return settings.build();
+    }
+
     public void testMergesHappening() throws Exception {
         final int numOfShards = randomIntBetween(1, 5);
         // some settings to keep num segments low
@@ -83,4 +99,60 @@ public void testMergesHappening() throws Exception {
         assertThat(count, lessThanOrEqualTo(upperNumberSegments));
     }
 
+    public void testMergesUseTheMergeThreadPool() throws Exception {
+        final String indexName = randomIdentifier();
+        createIndex(indexName, indexSettings(randomIntBetween(1, 3), 0).build());
+        long id = 0;
+        final int minMerges = randomIntBetween(1, 5);
+        long totalDocs = 0;
+
+        while (true) {
+            int docs = randomIntBetween(100, 200);
+            totalDocs += docs;
+
+            BulkRequestBuilder request = client().prepareBulk();
+            for (int j = 0; j < docs; ++j) {
+                request.add(
+                    new IndexRequest(indexName).id(Long.toString(id++))
+                        .source(jsonBuilder().startObject().field("l", randomLong()).endObject())
+                );
+            }
+            BulkResponse response = request.get();
+            assertNoFailures(response);
+            refresh(indexName);
+
+            var mergesResponse = client().admin().indices().prepareStats(indexName).clear().setMerge(true).get();
+            var primaries = mergesResponse.getIndices().get(indexName).getPrimaries();
+            if (primaries.merge.getTotal() >= minMerges) {
+                break;
+            }
+        }
+
+        forceMerge();
+        refresh(indexName);
+
+        // after a force merge there should only be 1 segment per shard
+        var shardsWithMultipleSegments = getShardSegments().stream()
+            .filter(shardSegments -> shardSegments.getSegments().size() > 1)
+            .toList();
+        assertTrue("there are shards with multiple segments " + shardsWithMultipleSegments, shardsWithMultipleSegments.isEmpty());
+
+        final long expectedTotalDocs = totalDocs;
+        assertHitCount(prepareSearch(indexName).setQuery(QueryBuilders.matchAllQuery()).setTrackTotalHits(true), expectedTotalDocs);
+
+        IndicesStatsResponse indicesStats = client().admin().indices().prepareStats(indexName).setMerge(true).get();
+        long mergeCount = indicesStats.getIndices().get(indexName).getPrimaries().merge.getTotal();
+        NodesStatsResponse nodesStatsResponse = client().admin().cluster().prepareNodesStats().setThreadPool(true).get();
+        assertThat(nodesStatsResponse.getNodes().size(), equalTo(1));
+
+        NodeStats nodeStats = nodesStatsResponse.getNodes().get(0);
+        if (useThreadPoolMerging) {
+            assertThat(
+                nodeStats.getThreadPool().stats().stream().filter(s -> ThreadPool.Names.MERGE.equals(s.name())).findAny().get().completed(),
+                equalTo(mergeCount)
+            );
+        } else {
+            assertTrue(nodeStats.getThreadPool().stats().stream().filter(s -> ThreadPool.Names.MERGE.equals(s.name())).findAny().isEmpty());
+        }
+    }
 }