System index reads in separate threadpool (#57936)

This commit introduces a new thread pool, `system_read`, which is
intended for use by system indices for all read operations (get and
search). The `system_read` pool is a fixed thread pool with a maximum
number of threads equal to lesser of half of the available processors
or 5. Given the combination of both get and read operations in this
thread pool, the queue size has been set to 2000. The motivation for
this change is to allow system read operations to be serviced in spite
of the number of user searches.

In order to avoid a significant performance hit due to pattern matching
on all search requests, a new metadata flag is added to mark indices
as system or non-system. Previously created system indices will have
flag added to their metadata upon upgrade to a version with this
capability.

Additionally, this change also introduces a new class, `SystemIndices`,
which encapsulates logic around system indices. Currently, the class
provides a method to check if an index is a system index and a method
to find a matching index descriptor given the name of an index.

Relates #50251
Relates #37867

build.gradle

@@ -174,8 +174,8 @@ tasks.register("verifyVersions") {
  * after the backport of the backcompat code is complete.
  */
 
-boolean bwc_tests_enabled = true
-final String bwc_tests_disabled_issue = "" /* place a PR link here when committing bwc changes */
+boolean bwc_tests_enabled = false
+final String bwc_tests_disabled_issue = "https://github.com/elastic/elasticsearch/pull/57936" /* place a PR link here when committing bwc changes */
 if (bwc_tests_enabled == false) {
   if (bwc_tests_disabled_issue.isEmpty()) {
     throw new GradleException("bwc_tests_disabled_issue must be set when bwc_tests_enabled == false")

docs/reference/modules/threadpool.asciidoc

@@ -1,9 +1,9 @@
 [[modules-threadpool]]
 === Thread pools
 
-A node uses several thread pools to manage memory consumption. 
-Queues associated with many of the thread pools enable pending requests 
-to be held instead of discarded. 
+A node uses several thread pools to manage memory consumption.
+Queues associated with many of the thread pools enable pending requests
+to be held instead of discarded.
 
 There are several thread pools, but the important ones include:
 
@@ -81,6 +81,11 @@ There are several thread pools, but the important ones include:
     Thread pool type is `scaling` with a keep-alive of `5m` and a default
     maximum size of `5`.
 
+`system_read`::
+    For read operations on system indices.
+    Thread pool type is `fixed` and a default maximum size of
+    `min(5, (`<<node.processors, `# of allocated processors`>>`) / 2)`.
+
 Changing a specific thread pool can be done by setting its type-specific
 parameters; for example, changing the number of threads in the `write` thread
 pool:

qa/full-cluster-restart/src/test/java/org/elasticsearch/upgrades/FullClusterRestartIT.java

@@ -1385,6 +1385,69 @@ public void testResize() throws Exception {
         }
     }
 
+    public void testCreateSystemIndexInOldVersion() throws Exception {
+        assumeTrue("only run on old cluster", isRunningAgainstOldCluster());
+        // create index
+        Request createTestIndex = new Request("PUT", "/test_index_old");
+        createTestIndex.setJsonEntity("{\"settings\": {\"index.number_of_replicas\": 0}}");
+        client().performRequest(createTestIndex);
+
+        Request bulk = new Request("POST", "/_bulk");
+        bulk.addParameter("refresh", "true");
+        bulk.setJsonEntity("{\"index\": {\"_index\": \"test_index_old\"}}\n" +
+            "{\"f1\": \"v1\", \"f2\": \"v2\"}\n");
+        client().performRequest(bulk);
+
+        // start a async reindex job
+        Request reindex = new Request("POST", "/_reindex");
+        reindex.setJsonEntity(
+            "{\n" +
+                "  \"source\":{\n" +
+                "    \"index\":\"test_index_old\"\n" +
+                "  },\n" +
+                "  \"dest\":{\n" +
+                "    \"index\":\"test_index_reindex\"\n" +
+                "  }\n" +
+                "}");
+        reindex.addParameter("wait_for_completion", "false");
+        Map<String, Object> response = entityAsMap(client().performRequest(reindex));
+        String taskId = (String) response.get("task");
+
+        // wait for task
+        Request getTask = new Request("GET", "/_tasks/" + taskId);
+        getTask.addParameter("wait_for_completion", "true");
+        client().performRequest(getTask);
+
+        // make sure .tasks index exists
+        assertBusy(() -> {
+            Request getTasksIndex = new Request("GET", "/.tasks");
+            assertThat(client().performRequest(getTasksIndex).getStatusLine().getStatusCode(), is(200));
+        });
+    }
+
+    @SuppressWarnings("unchecked" +
+        "")
+    public void testSystemIndexGetsUpdatedMetadata() throws Exception {
+        assumeFalse("only run in upgraded cluster", isRunningAgainstOldCluster());
+
+        assertBusy(() -> {
+            Request clusterStateRequest = new Request("GET", "/_cluster/state/metadata");
+            Map<String, Object> response = entityAsMap(client().performRequest(clusterStateRequest));
+            Map<String, Object> metadata = (Map<String, Object>) response.get("metadata");
+            assertNotNull(metadata);
+            Map<String, Object> indices = (Map<String, Object>) metadata.get("indices");
+            assertNotNull(indices);
+
+            Map<String, Object> tasksIndex = (Map<String, Object>) indices.get(".tasks");
+            assertNotNull(tasksIndex);
+            assertThat(tasksIndex.get("system"), is(true));
+
+            Map<String, Object> testIndex = (Map<String, Object>) indices.get("test_index_old");
+            assertNotNull(testIndex);
+            assertThat(testIndex.get("system"), is(false));
+        });
+    }
+
     public static void assertNumHits(String index, int numHits, int totalShards) throws IOException {
         Map<String, Object> resp = entityAsMap(client().performRequest(new Request("GET", "/" + index + "/_search")));
         assertNoFailures(resp);

qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/SystemIndicesUpgradeIT.java

@@ -0,0 +1,95 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.upgrades;
+
+import org.elasticsearch.client.Request;
+
+import java.util.Map;
+
+import static org.hamcrest.Matchers.is;
+
+public class SystemIndicesUpgradeIT extends AbstractRollingTestCase {
+
+    public void testOldDoesntHaveSystemIndexMetadata() throws Exception {
+        assumeTrue("only run in old cluster", CLUSTER_TYPE == ClusterType.OLD);
+        // create index
+        Request createTestIndex = new Request("PUT", "/test_index_old");
+        createTestIndex.setJsonEntity("{\"settings\": {\"index.number_of_replicas\": 0}}");
+        client().performRequest(createTestIndex);
+
+        Request bulk = new Request("POST", "/_bulk");
+        bulk.addParameter("refresh", "true");
+        bulk.setJsonEntity("{\"index\": {\"_index\": \"test_index_old\"}}\n" +
+            "{\"f1\": \"v1\", \"f2\": \"v2\"}\n");
+        client().performRequest(bulk);
+
+        // start a async reindex job
+        Request reindex = new Request("POST", "/_reindex");
+        reindex.setJsonEntity(
+            "{\n" +
+                "  \"source\":{\n" +
+                "    \"index\":\"test_index_old\"\n" +
+                "  },\n" +
+                "  \"dest\":{\n" +
+                "    \"index\":\"test_index_reindex\"\n" +
+                "  }\n" +
+                "}");
+        reindex.addParameter("wait_for_completion", "false");
+        Map<String, Object> response = entityAsMap(client().performRequest(reindex));
+        String taskId = (String) response.get("task");
+
+        // wait for task
+        Request getTask = new Request("GET", "/_tasks/" + taskId);
+        getTask.addParameter("wait_for_completion", "true");
+        client().performRequest(getTask);
+
+        // make sure .tasks index exists
+        assertBusy(() -> {
+            Request getTasksIndex = new Request("GET", "/.tasks");
+            assertThat(client().performRequest(getTasksIndex).getStatusLine().getStatusCode(), is(200));
+        });
+    }
+
+    public void testMixedCluster() {
+        assumeTrue("nothing to do in mixed cluster", CLUSTER_TYPE == ClusterType.MIXED);
+    }
+
+    @SuppressWarnings("unchecked")
+    public void testUpgradedCluster() throws Exception {
+        assumeTrue("only run on upgraded cluster", CLUSTER_TYPE == ClusterType.UPGRADED);
+
+        assertBusy(() -> {
+            Request clusterStateRequest = new Request("GET", "/_cluster/state/metadata");
+            Map<String, Object> response = entityAsMap(client().performRequest(clusterStateRequest));
+            Map<String, Object> metadata = (Map<String, Object>) response.get("metadata");
+            assertNotNull(metadata);
+            Map<String, Object> indices = (Map<String, Object>) metadata.get("indices");
+            assertNotNull(indices);
+
+            Map<String, Object> tasksIndex = (Map<String, Object>) indices.get(".tasks");
+            assertNotNull(tasksIndex);
+            assertThat(tasksIndex.get("system"), is(true));
+
+            Map<String, Object> testIndex = (Map<String, Object>) indices.get("test_index_old");
+            assertNotNull(testIndex);
+            assertThat(testIndex.get("system"), is(false));
+        });
+    }
+}

qa/rolling-upgrade/src/test/resources/rest-api-spec/test/upgraded_cluster/10_basic.yml

@@ -95,7 +95,9 @@
         index: .tasks
         body:
           query:
-            match_all: {}
+            match:
+              task.description:
+                query: 'reindexed_index_copy'
   - match: { hits.total: 1 }
   - match: { hits.hits.0._id: '/.+:\d+/' }
   - set: {hits.hits.0._id: task_id}

server/src/main/java/org/elasticsearch/action/get/TransportGetAction.java

@@ -115,8 +115,13 @@ protected Writeable.Reader<GetResponse> getResponseReader() {
 
     @Override
     protected String getExecutor(GetRequest request, ShardId shardId) {
-        IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
-        return indexService.getIndexSettings().isSearchThrottled() ? ThreadPool.Names.SEARCH_THROTTLED : super.getExecutor(request,
-            shardId);
+        final ClusterState clusterState = clusterService.state();
+        if (clusterState.metadata().index(shardId.getIndex()).isSystem()) {
+            return ThreadPool.Names.SYSTEM_READ;
+        } else if (indicesService.indexServiceSafe(shardId.getIndex()).getIndexSettings().isSearchThrottled()) {
+            return ThreadPool.Names.SEARCH_THROTTLED;
+        } else {
+            return super.getExecutor(request, shardId);
+        }
     }
 }

server/src/main/java/org/elasticsearch/action/get/TransportShardMultiGetAction.java

@@ -128,8 +128,13 @@ protected MultiGetShardResponse shardOperation(MultiGetShardRequest request, Sha
 
     @Override
     protected String getExecutor(MultiGetShardRequest request, ShardId shardId) {
-        IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
-        return indexService.getIndexSettings().isSearchThrottled() ? ThreadPool.Names.SEARCH_THROTTLED : super.getExecutor(request,
-            shardId);
+        final ClusterState clusterState = clusterService.state();
+        if (clusterState.metadata().index(shardId.getIndex()).isSystem()) {
+            return ThreadPool.Names.SYSTEM_READ;
+        } else if (indicesService.indexServiceSafe(shardId.getIndex()).getIndexSettings().isSearchThrottled()) {
+            return ThreadPool.Names.SEARCH_THROTTLED;
+        } else {
+            return super.getExecutor(request, shardId);
+        }
     }
 }

server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java

@@ -519,10 +519,17 @@ private void executeSearch(SearchTask task, SearchTimeProvider timeProvider, Sea
         BiFunction<String, String, Transport.Connection> connectionLookup = buildConnectionLookup(searchRequest.getLocalClusterAlias(),
             nodes::get, remoteConnections, searchTransportService::getConnection);
         boolean preFilterSearchShards = shouldPreFilterSearchShards(clusterState, searchRequest, indices, shardIterators.size());
-        searchAsyncAction(task, searchRequest, shardIterators, timeProvider, connectionLookup, clusterState,
+        final Executor asyncSearchExecutor = asyncSearchExecutor(indices, clusterState);
+        searchAsyncAction(task, searchRequest, asyncSearchExecutor, shardIterators, timeProvider, connectionLookup, clusterState,
             Collections.unmodifiableMap(aliasFilter), concreteIndexBoosts, routingMap, listener, preFilterSearchShards, clusters).start();
     }
 
+    Executor asyncSearchExecutor(final Index[] indices, final ClusterState clusterState) {
+        final boolean onlySystemIndices =
+            Arrays.stream(indices).allMatch(index -> clusterState.metadata().index(index.getName()).isSystem());
+        return onlySystemIndices ? threadPool.executor(ThreadPool.Names.SYSTEM_READ) : threadPool.executor(ThreadPool.Names.SEARCH);
+    }
+
     static BiFunction<String, String, Transport.Connection> buildConnectionLookup(String requestClusterAlias,
                                                               Function<String, DiscoveryNode> localNodes,
                                                               BiFunction<String, String, DiscoveryNode> remoteNodes,
@@ -584,6 +591,7 @@ static GroupShardsIterator<SearchShardIterator> mergeShardsIterators(GroupShards
     }
 
     private AbstractSearchAsyncAction<? extends SearchPhaseResult> searchAsyncAction(SearchTask task, SearchRequest searchRequest,
+                                                        Executor executor,
                                                         GroupShardsIterator<SearchShardIterator> shardIterators,
                                                         SearchTimeProvider timeProvider,
                                                         BiFunction<String, String, Transport.Connection> connectionLookup,
@@ -594,14 +602,14 @@ private AbstractSearchAsyncAction<? extends SearchPhaseResult> searchAsyncAction
                                                         ActionListener<SearchResponse> listener,
                                                         boolean preFilter,
                                                         SearchResponse.Clusters clusters) {
-        Executor executor = threadPool.executor(ThreadPool.Names.SEARCH);
         if (preFilter) {
             return new CanMatchPreFilterSearchPhase(logger, searchTransportService, connectionLookup,
                 aliasFilter, concreteIndexBoosts, indexRoutings, executor, searchRequest, listener, shardIterators,
                 timeProvider, clusterState, task, (iter) -> {
                 AbstractSearchAsyncAction<? extends SearchPhaseResult> action = searchAsyncAction(
                     task,
                     searchRequest,
+                    executor,
                     iter,
                     timeProvider,
                     connectionLookup,