[ML] Fixes for stop datafeed edge cases (#49290)

The following edge cases were fixed:

1. A request to force-stop a stopping datafeed is no longer
   ignored.  Force-stop is an important recovery mechanism
   if normal stop doesn't work for some reason, and needs
   to operate on a datafeed in any state other than stopped.
2. If the node that a datafeed is running on is removed from
   the cluster during a normal stop then the stop request is
   retried (and will likely succeed on this retry by simply
   cancelling the persistent task for the affected datafeed).
3. If there are multiple simultaneous force-stop requests for
   the same datafeed we no longer fail the one that is
   processed second.  The previous behaviour was wrong as
   stopping a stopped datafeed is not an error, so stopping
   a datafeed twice simultaneously should not be either.

Backport of #49191

Author: droberts195

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/MlTasks.java

@@ -111,6 +111,7 @@ public static JobState getJobStateModifiedForReassignments(@Nullable PersistentT
 
     public static DatafeedState getDatafeedState(String datafeedId, @Nullable PersistentTasksCustomMetaData tasks) {
         PersistentTasksCustomMetaData.PersistentTask<?> task = getDatafeedTask(datafeedId, tasks);
+        // TODO: report (task != null && task.getState() == null) as STARTING in version 8, and fix side effects
         if (task != null && task.getState() != null) {
             return (DatafeedState) task.getState();
         } else {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/utils/ExceptionsHelper.java

@@ -86,4 +86,8 @@ public static <T> T requireNonNull(T obj, String paramName) {
         }
         return obj;
     }
+
+    public static Throwable unwrapCause(Throwable t) {
+       return org.elasticsearch.ExceptionsHelper.unwrapCause(t);
+    }
 }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MachineLearning.java

@@ -441,7 +441,7 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
             }
         } else {
             autodetectProcessFactory = (job, autodetectParams, executorService, onProcessCrash) ->
-                    new BlackHoleAutodetectProcess(job.getId());
+                    new BlackHoleAutodetectProcess(job.getId(), onProcessCrash);
             // factor of 1.0 makes renormalization a no-op
             normalizerProcessFactory = (jobId, quantilesState, bucketSpan, executorService) -> new MultiplyingNormalizerProcess(1.0);
         }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportStopDatafeedAction.java

@@ -17,13 +17,13 @@
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.service.ClusterService;
-import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
 import org.elasticsearch.discovery.MasterNotDiscoveredException;
+import org.elasticsearch.persistent.PersistentTasksClusterService;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
 import org.elasticsearch.persistent.PersistentTasksService;
 import org.elasticsearch.tasks.Task;
@@ -32,12 +32,14 @@
 import org.elasticsearch.xpack.core.ml.MlTasks;
 import org.elasticsearch.xpack.core.ml.action.StopDatafeedAction;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedState;
+import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
 import org.elasticsearch.xpack.ml.MachineLearning;
 import org.elasticsearch.xpack.ml.datafeed.DatafeedConfigReader;
 import org.elasticsearch.xpack.ml.datafeed.persistence.DatafeedConfigProvider;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -72,32 +74,46 @@ public TransportStopDatafeedAction(Settings settings, TransportService transport
      * @param tasks Persistent task meta data
      * @param startedDatafeedIds Started datafeed ids are added to this list
      * @param stoppingDatafeedIds Stopping datafeed ids are added to this list
+     * @param notStoppedDatafeedIds Datafeed ids are added to this list for all datafeeds that are not stopped
      */
-    static void sortDatafeedIdsByTaskState(Set<String> expandedDatafeedIds,
+    static void sortDatafeedIdsByTaskState(Collection<String> expandedDatafeedIds,
                                            PersistentTasksCustomMetaData tasks,
                                            List<String> startedDatafeedIds,
-                                           List<String> stoppingDatafeedIds) {
+                                           List<String> stoppingDatafeedIds,
+                                           List<String> notStoppedDatafeedIds) {
 
         for (String expandedDatafeedId : expandedDatafeedIds) {
             addDatafeedTaskIdAccordingToState(expandedDatafeedId, MlTasks.getDatafeedState(expandedDatafeedId, tasks),
-                    startedDatafeedIds, stoppingDatafeedIds);
+                    startedDatafeedIds, stoppingDatafeedIds, notStoppedDatafeedIds);
         }
     }
 
     private static void addDatafeedTaskIdAccordingToState(String datafeedId,
                                                           DatafeedState datafeedState,
                                                           List<String> startedDatafeedIds,
-                                                          List<String> stoppingDatafeedIds) {
+                                                          List<String> stoppingDatafeedIds,
+                                                          List<String> notStoppedDatafeedIds) {
         switch (datafeedState) {
+            case STARTING:
+                // The STARTING state is not used anywhere at present, so this should never happen.
+                // At present datafeeds that have a persistent task that hasn't yet been assigned
+                // a state are reported as STOPPED (which is not great).  It could be considered a
+                // breaking change to introduce the STARTING state though, so let's aim to do it in
+                // version 8.  Also consider treating STARTING like STARTED for stop API behaviour.
+                notStoppedDatafeedIds.add(datafeedId);
+                break;
             case STARTED:
                 startedDatafeedIds.add(datafeedId);
+                notStoppedDatafeedIds.add(datafeedId);
                 break;
             case STOPPED:
                 break;
             case STOPPING:
                 stoppingDatafeedIds.add(datafeedId);
+                notStoppedDatafeedIds.add(datafeedId);
                 break;
             default:
+                assert false : "Unexpected datafeed state " + datafeedState;
                 break;
         }
     }
@@ -123,17 +139,18 @@ protected void doExecute(Task task, StopDatafeedAction.Request request, ActionLi
 
                         List<String> startedDatafeeds = new ArrayList<>();
                         List<String> stoppingDatafeeds = new ArrayList<>();
-                        sortDatafeedIdsByTaskState(expandedIds, tasks, startedDatafeeds, stoppingDatafeeds);
+                        List<String> notStoppedDatafeeds = new ArrayList<>();
+                        sortDatafeedIdsByTaskState(expandedIds, tasks, startedDatafeeds, stoppingDatafeeds, notStoppedDatafeeds);
                         if (startedDatafeeds.isEmpty() && stoppingDatafeeds.isEmpty()) {
                             listener.onResponse(new StopDatafeedAction.Response(true));
                             return;
                         }
                         request.setResolvedStartedDatafeedIds(startedDatafeeds.toArray(new String[startedDatafeeds.size()]));
 
                         if (request.isForce()) {
-                            forceStopDatafeed(request, listener, tasks, startedDatafeeds);
+                            forceStopDatafeed(request, listener, tasks, notStoppedDatafeeds);
                         } else {
-                            normalStopDatafeed(task, request, listener, tasks, startedDatafeeds, stoppingDatafeeds);
+                            normalStopDatafeed(task, request, listener, tasks, nodes, startedDatafeeds, stoppingDatafeeds);
                         }
                     },
                     listener::onFailure
@@ -142,20 +159,20 @@ protected void doExecute(Task task, StopDatafeedAction.Request request, ActionLi
     }
 
     private void normalStopDatafeed(Task task, StopDatafeedAction.Request request, ActionListener<StopDatafeedAction.Response> listener,
-                                    PersistentTasksCustomMetaData tasks,
+                                    PersistentTasksCustomMetaData tasks, DiscoveryNodes nodes,
                                     List<String> startedDatafeeds, List<String> stoppingDatafeeds) {
-        Set<String> executorNodes = new HashSet<>();
+        final Set<String> executorNodes = new HashSet<>();
         for (String datafeedId : startedDatafeeds) {
             PersistentTasksCustomMetaData.PersistentTask<?> datafeedTask = MlTasks.getDatafeedTask(datafeedId, tasks);
             if (datafeedTask == null) {
                 // This should not happen, because startedDatafeeds was derived from the same tasks that is passed to this method
                 String msg = "Requested datafeed [" + datafeedId + "] be stopped, but datafeed's task could not be found.";
                 assert datafeedTask != null : msg;
                 logger.error(msg);
-            } else if (datafeedTask.isAssigned()) {
+            } else if (PersistentTasksClusterService.needsReassignment(datafeedTask.getAssignment(), nodes) == false) {
                 executorNodes.add(datafeedTask.getExecutorNode());
             } else {
-                // This is the easy case - the datafeed is not currently assigned to a node,
+                // This is the easy case - the datafeed is not currently assigned to a valid node,
                 // so can be gracefully stopped simply by removing its persistent task.  (Usually
                 // a graceful stop cannot be achieved by simply removing the persistent task, but
                 // if the datafeed has no running code then graceful/forceful are the same.)
@@ -176,48 +193,62 @@ private void normalStopDatafeed(Task task, StopDatafeedAction.Request request, A
 
         ActionListener<StopDatafeedAction.Response> finalListener = ActionListener.wrap(
                 r -> waitForDatafeedStopped(allDataFeedsToWaitFor, request, r, listener),
-                listener::onFailure);
+                e -> {
+                    if (ExceptionsHelper.unwrapCause(e) instanceof FailedNodeException) {
+                        // A node has dropped out of the cluster since we started executing the requests.
+                        // Since stopping an already stopped datafeed is not an error we can try again.
+                        // The datafeeds that were running on the node that dropped out of the cluster
+                        // will just have their persistent tasks cancelled.  Datafeeds that were stopped
+                        // by the previous attempt will be noops in the subsequent attempt.
+                        doExecute(task, request, listener);
+                    } else {
+                        listener.onFailure(e);
+                    }
+                });
 
         super.doExecute(task, request, finalListener);
     }
 
     private void forceStopDatafeed(final StopDatafeedAction.Request request, final ActionListener<StopDatafeedAction.Response> listener,
-                                   PersistentTasksCustomMetaData tasks, final List<String> startedDatafeeds) {
+                                   PersistentTasksCustomMetaData tasks, final List<String> notStoppedDatafeeds) {
         final AtomicInteger counter = new AtomicInteger();
-        final AtomicArray<Exception> failures = new AtomicArray<>(startedDatafeeds.size());
+        final AtomicArray<Exception> failures = new AtomicArray<>(notStoppedDatafeeds.size());
 
-        for (String datafeedId : startedDatafeeds) {
+        for (String datafeedId : notStoppedDatafeeds) {
             PersistentTasksCustomMetaData.PersistentTask<?> datafeedTask = MlTasks.getDatafeedTask(datafeedId, tasks);
             if (datafeedTask != null) {
                 persistentTasksService.sendRemoveRequest(datafeedTask.getId(),
                         new ActionListener<PersistentTasksCustomMetaData.PersistentTask<?>>() {
                     @Override
                     public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> persistentTask) {
-                        if (counter.incrementAndGet() == startedDatafeeds.size()) {
+                        if (counter.incrementAndGet() == notStoppedDatafeeds.size()) {
                             sendResponseOrFailure(request.getDatafeedId(), listener, failures);
                         }
                     }
 
                     @Override
                     public void onFailure(Exception e) {
                         final int slot = counter.incrementAndGet();
-                        if ((e instanceof ResourceNotFoundException &&
-                            Strings.isAllOrWildcard(new String[]{request.getDatafeedId()})) == false) {
+                        // We validated that the datafeed names supplied in the request existed when we started processing the action.
+                        // If the related tasks don't exist at this point then they must have been stopped by a simultaneous stop request.
+                        // This is not an error.
+                        if (ExceptionsHelper.unwrapCause(e) instanceof ResourceNotFoundException == false) {
                             failures.set(slot - 1, e);
                         }
-                        if (slot == startedDatafeeds.size()) {
+                        if (slot == notStoppedDatafeeds.size()) {
                             sendResponseOrFailure(request.getDatafeedId(), listener, failures);
                         }
                     }
                 });
             } else {
-                // This should not happen, because startedDatafeeds was derived from the same tasks that is passed to this method
+                // This should not happen, because startedDatafeeds and stoppingDatafeeds
+                // were derived from the same tasks that were passed to this method
                 String msg = "Requested datafeed [" + datafeedId + "] be force-stopped, but datafeed's task could not be found.";
                 assert datafeedTask != null : msg;
                 logger.error(msg);
                 final int slot = counter.incrementAndGet();
                 failures.set(slot - 1, new RuntimeException(msg));
-                if (slot == startedDatafeeds.size()) {
+                if (slot == notStoppedDatafeeds.size()) {
                     sendResponseOrFailure(request.getDatafeedId(), listener, failures);
                 }
             }
@@ -233,24 +264,25 @@ protected void taskOperation(StopDatafeedAction.Request request, TransportStartD
                     threadPool.executor(MachineLearning.UTILITY_THREAD_POOL_NAME).execute(new AbstractRunnable() {
                         @Override
                         public void onFailure(Exception e) {
-                            if ((e instanceof ResourceNotFoundException &&
-                                Strings.isAllOrWildcard(new String[]{request.getDatafeedId()}))) {
-                                datafeedTask.stop("stop_datafeed (api)", request.getStopTimeout());
+                            // We validated that the datafeed names supplied in the request existed when we started processing the action.
+                            // If the related task for one of them doesn't exist at this point then it must have been removed by a
+                            // simultaneous force stop request.  This is not an error.
+                            if (ExceptionsHelper.unwrapCause(e) instanceof ResourceNotFoundException) {
                                 listener.onResponse(new StopDatafeedAction.Response(true));
                             } else {
                                 listener.onFailure(e);
                             }
                         }
 
                         @Override
-                        protected void doRun() throws Exception {
+                        protected void doRun() {
                             datafeedTask.stop("stop_datafeed (api)", request.getStopTimeout());
                             listener.onResponse(new StopDatafeedAction.Response(true));
                         }
                     });
                 },
                 e -> {
-                    if (e instanceof ResourceNotFoundException) {
+                    if (ExceptionsHelper.unwrapCause(e) instanceof ResourceNotFoundException) {
                         // the task has disappeared so must have stopped
                         listener.onResponse(new StopDatafeedAction.Response(true));
                     } else {
@@ -318,7 +350,7 @@ protected StopDatafeedAction.Response newResponse(StopDatafeedAction.Request req
                         .convertToElastic(failedNodeExceptions.get(0));
             } else {
                 // This can happen we the actual task in the node no longer exists,
-                // which means the datafeed(s) have already been closed.
+                // which means the datafeed(s) have already been stopped.
                 return new StopDatafeedAction.Response(true);
             }
         }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/BlackHoleAutodetectProcess.java

@@ -21,9 +21,11 @@
 
 import java.io.IOException;
 import java.time.ZonedDateTime;
+import java.util.Arrays;
 import java.util.Date;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
@@ -37,16 +39,21 @@
  */
 public class BlackHoleAutodetectProcess implements AutodetectProcess {
 
+    public static final String MAGIC_FAILURE_VALUE = "253402300799";
+    public static final String MAGIC_FAILURE_VALUE_AS_DATE = "9999-12-31 23:59:59";
+
     private static final String FLUSH_ID = "flush-1";
 
     private final String jobId;
     private final ZonedDateTime startTime;
     private final BlockingQueue<AutodetectResult> results = new LinkedBlockingDeque<>();
+    private final Runnable onProcessCrash;
     private volatile boolean open = true;
 
-    public BlackHoleAutodetectProcess(String jobId) {
+    public BlackHoleAutodetectProcess(String jobId, Runnable onProcessCrash) {
         this.jobId = jobId;
         startTime = ZonedDateTime.now();
+        this.onProcessCrash = Objects.requireNonNull(onProcessCrash);
     }
 
     @Override
@@ -59,7 +66,13 @@ public boolean isReady() {
     }
 
     @Override
-    public void writeRecord(String[] record) throws IOException {
+    public void writeRecord(String[] record) {
+        if (Arrays.asList(record).contains(MAGIC_FAILURE_VALUE)) {
+            open = false;
+            onProcessCrash.run();
+            AutodetectResult result = new AutodetectResult(null, null, null, null, null, null, null, null, null, null, null);
+            results.add(result);
+        }
     }
 
     @Override