[ML] Job In Index: Migrate config from the clusterstate

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

@@ -410,14 +410,14 @@ private void checkDatafeedIsStopped(Supplier<String> msg, String datafeedId, Per
             }
         }
 
-        private Builder putJobs(Collection<Job> jobs) {
+        public Builder putJobs(Collection<Job> jobs) {
             for (Job job : jobs) {
                 putJob(job, true);
             }
             return this;
         }
 
-        private Builder putDatafeeds(Collection<DatafeedConfig> datafeeds) {
+        public Builder putDatafeeds(Collection<DatafeedConfig> datafeeds) {
             for (DatafeedConfig datafeed : datafeeds) {
                 this.datafeeds.put(datafeed.getId(), datafeed);
             }

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

@@ -98,6 +98,23 @@ public static Set<String> openJobIds(@Nullable PersistentTasksCustomMetaData tas
                 .collect(Collectors.toSet());
     }
 
+    /**
+     * The datafeed Ids of started datafeed tasks
+     *
+     * @param tasks Persistent tasks. If null an empty set is returned.
+     * @return The Ids of running datafeed tasks
+     */
+    public static Set<String> startedDatafeedIds(@Nullable PersistentTasksCustomMetaData tasks) {
+        if (tasks == null) {
+            return Collections.emptySet();
+        }
+
+        return tasks.findTasks(DATAFEED_TASK_NAME, task -> true)
+                .stream()
+                .map(t -> t.getId().substring(DATAFEED_TASK_ID_PREFIX.length()))
+                .collect(Collectors.toSet());
+    }
+
     /**
      * Is there an ml anomaly detector job task for the job {@code jobId}?
      * @param jobId The job id

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/MlTasksTests.java

@@ -91,6 +91,26 @@ public void testOpenJobIds_GivenNull() {
         assertThat(MlTasks.openJobIds(null), empty());
     }
 
+    public void testStartedDatafeedIds() {
+        PersistentTasksCustomMetaData.Builder tasksBuilder =  PersistentTasksCustomMetaData.builder();
+        assertThat(MlTasks.openJobIds(tasksBuilder.build()), empty());
+
+        tasksBuilder.addTask(MlTasks.jobTaskId("job-1"), MlTasks.JOB_TASK_NAME, new OpenJobAction.JobParams("foo-1"),
+                new PersistentTasksCustomMetaData.Assignment("node-1", "test assignment"));
+        tasksBuilder.addTask(MlTasks.datafeedTaskId("df1"), MlTasks.DATAFEED_TASK_NAME,
+                new StartDatafeedAction.DatafeedParams("df1", 0L),
+                new PersistentTasksCustomMetaData.Assignment("node-1", "test assignment"));
+        tasksBuilder.addTask(MlTasks.datafeedTaskId("df2"), MlTasks.DATAFEED_TASK_NAME,
+                new StartDatafeedAction.DatafeedParams("df2", 0L),
+                new PersistentTasksCustomMetaData.Assignment("node-2", "test assignment"));
+
+        assertThat(MlTasks.startedDatafeedIds(tasksBuilder.build()), containsInAnyOrder("df1", "df2"));
+    }
+
+    public void testStartedDatafeedIds_GivenNull() {
+        assertThat(MlTasks.startedDatafeedIds(null), empty());
+    }
+
     public void testTaskExistsForJob() {
         PersistentTasksCustomMetaData.Builder tasksBuilder =  PersistentTasksCustomMetaData.builder();
         assertFalse(MlTasks.taskExistsForJob("job-1", tasksBuilder.build()));

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

@@ -443,7 +443,7 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
                 jobDataCountsPersister,
                 datafeedManager,
                 auditor,
-                new MlAssignmentNotifier(auditor, clusterService),
+                new MlAssignmentNotifier(auditor, threadPool, client, clusterService),
                 memoryTracker
         );
     }
@@ -458,7 +458,7 @@ public List<PersistentTasksExecutor<?>> getPersistentTasksExecutor(ClusterServic
 
         return Arrays.asList(
                 new TransportOpenJobAction.OpenJobPersistentTasksExecutor(settings, clusterService, autodetectProcessManager.get(),
-                    memoryTracker.get()),
+                    memoryTracker.get(), client),
                 new TransportStartDatafeedAction.StartDatafeedPersistentTasksExecutor(settings, datafeedManager.get())
         );
     }

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

@@ -7,7 +7,11 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.Version;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.ClusterChangedEvent;
+import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateListener;
 import org.elasticsearch.cluster.LocalNodeMasterListener;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -29,12 +33,23 @@ public class MlAssignmentNotifier implements ClusterStateListener, LocalNodeMast
 
     private final Auditor auditor;
     private final ClusterService clusterService;
-
+    private final MlConfigMigrator mlConfigMigrator;
+    private final ThreadPool threadPool;
     private final AtomicBoolean enabled = new AtomicBoolean(false);
 
-    MlAssignmentNotifier(Auditor auditor, ClusterService clusterService) {
+    MlAssignmentNotifier(Auditor auditor, ThreadPool threadPool, Client client, ClusterService clusterService) {
+        this.auditor = auditor;
+        this.clusterService = clusterService;
+        this.mlConfigMigrator = new MlConfigMigrator(client, clusterService);
+        this.threadPool = threadPool;
+        clusterService.addLocalNodeMasterListener(this);
+    }
+
+    MlAssignmentNotifier(Auditor auditor, ThreadPool threadPool, MlConfigMigrator mlConfigMigrator, ClusterService clusterService) {
         this.auditor = auditor;
         this.clusterService = clusterService;
+        this.mlConfigMigrator = mlConfigMigrator;
+        this.threadPool = threadPool;
         clusterService.addLocalNodeMasterListener(this);
     }
 
@@ -71,6 +86,22 @@ public void clusterChanged(ClusterChangedEvent event) {
             return;
         }
 
+        Version minNodeVersion = event.state().nodes().getMinNodeVersion();
+        if (minNodeVersion.onOrAfter(Version.V_6_6_0)) {
+            // ok to migrate
+            mlConfigMigrator.migrateConfigsWithoutTasks(event.state(), ActionListener.wrap(
+                    response -> threadPool.executor(executorName()).execute(() -> auditChangesToMlTasks(current, previous, event.state())),
+                    e -> logger.error("error migrating ml configurations", e)
+            ));
+        } else {
+            threadPool.executor(executorName()).execute(() -> auditChangesToMlTasks(current, previous, event.state()));
+        }
+
+    }
+
+    private void auditChangesToMlTasks(PersistentTasksCustomMetaData current, PersistentTasksCustomMetaData previous,
+                                       ClusterState state) {
+
         for (PersistentTask<?> currentTask : current.tasks()) {
             Assignment currentAssignment = currentTask.getAssignment();
             PersistentTask<?> previousTask = previous != null ? previous.getTask(currentTask.getId()) : null;
@@ -83,7 +114,7 @@ public void clusterChanged(ClusterChangedEvent event) {
                 if (currentAssignment.getExecutorNode() == null) {
                     auditor.warning(jobId, "No node found to open job. Reasons [" + currentAssignment.getExplanation() + "]");
                 } else {
-                    DiscoveryNode node = event.state().nodes().get(currentAssignment.getExecutorNode());
+                    DiscoveryNode node = state.nodes().get(currentAssignment.getExecutorNode());
                     auditor.info(jobId, "Opening job on node [" + node.toString() + "]");
                 }
             } else if (MlTasks.DATAFEED_TASK_NAME.equals(currentTask.getTaskName())) {
@@ -97,7 +128,7 @@ public void clusterChanged(ClusterChangedEvent event) {
                         auditor.warning(jobId, msg);
                     }
                 } else {
-                    DiscoveryNode node = event.state().nodes().get(currentAssignment.getExecutorNode());
+                    DiscoveryNode node = state.nodes().get(currentAssignment.getExecutorNode());
                     if (jobId != null) {
                         auditor.info(jobId, "Starting datafeed [" + datafeedParams.getDatafeedId() + "] on node [" + node + "]");
                     }