[ML] Data frame analytics as persistent tasks

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

@@ -53,9 +53,9 @@
 import org.elasticsearch.xpack.core.indexlifecycle.IndexLifecycleMetadata;
 import org.elasticsearch.xpack.core.indexlifecycle.LifecycleAction;
 import org.elasticsearch.xpack.core.indexlifecycle.LifecycleType;
-import org.elasticsearch.xpack.core.indexlifecycle.SetPriorityAction;
 import org.elasticsearch.xpack.core.indexlifecycle.ReadOnlyAction;
 import org.elasticsearch.xpack.core.indexlifecycle.RolloverAction;
+import org.elasticsearch.xpack.core.indexlifecycle.SetPriorityAction;
 import org.elasticsearch.xpack.core.indexlifecycle.ShrinkAction;
 import org.elasticsearch.xpack.core.indexlifecycle.TimeseriesLifecycleType;
 import org.elasticsearch.xpack.core.indexlifecycle.UnfollowAction;
@@ -87,6 +87,8 @@
 import org.elasticsearch.xpack.core.ml.action.GetCalendarEventsAction;
 import org.elasticsearch.xpack.core.ml.action.GetCalendarsAction;
 import org.elasticsearch.xpack.core.ml.action.GetCategoriesAction;
+import org.elasticsearch.xpack.core.ml.action.GetDataFrameAnalyticsAction;
+import org.elasticsearch.xpack.core.ml.action.GetDataFrameAnalyticsStatsAction;
 import org.elasticsearch.xpack.core.ml.action.GetDatafeedsAction;
 import org.elasticsearch.xpack.core.ml.action.GetDatafeedsStatsAction;
 import org.elasticsearch.xpack.core.ml.action.GetFiltersAction;
@@ -105,12 +107,13 @@
 import org.elasticsearch.xpack.core.ml.action.PostDataAction;
 import org.elasticsearch.xpack.core.ml.action.PreviewDatafeedAction;
 import org.elasticsearch.xpack.core.ml.action.PutCalendarAction;
+import org.elasticsearch.xpack.core.ml.action.PutDataFrameAnalyticsAction;
 import org.elasticsearch.xpack.core.ml.action.PutDatafeedAction;
 import org.elasticsearch.xpack.core.ml.action.PutFilterAction;
 import org.elasticsearch.xpack.core.ml.action.PutJobAction;
 import org.elasticsearch.xpack.core.ml.action.RevertModelSnapshotAction;
-import org.elasticsearch.xpack.core.ml.action.RunAnalyticsAction;
 import org.elasticsearch.xpack.core.ml.action.SetUpgradeModeAction;
+import org.elasticsearch.xpack.core.ml.action.StartDataFrameAnalyticsAction;
 import org.elasticsearch.xpack.core.ml.action.StartDatafeedAction;
 import org.elasticsearch.xpack.core.ml.action.StopDatafeedAction;
 import org.elasticsearch.xpack.core.ml.action.UpdateCalendarJobAction;
@@ -294,8 +297,11 @@ public List<Action<? extends ActionResponse>> getClientActions() {
                 PostCalendarEventsAction.INSTANCE,
                 PersistJobAction.INSTANCE,
                 FindFileStructureAction.INSTANCE,
-                RunAnalyticsAction.INSTANCE,
                 SetUpgradeModeAction.INSTANCE,
+                PutDataFrameAnalyticsAction.INSTANCE,
+                GetDataFrameAnalyticsAction.INSTANCE,
+                GetDataFrameAnalyticsStatsAction.INSTANCE,
+                StartDataFrameAnalyticsAction.INSTANCE,
                 // security
                 ClearRealmCacheAction.INSTANCE,
                 ClearRolesCacheAction.INSTANCE,
@@ -450,6 +456,8 @@ public List<NamedXContentRegistry.Entry> getNamedXContent() {
                         StartDatafeedAction.DatafeedParams::fromXContent),
                 new NamedXContentRegistry.Entry(PersistentTaskParams.class, new ParseField(MlTasks.JOB_TASK_NAME),
                         OpenJobAction.JobParams::fromXContent),
+                new NamedXContentRegistry.Entry(PersistentTaskParams.class, new ParseField(MlTasks.DATA_FRAME_ANALYTICS_TASK_NAME),
+                        StartDataFrameAnalyticsAction.TaskParams::fromXContent),
                 // ML - Task states
                 new NamedXContentRegistry.Entry(PersistentTaskState.class, new ParseField(DatafeedState.NAME), DatafeedState::fromXContent),
                 new NamedXContentRegistry.Entry(PersistentTaskState.class, new ParseField(JobTaskState.NAME), JobTaskState::fromXContent),

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

@@ -11,6 +11,8 @@
 import org.elasticsearch.persistent.PersistentTasksClusterService;
 import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedState;
+import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsState;
+import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsTaskState;
 import org.elasticsearch.xpack.core.ml.job.config.JobState;
 import org.elasticsearch.xpack.core.ml.job.config.JobTaskState;
 
@@ -23,9 +25,11 @@ public final class MlTasks {
 
     public static final String JOB_TASK_NAME = "xpack/ml/job";
     public static final String DATAFEED_TASK_NAME = "xpack/ml/datafeed";
+    public static final String DATA_FRAME_ANALYTICS_TASK_NAME = "xpack/ml/data_frame/analytics";
 
     public static final String JOB_TASK_ID_PREFIX = "job-";
     public static final String DATAFEED_TASK_ID_PREFIX = "datafeed-";
+    private static final String DATA_FRAME_ANALYTICS_TASK_ID_PREFIX = "data_frame_analytics-";
 
     public static final PersistentTasksCustomMetaData.Assignment AWAITING_UPGRADE =
         new PersistentTasksCustomMetaData.Assignment(null,
@@ -50,6 +54,13 @@ public static String datafeedTaskId(String datafeedId) {
         return DATAFEED_TASK_ID_PREFIX + datafeedId;
     }
 
+    /**
+     * Namespaces the task ids for data frame analytics.
+     */
+    public static String dataFrameAnalyticsTaskId(String id) {
+        return DATA_FRAME_ANALYTICS_TASK_ID_PREFIX + id;
+    }
+
     @Nullable
     public static PersistentTasksCustomMetaData.PersistentTask<?> getJobTask(String jobId, @Nullable PersistentTasksCustomMetaData tasks) {
         return tasks == null ? null : tasks.getTask(jobTaskId(jobId));
@@ -61,6 +72,12 @@ public static PersistentTasksCustomMetaData.PersistentTask<?> getDatafeedTask(St
         return tasks == null ? null : tasks.getTask(datafeedTaskId(datafeedId));
     }
 
+    @Nullable
+    public static PersistentTasksCustomMetaData.PersistentTask<?> getDataFrameAnalyticsTask(String analyticsId,
+                                                                                            @Nullable PersistentTasksCustomMetaData tasks) {
+        return tasks == null ? null : tasks.getTask(dataFrameAnalyticsTaskId(analyticsId));
+    }
+
     /**
      * Note that the return value of this method does NOT take node relocations into account.
      * Use {@link #getJobStateModifiedForReassignments} to return a value adjusted to the most
@@ -120,6 +137,16 @@ public static DatafeedState getDatafeedState(String datafeedId, @Nullable Persis
         }
     }
 
+    public static DataFrameAnalyticsState getDataFrameAnalyticsState(String analyticsId, @Nullable PersistentTasksCustomMetaData tasks) {
+        PersistentTasksCustomMetaData.PersistentTask<?> task = getDataFrameAnalyticsTask(analyticsId, tasks);
+        if (task != null && task.getState() != null) {
+            DataFrameAnalyticsTaskState taskState = (DataFrameAnalyticsTaskState) task.getState();
+            return taskState.getState();
+        } else {
+            return DataFrameAnalyticsState.STOPPED;
+        }
+    }
+
     /**
      * The job Ids of anomaly detector job tasks.
      * All anomaly detector jobs are returned regardless of the status of the

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

@@ -0,0 +1,75 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.core.ml.action;
+
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.xpack.core.ml.action.util.PageParams;
+
+import java.io.IOException;
+import java.util.Objects;
+
+public abstract class AbstractGetResourcesRequest extends ActionRequest {
+
+    private String resourceId;
+    private PageParams pageParams = PageParams.defaultParams();
+
+    public void setResourceId(String resourceId) {
+        this.resourceId = resourceId;
+    }
+
+    public String getResourceId() {
+        return resourceId;
+    }
+
+    public void setPageParams(PageParams pageParams) {
+        this.pageParams = pageParams;
+    }
+
+    public PageParams getPageParams() {
+        return pageParams;
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        return null;
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        super.readFrom(in);
+        resourceId = in.readOptionalString();
+        pageParams = in.readOptionalWriteable(PageParams::new);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeOptionalString(resourceId);
+        out.writeOptionalWriteable(pageParams);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(resourceId, pageParams);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj == null) {
+            return false;
+        }
+        if (obj instanceof AbstractGetResourcesRequest == false) {
+            return false;
+        }
+        AbstractGetResourcesRequest other = (AbstractGetResourcesRequest) obj;
+        return Objects.equals(resourceId, other.resourceId);
+    }
+
+    public abstract String getResourceIdField();
+}

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

@@ -0,0 +1,84 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.core.ml.action;
+
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.xcontent.StatusToXContentObject;
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.xpack.core.ml.action.util.QueryPage;
+
+import java.io.IOException;
+import java.util.Objects;
+
+public abstract class AbstractGetResourcesResponse<T extends ToXContent & Writeable> extends ActionResponse
+    implements StatusToXContentObject {
+
+    private QueryPage<T> resources;
+
+    protected AbstractGetResourcesResponse() {}
+
+    protected AbstractGetResourcesResponse(QueryPage<T> resources) {
+        this.resources = Objects.requireNonNull(resources);
+    }
+
+    public QueryPage<T> getResources() {
+        return resources;
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        super.readFrom(in);
+        resources = new QueryPage<>(in, getReader());
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        resources.writeTo(out);
+    }
+
+    @Override
+    public RestStatus status() {
+        return RestStatus.OK;
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        resources.doXContentBody(builder, params);
+        builder.endObject();
+        return builder;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(resources);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj == null) {
+            return false;
+        }
+        if (obj instanceof AbstractGetResourcesResponse == false) {
+            return false;
+        }
+        AbstractGetResourcesResponse other = (AbstractGetResourcesResponse) obj;
+        return Objects.equals(resources, other.resources);
+    }
+
+    @Override
+    public final String toString() {
+        return Strings.toString(this);
+    }
+    protected abstract Reader<T> getReader();
+}

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

@@ -0,0 +1,81 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.core.ml.action;
+
+import org.elasticsearch.action.Action;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.support.master.AcknowledgedRequest;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.action.support.master.MasterNodeOperationRequestBuilder;
+import org.elasticsearch.client.ElasticsearchClient;
+import org.elasticsearch.common.xcontent.ToXContentFragment;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.xpack.core.ml.dataframe.DataFrameAnalyticsConfig;
+import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
+
+import java.io.IOException;
+import java.util.Objects;
+
+public class DeleteDataFrameAnalyticsAction extends Action<AcknowledgedResponse> {
+
+    public static final DeleteDataFrameAnalyticsAction INSTANCE = new DeleteDataFrameAnalyticsAction();
+    public static final String NAME = "cluster:admin/xpack/ml/data_frame/analytics/delete";
+
+    private DeleteDataFrameAnalyticsAction() {
+        super(NAME);
+    }
+
+    @Override
+    public AcknowledgedResponse newResponse() {
+        return new AcknowledgedResponse();
+    }
+
+    public static class Request extends AcknowledgedRequest<Request> implements ToXContentFragment {
+
+        private String id;
+
+        public Request() {}
+
+        public Request(String id) {
+            this.id = ExceptionsHelper.requireNonNull(id, DataFrameAnalyticsConfig.ID);
+        }
+
+        public String getId() {
+            return id;
+        }
+
+        @Override
+        public ActionRequestValidationException validate() {
+            return null;
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.field(DataFrameAnalyticsConfig.ID.getPreferredName(), id);
+            return builder;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            DeleteDataFrameAnalyticsAction.Request request = (DeleteDataFrameAnalyticsAction.Request) o;
+            return Objects.equals(id, request.id);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(id);
+        }
+    }
+
+    public static class RequestBuilder extends MasterNodeOperationRequestBuilder<Request, AcknowledgedResponse, RequestBuilder> {
+
+        protected RequestBuilder(ElasticsearchClient client, DeleteDataFrameAnalyticsAction action) {
+            super(client, action, new Request());
+        }
+    }
+}