Periodically try to reassign unassigned persistent tasks

David Roberts · David Roberts · commit 8ff4e2e5aa60 · 2018-11-29T17:43:33.000Z
Previously persistent task assignment was checked in the following situations: - Persistent tasks are changed - A node joins or leaves the cluster - The routing table is changed - Custom metadata in the cluster state is changed - A new master node is elected However, there could be situations when a persistent task that could not be assigned to a node could become assignable due to some other change, such as memory usage on the nodes. This change adds a timed recheck of persistent task assignment to account for such situations. The timer is suspended while checks triggered by cluster state changes are in-flight to avoid adding burden to an already busy cluster. Closes #35792
diff --git a/docs/reference/modules/cluster/misc.asciidoc b/docs/reference/modules/cluster/misc.asciidoc
@@ -130,10 +130,10 @@ Plugins can create a kind of tasks called persistent tasks. Those tasks are
 usually long-live tasks and are stored in the cluster state, allowing the
 tasks to be revived after a full cluster restart.
 
-Every time a persistent task is created, the master nodes takes care of
+Every time a persistent task is created, the master node takes care of
 assigning the task to a node of the cluster, and the assigned node will then
 pick up the task and execute it locally. The process of assigning persistent
-tasks to nodes is controlled by the following property, which can be updated
+tasks to nodes is controlled by the following properties, which can be updated
 dynamically:
 
 `cluster.persistent_tasks.allocation.enable`::
@@ -148,3 +148,13 @@ This setting does not affect the persistent tasks that are already being execute
 Only newly created persistent tasks, or tasks that must be reassigned (after a node
 left the cluster, for example), are impacted by this setting.
 --
+
+`cluster.persistent_tasks.allocation.recheck_interval`::
+
+     The master node will automatically check whether persistent tasks need to
+     be assigned when the cluster state changes significantly. However, there
+     may be other factors, such as memory usage, that affect whether persistent
+     tasks can be assigned to nodes but do not cause the cluster state to change.
+     This setting controls how often assignment checks are performed to react to
+     these factors. The default is 30 seconds. The minimum permitted value is 10
+     seconds.
diff --git a/server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java b/server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java
@@ -80,6 +80,7 @@
 import org.elasticsearch.monitor.os.OsService;
 import org.elasticsearch.monitor.process.ProcessService;
 import org.elasticsearch.node.Node;
+import org.elasticsearch.persistent.PersistentTasksClusterService;
 import org.elasticsearch.persistent.decider.EnableAssignmentDecider;
 import org.elasticsearch.plugins.PluginsService;
 import org.elasticsearch.repositories.fs.FsRepository;
@@ -443,6 +444,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                     Node.BREAKER_TYPE_KEY,
                     OperationRouting.USE_ADAPTIVE_REPLICA_SELECTION_SETTING,
                     IndexGraveyard.SETTING_MAX_TOMBSTONES,
+                    PersistentTasksClusterService.CLUSTER_TASKS_ALLOCATION_RECHECK_INTERVAL_SETTING,
                     EnableAssignmentDecider.CLUSTER_TASKS_ALLOCATION_ENABLE_SETTING
             )));
 
diff --git a/server/src/main/java/org/elasticsearch/node/Node.java b/server/src/main/java/org/elasticsearch/node/Node.java
@@ -499,7 +499,7 @@ protected Node(
 
             final PersistentTasksExecutorRegistry registry = new PersistentTasksExecutorRegistry(tasksExecutors);
             final PersistentTasksClusterService persistentTasksClusterService =
-                new PersistentTasksClusterService(settings, registry, clusterService);
+                new PersistentTasksClusterService(settings, registry, clusterService, threadPool);
             final PersistentTasksService persistentTasksService = new PersistentTasksService(clusterService, threadPool, client);
 
             modules.add(b -> {
diff --git a/server/src/main/java/org/elasticsearch/persistent/PersistentTasksClusterService.java b/server/src/main/java/org/elasticsearch/persistent/PersistentTasksClusterService.java
@@ -31,30 +31,54 @@
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
+import org.elasticsearch.common.util.concurrent.FutureUtils;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData.Assignment;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
 import org.elasticsearch.persistent.decider.AssignmentDecision;
 import org.elasticsearch.persistent.decider.EnableAssignmentDecider;
+import org.elasticsearch.threadpool.ThreadPool;
 
 import java.util.Objects;
+import java.util.concurrent.Future;
 
 /**
  * Component that runs only on the master node and is responsible for assigning running tasks to nodes
  */
 public class PersistentTasksClusterService implements ClusterStateListener {
 
+    public static final Setting<TimeValue> CLUSTER_TASKS_ALLOCATION_RECHECK_INTERVAL_SETTING =
+        Setting.timeSetting("cluster.persistent_tasks.allocation.recheck_interval", TimeValue.timeValueSeconds(30),
+            TimeValue.timeValueSeconds(10), Setting.Property.Dynamic, Setting.Property.NodeScope);
+
     private static final Logger logger = LogManager.getLogger(PersistentTasksClusterService.class);
 
     private final ClusterService clusterService;
     private final PersistentTasksExecutorRegistry registry;
     private final EnableAssignmentDecider decider;
+    private final ThreadPool threadPool;
+    private final PeriodicRechecker periodicRechecker;
+    private volatile TimeValue recheckInterval;
 
-    public PersistentTasksClusterService(Settings settings, PersistentTasksExecutorRegistry registry, ClusterService clusterService) {
+    public PersistentTasksClusterService(Settings settings, PersistentTasksExecutorRegistry registry, ClusterService clusterService,
+                                         ThreadPool threadPool) {
         this.clusterService = clusterService;
         clusterService.addListener(this);
+        clusterService.getClusterSettings().addSettingsUpdateConsumer(CLUSTER_TASKS_ALLOCATION_RECHECK_INTERVAL_SETTING,
+            this::setRecheckInterval);
         this.registry = registry;
         this.decider = new EnableAssignmentDecider(settings, clusterService.getClusterSettings());
+        this.recheckInterval = CLUSTER_TASKS_ALLOCATION_RECHECK_INTERVAL_SETTING.get(settings);
+        this.threadPool = threadPool;
+        this.periodicRechecker = new PeriodicRechecker();
+    }
+
+    void setRecheckInterval(TimeValue recheckInterval) {
+        this.recheckInterval = recheckInterval;
+        periodicRechecker.rescheduleIfScheduled();
     }
 
     /**
@@ -241,24 +265,36 @@ private <Params extends PersistentTaskParams> Assignment createAssignment(final
 
     @Override
     public void clusterChanged(ClusterChangedEvent event) {
+        periodicRechecker.cancel();
         if (event.localNodeMaster()) {
             if (shouldReassignPersistentTasks(event)) {
-                logger.trace("checking task reassignment for cluster state {}", event.state().getVersion());
-                clusterService.submitStateUpdateTask("reassign persistent tasks", new ClusterStateUpdateTask() {
-                    @Override
-                    public ClusterState execute(ClusterState currentState) {
-                        return reassignTasks(currentState);
-                    }
-
-                    @Override
-                    public void onFailure(String source, Exception e) {
-                        logger.warn("failed to reassign persistent tasks", e);
-                    }
-                });
+                reassignPersistentTasks(event.state().getVersion());
+            } else {
+                periodicRechecker.schedule();
             }
         }
     }
 
+    /**
+     * Submit a cluster state update to reassign any persistent tasks that need reassigning
+     */
+    private void reassignPersistentTasks(long currentStateVersion) {
+        logger.trace("checking task reassignment for cluster state {}", currentStateVersion);
+        clusterService.submitStateUpdateTask("reassign persistent tasks", new ClusterStateUpdateTask() {
+            @Override
+            public ClusterState execute(ClusterState currentState) {
+                ClusterState newState = reassignTasks(currentState);
+                periodicRechecker.schedule();
+                return newState;
+            }
+
+            @Override
+            public void onFailure(String source, Exception e) {
+                logger.warn("failed to reassign persistent tasks", e);
+            }
+        });
+    }
+
     /**
      * Returns true if the cluster state change(s) require to reassign some persistent tasks. It can happen in the following
      * situations: a node left or is added, the routing table changed, the master node changed, the metadata changed or the
@@ -278,12 +314,21 @@ boolean shouldReassignPersistentTasks(final ClusterChangedEvent event) {
             || event.metaDataChanged()
             || masterChanged) {
 
-            for (PersistentTask<?> task : tasks.tasks()) {
-                if (needsReassignment(task.getAssignment(), event.state().nodes())) {
-                    Assignment assignment = createAssignment(task.getTaskName(), task.getParams(), event.state());
-                    if (Objects.equals(assignment, task.getAssignment()) == false) {
-                        return true;
-                    }
+            return anyTaskNeedsReassignment(tasks, event.state());
+        }
+        return false;
+    }
+
+    /**
+     * Returns true if any persistent task provided requires reassignment,
+     * i.e. is not assigned or is assigned to a non-existing node.
+     */
+    private boolean anyTaskNeedsReassignment(final PersistentTasksCustomMetaData tasks, final ClusterState state) {
+        for (PersistentTask<?> task : tasks.tasks()) {
+            if (needsReassignment(task.getAssignment(), state.nodes())) {
+                Assignment assignment = createAssignment(task.getTaskName(), task.getParams(), state);
+                if (Objects.equals(assignment, task.getAssignment()) == false) {
+                    return true;
                 }
             }
         }
@@ -347,4 +392,51 @@ private static ClusterState update(ClusterState currentState, PersistentTasksCus
             return currentState;
         }
     }
+
+    /**
+     * Class to periodically try to reassign unassigned persistent tasks.
+     */
+    private class PeriodicRechecker implements Runnable {
+
+        private volatile Future<?> nextRun;
+
+        void schedule() {
+            try {
+                synchronized (this) {
+                    FutureUtils.cancel(nextRun);
+                    nextRun = threadPool.schedule(recheckInterval, ThreadPool.Names.GENERIC, this);
+                }
+            } catch (EsRejectedExecutionException e) {
+                logger.debug("could not schedule periodic persistent task assignment check", e);
+            }
+        }
+
+        synchronized void cancel() {
+            FutureUtils.cancel(nextRun);
+            nextRun = null;
+        }
+
+        synchronized void rescheduleIfScheduled() {
+            if (nextRun != null) {
+                schedule();
+            }
+        }
+
+        @Override
+        public void run() {
+            synchronized (this) {
+                nextRun = null;
+            }
+            if (clusterService.localNode().isMasterNode()) {
+                logger.trace("periodic persistent task assignment check running");
+                ClusterState state = clusterService.state();
+                final PersistentTasksCustomMetaData tasks = state.getMetaData().custom(PersistentTasksCustomMetaData.TYPE);
+                if (tasks != null && anyTaskNeedsReassignment(tasks, state)) {
+                    reassignPersistentTasks(state.getVersion());
+                } else {
+                    schedule();
+                }
+            }
+        }
+    }
 }
diff --git a/server/src/test/java/org/elasticsearch/persistent/PersistentTasksClusterServiceTests.java b/server/src/test/java/org/elasticsearch/persistent/PersistentTasksClusterServiceTests.java
diff --git a/server/src/test/java/org/elasticsearch/persistent/PersistentTasksDecidersTestCase.java b/server/src/test/java/org/elasticsearch/persistent/PersistentTasksDecidersTestCase.java
