Allow index patterns

Author: jasontedor

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/CcrStatsAction.java

@@ -9,7 +9,9 @@
 import org.elasticsearch.action.Action;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.FailedNodeException;
+import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.TaskOperationFailure;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.tasks.BaseTasksRequest;
 import org.elasticsearch.action.support.tasks.BaseTasksResponse;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -83,21 +85,41 @@ public XContentBuilder toXContent(final XContentBuilder builder, final Params pa
         }
     }
 
-    public static class TasksRequest extends BaseTasksRequest<TasksRequest> {
+    public static class TasksRequest extends BaseTasksRequest<TasksRequest> implements IndicesRequest {
 
-        private String indexName;
+        private String[] indices;
 
-        public void setIndexName(final String indexName) {
-            this.indexName = indexName;
+        @Override
+        public String[] indices() {
+            return indices;
+        }
+
+        public void setIndices(final String[] indices) {
+            this.indices = indices;
+        }
+
+        private IndicesOptions indicesOptions = IndicesOptions.strictExpandOpenAndForbidClosed();
+
+        @Override
+        public IndicesOptions indicesOptions() {
+            return indicesOptions;
+        }
+
+        public void setIndicesOptions(final IndicesOptions indicesOptions) {
+            this.indicesOptions = indicesOptions;
         }
 
         @Override
         public boolean match(final Task task) {
-            if (task instanceof ShardFollowNodeTask) {
-                final ShardFollowNodeTask shardTask = (ShardFollowNodeTask) task;
-                return indexName.equals("") || indexName.equals("_all") || shardTask.getFollowShardId().getIndexName().equals(indexName);
-            }
-            return false;
+            /*
+             * This is a limitation of the current tasks API. When the transport action is executed, the tasks API invokes this match method
+             * to find the tasks on which to execute the task-level operation (see TransportTasksAction#nodeOperation and
+             * TransportTasksAction#taskOperation). If we do the matching here, then we can not match index patterns. Therefore, we defer
+             * deciding whether or not the task matches the request to the transport action (see TransportCcrStatsAction#taskOperation)
+             * where we can decide on the basis of the cluster state whether or not the task matches any of the index patterns in the
+             * request.
+             */
+            return task instanceof ShardFollowNodeTask;
         }
 
         @Override
@@ -108,13 +130,15 @@ public ActionRequestValidationException validate() {
         @Override
         public void readFrom(final StreamInput in) throws IOException {
             super.readFrom(in);
-            indexName = in.readOptionalString();
+            indices = in.readStringArray();
+            indicesOptions = IndicesOptions.readIndicesOptions(in);
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
-            out.writeOptionalString(indexName);
+            out.writeStringArray(indices);
+            indicesOptions.writeIndicesOptions(out);
         }
 
     }

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/TransportCcrStatsAction.java

@@ -11,27 +11,39 @@
 import org.elasticsearch.action.TaskOperationFailure;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.tasks.TransportTasksAction;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.indices.cluster.IndicesClusterStateService;
+import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.ccr.Ccr;
 
 import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
+import java.util.function.Consumer;
 
 public class TransportCcrStatsAction extends TransportTasksAction<
         ShardFollowNodeTask,
         CcrStatsAction.TasksRequest,
         CcrStatsAction.TasksResponse, CcrStatsAction.TaskResponse> {
 
+    private final IndexNameExpressionResolver resolver;
+
     @Inject
     public TransportCcrStatsAction(
             final Settings settings,
             final ClusterService clusterService,
             final TransportService transportService,
-            final ActionFilters actionFilters) {
+            final ActionFilters actionFilters,
+            final IndexNameExpressionResolver resolver) {
         super(
                 settings,
                 CcrStatsAction.NAME,
@@ -41,6 +53,7 @@ public TransportCcrStatsAction(
                 CcrStatsAction.TasksRequest::new,
                 CcrStatsAction.TasksResponse::new,
                 Ccr.CCR_THREAD_POOL_NAME);
+        this.resolver = resolver;
     }
 
     @Override
@@ -57,6 +70,20 @@ protected CcrStatsAction.TaskResponse readTaskResponse(final StreamInput in) thr
         return new CcrStatsAction.TaskResponse(in);
     }
 
+    @Override
+    protected void processTasks(final CcrStatsAction.TasksRequest request, final Consumer<ShardFollowNodeTask> operation) {
+        final ClusterState state = clusterService.state();
+        final Set<String> concreteIndices = new HashSet<>(Arrays.asList(resolver.concreteIndexNames(state, request)));
+        for (final Task task : taskManager.getTasks().values()) {
+            if (task instanceof ShardFollowNodeTask) {
+                final ShardFollowNodeTask shardFollowNodeTask = (ShardFollowNodeTask) task;
+                if (concreteIndices.contains(shardFollowNodeTask.getFollowShardId().getIndexName())) {
+                    operation.accept(shardFollowNodeTask);
+                }
+            }
+        }
+    }
+
     @Override
     protected void taskOperation(
             final CcrStatsAction.TasksRequest request,

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/rest/RestCcrStatsAction.java

@@ -6,7 +6,9 @@
 
 package org.elasticsearch.xpack.ccr.rest;
 
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestController;
@@ -32,7 +34,8 @@ public String getName() {
     @Override
     protected RestChannelConsumer prepareRequest(final RestRequest restRequest, final NodeClient client) throws IOException {
         final CcrStatsAction.TasksRequest request = new CcrStatsAction.TasksRequest();
-        request.setIndexName(restRequest.param("index", "_all"));
+        request.setIndices(Strings.splitStringByCommaToArray(restRequest.param("index")));
+        request.setIndicesOptions(IndicesOptions.fromRequest(restRequest, request.indicesOptions()));
         return channel -> client.execute(CcrStatsAction.INSTANCE, request, new RestToXContentListener<>(channel));
     }