Task Management: Add framework for registering and communicating with tasks

core/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -28,6 +28,8 @@
 import org.elasticsearch.action.admin.cluster.node.liveness.TransportLivenessAction;
 import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsAction;
 import org.elasticsearch.action.admin.cluster.node.stats.TransportNodesStatsAction;
+import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksAction;
+import org.elasticsearch.action.admin.cluster.node.tasks.list.TransportListTasksAction;
 import org.elasticsearch.action.admin.cluster.repositories.delete.DeleteRepositoryAction;
 import org.elasticsearch.action.admin.cluster.repositories.delete.TransportDeleteRepositoryAction;
 import org.elasticsearch.action.admin.cluster.repositories.get.GetRepositoriesAction;
@@ -255,6 +257,7 @@ protected void configure() {
         registerAction(NodesInfoAction.INSTANCE, TransportNodesInfoAction.class);
         registerAction(NodesStatsAction.INSTANCE, TransportNodesStatsAction.class);
         registerAction(NodesHotThreadsAction.INSTANCE, TransportNodesHotThreadsAction.class);
+        registerAction(ListTasksAction.INSTANCE, TransportListTasksAction.class);
 
         registerAction(ClusterStatsAction.INSTANCE, TransportClusterStatsAction.class);
         registerAction(ClusterStateAction.INSTANCE, TransportClusterStateAction.class);

core/src/main/java/org/elasticsearch/action/TaskOperationFailure.java

@@ -0,0 +1,117 @@
+/*
+ * 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.action;
+
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.rest.RestStatus;
+
+import java.io.IOException;
+
+import static org.elasticsearch.ExceptionsHelper.detailedMessage;
+
+/**
+ * Information about task operation failures
+ *
+ * The class is final due to serialization limitations
+ */
+public final class TaskOperationFailure implements Writeable<TaskOperationFailure>, ToXContent {
+
+    private final String nodeId;
+
+    private final long taskId;
+
+    private final Throwable reason;
+
+    private final RestStatus status;
+
+    public TaskOperationFailure(StreamInput in) throws IOException {
+        nodeId = in.readString();
+        taskId = in.readLong();
+        reason = in.readThrowable();
+        status = RestStatus.readFrom(in);
+    }
+
+    public TaskOperationFailure(String nodeId, long taskId, Throwable t) {
+        this.nodeId = nodeId;
+        this.taskId = taskId;
+        this.reason = t;
+        status = ExceptionsHelper.status(t);
+    }
+
+    public String getNodeId() {
+        return this.nodeId;
+    }
+
+    public long getTaskId() {
+        return this.taskId;
+    }
+
+    public String getReason() {
+        return detailedMessage(reason);
+    }
+
+    public RestStatus getStatus() {
+        return status;
+    }
+
+    public Throwable getCause() {
+        return reason;
+    }
+
+    @Override
+    public TaskOperationFailure readFrom(StreamInput in) throws IOException {
+        return new TaskOperationFailure(in);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeString(nodeId);
+        out.writeLong(taskId);
+        out.writeThrowable(reason);
+        RestStatus.writeTo(out, status);
+    }
+
+    @Override
+    public String toString() {
+        return "[" + nodeId + "][" + taskId + "] failed, reason [" + getReason() + "]";
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.field("task_id", getTaskId());
+        builder.field("node_id", getNodeId());
+        builder.field("status", status.name());
+        if (reason != null) {
+            builder.field("reason");
+            builder.startObject();
+            ElasticsearchException.toXContent(builder, params, reason);
+            builder.endObject();
+        }
+        return builder;
+
+    }
+
+}

core/src/main/java/org/elasticsearch/action/admin/cluster/health/TransportClusterHealthAction.java

@@ -38,6 +38,7 @@
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.gateway.GatewayAllocator;
 import org.elasticsearch.index.IndexNotFoundException;
+import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
@@ -75,7 +76,13 @@ protected ClusterHealthResponse newResponse() {
     }
 
     @Override
-    protected void masterOperation(final ClusterHealthRequest request, final ClusterState unusedState, final ActionListener<ClusterHealthResponse> listener) {
+    protected final void masterOperation(ClusterHealthRequest request, ClusterState state, ActionListener<ClusterHealthResponse> listener) throws Exception {
+        logger.warn("attempt to execute a cluster health operation without a task");
+        throw new UnsupportedOperationException("task parameter is required for this operation");
+    }
+
+    @Override
+    protected void masterOperation(Task task, final ClusterHealthRequest request, final ClusterState unusedState, final ActionListener<ClusterHealthResponse> listener) {
         if (request.waitForEvents() != null) {
             final long endTimeMS = TimeValue.nsecToMSec(System.nanoTime()) + request.timeout().millis();
             clusterService.submitStateUpdateTask("cluster_health (wait_for_events [" + request.waitForEvents() + "])", new ClusterStateUpdateTask(request.waitForEvents()) {
@@ -95,7 +102,7 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS
                 @Override
                 public void onNoLongerMaster(String source) {
                     logger.trace("stopped being master while waiting for events with priority [{}]. retrying.", request.waitForEvents());
-                    doExecute(request, listener);
+                    doExecute(task, request, listener);
                 }
 
                 @Override

core/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/ListTasksAction.java

@@ -0,0 +1,46 @@
+/*
+ * 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.action.admin.cluster.node.tasks.list;
+
+import org.elasticsearch.action.Action;
+import org.elasticsearch.client.ElasticsearchClient;
+
+/**
+ * Action for retrieving a list of currently running tasks
+ */
+public class ListTasksAction extends Action<ListTasksRequest, ListTasksResponse, ListTasksRequestBuilder> {
+
+    public static final ListTasksAction INSTANCE = new ListTasksAction();
+    public static final String NAME = "cluster:monitor/tasks/lists";
+
+    private ListTasksAction() {
+        super(NAME);
+    }
+
+    @Override
+    public ListTasksResponse newResponse() {
+        return new ListTasksResponse();
+    }
+
+    @Override
+    public ListTasksRequestBuilder newRequestBuilder(ElasticsearchClient client) {
+        return new ListTasksRequestBuilder(client, this);
+    }
+}

core/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/ListTasksRequest.java

@@ -0,0 +1,69 @@
+/*
+ * 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.action.admin.cluster.node.tasks.list;
+
+import org.elasticsearch.action.support.tasks.BaseTasksRequest;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+
+import java.io.IOException;
+
+/**
+ * A request to get node tasks
+ */
+public class ListTasksRequest extends BaseTasksRequest<ListTasksRequest> {
+
+    private boolean detailed = false;
+
+    /**
+     * Get information from nodes based on the nodes ids specified. If none are passed, information
+     * for all nodes will be returned.
+     */
+    public ListTasksRequest(String... nodesIds) {
+        super(nodesIds);
+    }
+
+    /**
+     * Should the detailed task information be returned.
+     */
+    public boolean detailed() {
+        return this.detailed;
+    }
+
+    /**
+     * Should the node settings be returned.
+     */
+    public ListTasksRequest detailed(boolean detailed) {
+        this.detailed = detailed;
+        return this;
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        super.readFrom(in);
+        detailed = in.readBoolean();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeBoolean(detailed);
+    }
+}

core/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/ListTasksRequestBuilder.java

@@ -0,0 +1,41 @@
+/*
+ * 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.action.admin.cluster.node.tasks.list;
+
+import org.elasticsearch.action.support.tasks.TasksRequestBuilder;
+import org.elasticsearch.client.ElasticsearchClient;
+
+/**
+ * Builder for the request to retrieve the list of tasks running on the specified nodes
+ */
+public class ListTasksRequestBuilder extends TasksRequestBuilder<ListTasksRequest, ListTasksResponse, ListTasksRequestBuilder> {
+
+    public ListTasksRequestBuilder(ElasticsearchClient client, ListTasksAction action) {
+        super(client, action, new ListTasksRequest());
+    }
+
+    /**
+     * Should detailed task information be returned.
+     */
+    public ListTasksRequestBuilder setDetailed(boolean detailed) {
+        request.detailed(detailed);
+        return this;
+    }
+}