Move persistent tasks to core

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

@@ -312,9 +312,12 @@
 import org.elasticsearch.rest.action.search.RestMultiSearchAction;
 import org.elasticsearch.rest.action.search.RestSearchAction;
 import org.elasticsearch.rest.action.search.RestSearchScrollAction;
-import org.elasticsearch.tasks.TaskManager;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.usage.UsageService;
+import org.elasticsearch.persistent.CompletionPersistentTaskAction;
+import org.elasticsearch.persistent.RemovePersistentTaskAction;
+import org.elasticsearch.persistent.StartPersistentTaskAction;
+import org.elasticsearch.persistent.UpdatePersistentTaskStatusAction;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -507,6 +510,12 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
 
         actionPlugins.stream().flatMap(p -> p.getActions().stream()).forEach(actions::register);
 
+        // Persistent tasks:
+        actions.register(StartPersistentTaskAction.INSTANCE, StartPersistentTaskAction.TransportAction.class);
+        actions.register(UpdatePersistentTaskStatusAction.INSTANCE, UpdatePersistentTaskStatusAction.TransportAction.class);
+        actions.register(CompletionPersistentTaskAction.INSTANCE, CompletionPersistentTaskAction.TransportAction.class);
+        actions.register(RemovePersistentTaskAction.INSTANCE, RemovePersistentTaskAction.TransportAction.class);
+
         return unmodifiableMap(actions.getRegistry());
     }
 

server/src/main/java/org/elasticsearch/node/Node.java

@@ -117,6 +117,7 @@
 import org.elasticsearch.plugins.MapperPlugin;
 import org.elasticsearch.plugins.MetaDataUpgrader;
 import org.elasticsearch.plugins.NetworkPlugin;
+import org.elasticsearch.plugins.PersistentTaskPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.PluginsService;
 import org.elasticsearch.plugins.RepositoryPlugin;
@@ -139,6 +140,10 @@
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.usage.UsageService;
 import org.elasticsearch.watcher.ResourceWatcherService;
+import org.elasticsearch.persistent.PersistentTasksClusterService;
+import org.elasticsearch.persistent.PersistentTasksExecutor;
+import org.elasticsearch.persistent.PersistentTasksExecutorRegistry;
+import org.elasticsearch.persistent.PersistentTasksService;
 
 import java.io.BufferedWriter;
 import java.io.Closeable;
@@ -461,6 +466,17 @@ protected Node(final Environment environment, Collection<Class<? extends Plugin>
                 threadPool, scriptModule.getScriptService(), bigArrays, searchModule.getFetchPhase(),
                 responseCollectorService);
 
+            final List<PersistentTasksExecutor<?>> tasksExecutors = pluginsService
+                .filterPlugins(PersistentTaskPlugin.class).stream()
+                .map(p -> p.getPersistentTasksExecutor(clusterService))
+                .flatMap(List::stream)
+                .collect(toList());
+
+            final PersistentTasksExecutorRegistry registry = new PersistentTasksExecutorRegistry(settings, tasksExecutors);
+            final PersistentTasksClusterService persistentTasksClusterService =
+                new PersistentTasksClusterService(settings, registry, clusterService);
+            final PersistentTasksService persistentTasksService = new PersistentTasksService(settings, clusterService, threadPool, client);
+
             modules.add(b -> {
                     b.bind(Node.class).toInstance(this);
                     b.bind(NodeService.class).toInstance(nodeService);
@@ -504,6 +520,9 @@ protected Node(final Environment environment, Collection<Class<? extends Plugin>
                     }
                     httpBind.accept(b);
                     pluginComponents.stream().forEach(p -> b.bind((Class) p.getClass()).toInstance(p));
+                    b.bind(PersistentTasksService.class).toInstance(persistentTasksService);
+                    b.bind(PersistentTasksClusterService.class).toInstance(persistentTasksClusterService);
+                    b.bind(PersistentTasksExecutorRegistry.class).toInstance(registry);
                 }
             );
             injector = modules.createInjector();

server/src/main/java/org/elasticsearch/persistent/AllocatedPersistentTask.java

@@ -0,0 +1,179 @@
+/*
+ * 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.persistent;
+
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.apache.logging.log4j.util.Supplier;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksRequest;
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.tasks.CancellableTask;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskCancelledException;
+import org.elasticsearch.tasks.TaskId;
+import org.elasticsearch.tasks.TaskManager;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Predicate;
+
+/**
+ * Represents a executor node operation that corresponds to a persistent task
+ */
+public class AllocatedPersistentTask extends CancellableTask {
+    private volatile String persistentTaskId;
+    private volatile long allocationId;
+
+    private final AtomicReference<State> state;
+    @Nullable
+    private volatile Exception failure;
+
+    private volatile PersistentTasksService persistentTasksService;
+    private volatile Logger logger;
+    private volatile TaskManager taskManager;
+
+
+    public AllocatedPersistentTask(long id, String type, String action, String description, TaskId parentTask,
+                                   Map<String, String> headers) {
+        super(id, type, action, description, parentTask, headers);
+        this.state = new AtomicReference<>(State.STARTED);
+    }
+
+    @Override
+    public boolean shouldCancelChildrenOnCancellation() {
+        return true;
+    }
+
+    // In case of persistent tasks we always need to return: `false`
+    // because in case of persistent task the parent task isn't a task in the task manager, but in cluster state.
+    // This instructs the task manager not to try to kill this persistent task when the task manager cannot find
+    // a fake parent node id "cluster" in the cluster state
+    @Override
+    public final boolean cancelOnParentLeaving() {
+        return false;
+    }
+
+    @Override
+    public Status getStatus() {
+        return new PersistentTasksNodeService.Status(state.get());
+    }
+
+    /**
+     * Updates the persistent state for the corresponding persistent task.
+     * <p>
+     * This doesn't affect the status of this allocated task.
+     */
+    public void updatePersistentStatus(Task.Status status, ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>> listener) {
+        persistentTasksService.updateStatus(persistentTaskId, allocationId, status, listener);
+    }
+
+    public String getPersistentTaskId() {
+        return persistentTaskId;
+    }
+
+    void init(PersistentTasksService persistentTasksService, TaskManager taskManager, Logger logger, String persistentTaskId, long
+            allocationId) {
+        this.persistentTasksService = persistentTasksService;
+        this.logger = logger;
+        this.taskManager = taskManager;
+        this.persistentTaskId = persistentTaskId;
+        this.allocationId = allocationId;
+    }
+
+    public Exception getFailure() {
+        return failure;
+    }
+
+    boolean markAsCancelled() {
+        return state.compareAndSet(AllocatedPersistentTask.State.STARTED, AllocatedPersistentTask.State.PENDING_CANCEL);
+    }
+
+    public State getState() {
+        return state.get();
+    }
+
+    public long getAllocationId() {
+        return allocationId;
+    }
+
+    public enum State {
+        STARTED,  // the task is currently running
+        PENDING_CANCEL, // the task is cancelled on master, cancelling it locally
+        COMPLETED     // the task is done running and trying to notify caller
+    }
+
+    /**
+     * Waits for this persistent task to have the desired state.
+     */
+    public void waitForPersistentTaskStatus(Predicate<PersistentTasksCustomMetaData.PersistentTask<?>> predicate,
+                                            @Nullable TimeValue timeout,
+                                            PersistentTasksService.WaitForPersistentTaskStatusListener<?> listener) {
+        persistentTasksService.waitForPersistentTaskStatus(persistentTaskId, predicate, timeout, listener);
+    }
+
+    public void markAsCompleted() {
+        completeAndNotifyIfNeeded(null);
+    }
+
+    public void markAsFailed(Exception e) {
+        if (CancelTasksRequest.DEFAULT_REASON.equals(getReasonCancelled())) {
+            completeAndNotifyIfNeeded(null);
+        } else {
+            completeAndNotifyIfNeeded(e);
+        }
+
+    }
+
+    private void completeAndNotifyIfNeeded(@Nullable Exception failure) {
+        State prevState = state.getAndSet(AllocatedPersistentTask.State.COMPLETED);
+        if (prevState == State.COMPLETED) {
+            logger.warn("attempt to complete task [{}] with id [{}] in the [{}] state", getAction(), getPersistentTaskId(), prevState);
+        } else {
+            if (failure != null) {
+                logger.warn((Supplier<?>) () -> new ParameterizedMessage(
+                        "task {} failed with an exception", getPersistentTaskId()), failure);
+            }
+            try {
+                this.failure = failure;
+                if (prevState == State.STARTED) {
+                    logger.trace("sending notification for completed task [{}] with id [{}]", getAction(), getPersistentTaskId());
+                    persistentTasksService.sendCompletionNotification(getPersistentTaskId(), getAllocationId(), failure, new
+                            ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
+                                @Override
+                                public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> persistentTask) {
+                                    logger.trace("notification for task [{}] with id [{}] was successful", getAction(),
+                                            getPersistentTaskId());
+                                }
+
+                                @Override
+                                public void onFailure(Exception e) {
+                                    logger.warn((Supplier<?>) () ->
+                                            new ParameterizedMessage("notification for task [{}] with id [{}] failed",
+                                                    getAction(), getPersistentTaskId()), e);
+                                }
+                            });
+                }
+            } finally {
+                taskManager.unregister(this);
+            }
+        }
+    }
+}