[ML] Allocate jobs based on JobParams rather than cluster state config (

#33994)

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

@@ -44,6 +44,7 @@
 import org.elasticsearch.xpack.core.logstash.LogstashFeatureSetUsage;
 import org.elasticsearch.xpack.core.ml.MachineLearningFeatureSetUsage;
 import org.elasticsearch.xpack.core.ml.MlMetadata;
+import org.elasticsearch.xpack.core.ml.MlTasks;
 import org.elasticsearch.xpack.core.ml.action.CloseJobAction;
 import org.elasticsearch.xpack.core.ml.action.DeleteCalendarAction;
 import org.elasticsearch.xpack.core.ml.action.DeleteCalendarEventAction;
@@ -331,9 +332,9 @@ public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
                 new NamedWriteableRegistry.Entry(MetaData.Custom.class, "ml", MlMetadata::new),
                 new NamedWriteableRegistry.Entry(NamedDiff.class, "ml", MlMetadata.MlMetadataDiff::new),
                 // ML - Persistent action requests
-                new NamedWriteableRegistry.Entry(PersistentTaskParams.class, StartDatafeedAction.TASK_NAME,
+                new NamedWriteableRegistry.Entry(PersistentTaskParams.class, MlTasks.DATAFEED_TASK_NAME,
                         StartDatafeedAction.DatafeedParams::new),
-                new NamedWriteableRegistry.Entry(PersistentTaskParams.class, OpenJobAction.TASK_NAME,
+                new NamedWriteableRegistry.Entry(PersistentTaskParams.class, MlTasks.JOB_TASK_NAME,
                         OpenJobAction.JobParams::new),
                 // ML - Task states
                 new NamedWriteableRegistry.Entry(PersistentTaskState.class, JobTaskState.NAME, JobTaskState::new),
@@ -379,9 +380,9 @@ public List<NamedXContentRegistry.Entry> getNamedXContent() {
                 new NamedXContentRegistry.Entry(MetaData.Custom.class, new ParseField("ml"),
                         parser -> MlMetadata.LENIENT_PARSER.parse(parser, null).build()),
                 // ML - Persistent action requests
-                new NamedXContentRegistry.Entry(PersistentTaskParams.class, new ParseField(StartDatafeedAction.TASK_NAME),
+                new NamedXContentRegistry.Entry(PersistentTaskParams.class, new ParseField(MlTasks.DATAFEED_TASK_NAME),
                         StartDatafeedAction.DatafeedParams::fromXContent),
-                new NamedXContentRegistry.Entry(PersistentTaskParams.class, new ParseField(OpenJobAction.TASK_NAME),
+                new NamedXContentRegistry.Entry(PersistentTaskParams.class, new ParseField(MlTasks.JOB_TASK_NAME),
                         OpenJobAction.JobParams::fromXContent),
                 // ML - Task states
                 new NamedXContentRegistry.Entry(PersistentTaskState.class, new ParseField(DatafeedState.NAME), DatafeedState::fromXContent),

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

@@ -12,14 +12,17 @@
 import org.elasticsearch.xpack.core.ml.job.config.JobState;
 import org.elasticsearch.xpack.core.ml.job.config.JobTaskState;
 
-import java.util.Collection;
+import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
 
 public final class MlTasks {
 
-    public static final String JOB_TASK_PREFIX = "job-";
-    public static final String DATAFEED_TASK_PREFIX = "datafeed-";
+    public static final String JOB_TASK_NAME = "xpack/ml/job";
+    public static final String DATAFEED_TASK_NAME = "xpack/ml/datafeed";
+
+    private static final String JOB_TASK_ID_PREFIX = "job-";
+    private static final String DATAFEED_TASK_ID_PREFIX = "datafeed-";
 
     private MlTasks() {
     }
@@ -29,15 +32,15 @@ private MlTasks() {
      * A datafeed id can be used as a job id, because they are stored separately in cluster state.
      */
     public static String jobTaskId(String jobId) {
-        return JOB_TASK_PREFIX + jobId;
+        return JOB_TASK_ID_PREFIX + jobId;
     }
 
     /**
      * Namespaces the task ids for datafeeds.
      * A job id can be used as a datafeed id, because they are stored separately in cluster state.
      */
     public static String datafeedTaskId(String datafeedId) {
-        return DATAFEED_TASK_PREFIX + datafeedId;
+        return DATAFEED_TASK_ID_PREFIX + datafeedId;
     }
 
     @Nullable
@@ -76,15 +79,41 @@ public static DatafeedState getDatafeedState(String datafeedId, @Nullable Persis
     }
 
     /**
-     * The job Ids of anomaly detector job tasks
-     * @param tasks Active tasks
+     * The job Ids of anomaly detector job tasks.
+     * All anomaly detector jobs are returned regardless of the status of the
+     * task (OPEN, CLOSED, FAILED etc).
+     *
+     * @param tasks Persistent tasks
      * @return The job Ids of anomaly detector job tasks
      */
     public static Set<String> openJobIds(PersistentTasksCustomMetaData tasks) {
-        Collection<PersistentTasksCustomMetaData.PersistentTask<?>> activeTasks = tasks.tasks();
-
-        return activeTasks.stream().filter(t -> t.getId().startsWith(JOB_TASK_PREFIX))
-                .map(t -> t.getId().substring(JOB_TASK_PREFIX.length()))
+        return tasks.findTasks(JOB_TASK_NAME, task -> true)
+                .stream()
+                .map(t -> t.getId().substring(JOB_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
+     * @param tasks Persistent tasks
+     * @return
+     */
+    public static boolean taskExistsForJob(String jobId, PersistentTasksCustomMetaData tasks) {
+        return openJobIds(tasks).contains(jobId);
+    }
+
+    /**
+     * Read the active anomaly detector job tasks.
+     * Active tasks are not {@code JobState.CLOSED} or {@code JobState.FAILED}.
+     *
+     * @param tasks Persistent tasks
+     * @return The job tasks excluding closed and failed jobs
+     */
+    public static List<PersistentTasksCustomMetaData.PersistentTask<?>> activeJobTasks(PersistentTasksCustomMetaData tasks) {
+        return tasks.findTasks(JOB_TASK_NAME, task -> true)
+                .stream()
+                .filter(task -> ((JobTaskState) task.getState()).getState().isAnyOf(JobState.CLOSED, JobState.FAILED) == false)
+                .collect(Collectors.toList());
+    }
 }

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

@@ -13,6 +13,7 @@
 import org.elasticsearch.action.support.master.MasterNodeRequest;
 import org.elasticsearch.client.ElasticsearchClient;
 import org.elasticsearch.cluster.metadata.MetaData;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -25,6 +26,7 @@
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.xpack.core.XPackPlugin;
 import org.elasticsearch.xpack.core.ml.MachineLearningField;
+import org.elasticsearch.xpack.core.ml.MlTasks;
 import org.elasticsearch.xpack.core.ml.job.config.Job;
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
 
@@ -35,7 +37,7 @@ public class OpenJobAction extends Action<OpenJobAction.Request, AcknowledgedRes
 
     public static final OpenJobAction INSTANCE = new OpenJobAction();
     public static final String NAME = "cluster:admin/xpack/ml/job/open";
-    public static final String TASK_NAME = "xpack/ml/job";
+
 
     private OpenJobAction() {
         super(NAME);
@@ -136,10 +138,9 @@ public static class JobParams implements XPackPlugin.XPackPersistentTaskParams {
 
         /** TODO Remove in 7.0.0 */
         public static final ParseField IGNORE_DOWNTIME = new ParseField("ignore_downtime");
-
         public static final ParseField TIMEOUT = new ParseField("timeout");
-        public static ObjectParser<JobParams, Void> PARSER = new ObjectParser<>(TASK_NAME, JobParams::new);
 
+        public static ObjectParser<JobParams, Void> PARSER = new ObjectParser<>(MlTasks.JOB_TASK_NAME, JobParams::new);
         static {
             PARSER.declareString(JobParams::setJobId, Job.ID);
             PARSER.declareBoolean((p, v) -> {}, IGNORE_DOWNTIME);
@@ -163,6 +164,7 @@ public static JobParams parseRequest(String jobId, XContentParser parser) {
         // A big state can take a while to restore.  For symmetry with the _close endpoint any
         // changes here should be reflected there too.
         private TimeValue timeout = MachineLearningField.STATE_PERSIST_RESTORE_TIMEOUT;
+        private Job job;
 
         JobParams() {
         }
@@ -178,6 +180,9 @@ public JobParams(StreamInput in) throws IOException {
                 in.readBoolean();
             }
             timeout = TimeValue.timeValueMillis(in.readVLong());
+            if (in.getVersion().onOrAfter(Version.CURRENT)) {
+                job = in.readOptionalWriteable(Job::new);
+            }
         }
 
         public String getJobId() {
@@ -196,9 +201,18 @@ public void setTimeout(TimeValue timeout) {
             this.timeout = timeout;
         }
 
+        @Nullable
+        public Job getJob() {
+            return job;
+        }
+
+        public void setJob(Job job) {
+            this.job = job;
+        }
+
         @Override
         public String getWriteableName() {
-            return TASK_NAME;
+            return MlTasks.JOB_TASK_NAME;
         }
 
         @Override
@@ -209,6 +223,9 @@ public void writeTo(StreamOutput out) throws IOException {
                 out.writeBoolean(true);
             }
             out.writeVLong(timeout.millis());
+            if (out.getVersion().onOrAfter(Version.CURRENT)) {
+                out.writeOptionalWriteable(job);
+            }
         }
 
         @Override
@@ -217,12 +234,13 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.field(Job.ID.getPreferredName(), jobId);
             builder.field(TIMEOUT.getPreferredName(), timeout.getStringRep());
             builder.endObject();
+            // The job field is streamed but not persisted
             return builder;
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(jobId, timeout);
+            return Objects.hash(jobId, timeout, job);
         }
 
         @Override
@@ -235,7 +253,8 @@ public boolean equals(Object obj) {
             }
             OpenJobAction.JobParams other = (OpenJobAction.JobParams) obj;
             return Objects.equals(jobId, other.jobId) &&
-                    Objects.equals(timeout, other.timeout);
+                    Objects.equals(timeout, other.timeout) &&
+                    Objects.equals(job, other.job);
         }
 
         @Override

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

@@ -26,6 +26,7 @@
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.index.mapper.DateFieldMapper;
 import org.elasticsearch.xpack.core.XPackPlugin;
+import org.elasticsearch.xpack.core.ml.MlTasks;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfig;
 import org.elasticsearch.xpack.core.ml.job.messages.Messages;
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
@@ -43,7 +44,6 @@ public class StartDatafeedAction
 
     public static final StartDatafeedAction INSTANCE = new StartDatafeedAction();
     public static final String NAME = "cluster:admin/xpack/ml/datafeed/start";
-    public static final String TASK_NAME = "xpack/ml/datafeed";
 
     private StartDatafeedAction() {
         super(NAME);
@@ -147,7 +147,7 @@ public boolean equals(Object obj) {
 
     public static class DatafeedParams implements XPackPlugin.XPackPersistentTaskParams {
 
-        public static ObjectParser<DatafeedParams, Void> PARSER = new ObjectParser<>(TASK_NAME, DatafeedParams::new);
+        public static ObjectParser<DatafeedParams, Void> PARSER = new ObjectParser<>(MlTasks.DATAFEED_TASK_NAME, DatafeedParams::new);
 
         static {
             PARSER.declareString((params, datafeedId) -> params.datafeedId = datafeedId, DatafeedConfig.ID);
@@ -235,7 +235,7 @@ public void setTimeout(TimeValue timeout) {
 
         @Override
         public String getWriteableName() {
-            return TASK_NAME;
+            return MlTasks.DATAFEED_TASK_NAME;
         }
 
         @Override

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

@@ -13,7 +13,7 @@
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.persistent.PersistentTaskState;
-import org.elasticsearch.xpack.core.ml.action.StartDatafeedAction;
+import org.elasticsearch.xpack.core.ml.MlTasks;
 
 import java.io.IOException;
 import java.util.Locale;
@@ -24,7 +24,7 @@ public enum DatafeedState implements PersistentTaskState {
 
     STARTED, STOPPED, STARTING, STOPPING;
 
-    public static final String NAME = StartDatafeedAction.TASK_NAME;
+    public static final String NAME = MlTasks.DATAFEED_TASK_NAME;
 
     private static final ConstructingObjectParser<DatafeedState, Void> PARSER =
             new ConstructingObjectParser<>(NAME, args -> fromString((String) args[0]));

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/job/config/JobTaskState.java

@@ -14,7 +14,7 @@
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.persistent.PersistentTaskState;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
-import org.elasticsearch.xpack.core.ml.action.OpenJobAction;
+import org.elasticsearch.xpack.core.ml.MlTasks;
 
 import java.io.IOException;
 import java.util.Objects;
@@ -23,7 +23,7 @@
 
 public class JobTaskState implements PersistentTaskState {
 
-    public static final String NAME = OpenJobAction.TASK_NAME;
+    public static final String NAME = MlTasks.JOB_TASK_NAME;
 
     private static ParseField STATE = new ParseField("state");
     private static ParseField ALLOCATION_ID = new ParseField("allocation_id");