System index reads in separate threadpool

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,