[7.x][ML] Reset anomaly detection job API (#73908) (#74093)

Adds a new API that allows a user to reset
an anomaly detection job.

To use the API do:

```
POST _ml/anomaly_detectors/<job_id>_reset
```

The API removes all data associated to the job.
In particular, it deletes model state, results and stats.

However, job notifications and user annotations are not removed.

Also, the API can be called asynchronously by setting the parameter
`wait_for_completion` to `false` (defaults to `true`). When run
that way the API returns the task id for further monitoring.

In order to prevent the job from opening while it is resetting,
a new job field has been added called `blocked`. It is an object
that contains a `reason` and the `task_id`. `reason` can take
a value from ["delete", "reset", "revert"] as all these
operations should block the job from opening. The `task_id` is also
included in order to allow tracking the task if necessary.

Finally, this commit also sets the `blocked` field when
the revert snapshot API is called as a job should not be opened
while it is reverted to a different model snapshot.

Backport of #73908

Author: dimitris-athanasiou

docs/reference/ml/anomaly-detection/apis/get-job.asciidoc

@@ -59,6 +59,24 @@ include::{es-repo-dir}/ml/ml-shared.asciidoc[tag=exclude-generated]
 The API returns an array of {anomaly-job} resources. For the full list of
 properties, see <<ml-put-job-request-body,create {anomaly-jobs} API>>.
 
+//Begin blocked
+`blocked`::
+(object) When present, it explains that a task is executed on the job
+that blocks it from opening.
++
+.Properties of `blocked`
+[%collapsible%open]
+====
+`reason`:::
+(string) The reason the job is blocked. Values may be `delete`, `reset`, `revert`.
+Each value means the corresponding action is being executed.
+
+`task_id`:::
+(string) The task id of the blocking action. You can use the <<tasks>> API to
+monitor progress.
+====
+//End blocked
+
 `create_time`::
 (string) The time the job was created. For example, `1491007356077`. This
 property is informational; you cannot change its value.

docs/reference/ml/anomaly-detection/apis/index.asciidoc

@@ -48,6 +48,8 @@ include::open-job.asciidoc[leveloffset=+2]
 include::post-data.asciidoc[leveloffset=+2]
 //PREVIEW
 include::preview-datafeed.asciidoc[leveloffset=+2]
+//RESET
+include::reset-job.asciidoc[leveloffset=+2]
 //REVERT
 include::revert-snapshot.asciidoc[leveloffset=+2]
 //SET/START/STOP

docs/reference/ml/anomaly-detection/apis/ml-apis.asciidoc

@@ -19,6 +19,7 @@ See also <<ml-df-analytics-apis>>.
 * <<ml-post-data,Post data to {anomaly-jobs}>>
 * <<ml-update-job,Update {anomaly-jobs}>>
 * <<ml-forecast,Create>> or <<ml-delete-forecast,delete forecasts>>
+* <<ml-reset-job,Reset {anomaly-jobs}>> 
 
 
 [discrete]

docs/reference/ml/anomaly-detection/apis/reset-job.asciidoc

@@ -0,0 +1,86 @@
+[role="xpack"]
+[testenv="platinum"]
+[[ml-reset-job]]
+= Reset {anomaly-jobs} API
+++++
+<titleabbrev>Reset jobs</titleabbrev>
+++++
+
+Resets an existing {anomaly-job}.
+
+[[ml-reset-job-request]]
+== {api-request-title}
+
+`POST _ml/anomaly_detectors/<job_id>/_reset`
+
+[[ml-reset-job-prereqs]]
+== {api-prereq-title}
+
+* Requires the `manage_ml` cluster privilege. This privilege is included in the 
+`machine_learning_admin` built-in role.
+* Before you can reset a job, you must close it. You can set `force` to `true`
+when closing the job to avoid waiting for the job to complete. See 
+<<ml-close-job>>.
+
+[[ml-reset-job-desc]]
+== {api-description-title}
+
+All model state and results are deleted.
+The job is ready to start over as if it had just been created.
+
+It is not currently possible to reset multiple jobs using wildcards or a comma
+separated list.
+
+[[ml-reset-job-path-parms]]
+== {api-path-parms-title}
+
+`<job_id>`::
+(Required, string)
+include::{es-repo-dir}/ml/ml-shared.asciidoc[tag=job-id-anomaly-detection]
+
+[[ml-reset-job-query-parms]]
+== {api-query-parms-title}
+
+`wait_for_completion`::
+  (Optional, Boolean) Specifies whether the request should return immediately or
+  wait until the job reset completes. Defaults to `true`.
+
+[[ml-reset-job-example]]
+== {api-examples-title}
+
+[source,console]
+--------------------------------------------------
+POST _ml/anomaly_detectors/total-requests/_reset
+--------------------------------------------------
+// TEST[skip:setup:server_metrics_job]
+
+When the job is reset, you receive the following results:
+
+[source,console-result]
+----
+{
+  "acknowledged": true
+}
+----
+
+In the next example we reset the `total-requests` job asynchronously:
+
+[source,console]
+--------------------------------------------------
+POST _ml/anomaly_detectors/total-requests/_reset?wait_for_completion=false
+--------------------------------------------------
+// TEST[skip:setup:server_metrics_job]
+
+When `wait_for_completion` is set to `false`, the response contains the id
+of the job reset task:
+
+[source,console-result]
+----
+{
+  "task": "oTUltX4IQMOUUVeiohTt8A:39"
+}
+----
+// TESTRESPONSE[s/"task": "oTUltX4IQMOUUVeiohTt8A:39"/"task": $body.task/]
+
+If you want to check the status of the reset task, use the <<tasks>> by referencing 
+the task ID.

rest-api-spec/src/main/resources/rest-api-spec/api/ml.reset_job.json

@@ -0,0 +1,36 @@
+{
+  "ml.reset_job":{
+    "documentation":{
+      "url":"https://www.elastic.co/guide/en/elasticsearch/reference/current/ml-reset-job.html",
+      "description":"Resets an existing anomaly detection job."
+    },
+    "stability":"stable",
+    "visibility":"public",
+    "headers":{
+      "accept": [ "application/json"]
+    },
+    "url":{
+      "paths":[
+        {
+          "path":"/_ml/anomaly_detectors/{job_id}/_reset",
+          "methods":[
+            "POST"
+          ],
+          "parts":{
+            "job_id":{
+              "type":"string",
+              "description":"The ID of the job to reset"
+            }
+          }
+        }
+      ]
+    },
+    "params":{
+      "wait_for_completion":{
+        "type":"boolean",
+        "description":"Should this request wait until the operation has completed before returning",
+        "default":true
+      }
+    }
+  }
+}

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

@@ -0,0 +1,118 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.ml.action;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.support.master.AcknowledgedRequest;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.tasks.CancellableTask;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
+import org.elasticsearch.xpack.core.ml.MlTasks;
+import org.elasticsearch.xpack.core.ml.job.config.Job;
+import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Objects;
+
+public class ResetJobAction extends ActionType<AcknowledgedResponse> {
+
+    public static final String NAME = "cluster:admin/xpack/ml/job/reset";
+    public static final ResetJobAction INSTANCE = new ResetJobAction();
+
+    public static final Version VERSION_INTRODUCED = Version.V_7_14_0;
+
+    private ResetJobAction() {
+        super(NAME, AcknowledgedResponse::readFrom);
+    }
+
+    public static class Request extends AcknowledgedRequest<Request> {
+
+        private String jobId;
+
+        /**
+         * Internal parameter that allows resetting an open job
+         * when a job is reallocated to a new node.
+         */
+        private boolean skipJobStateValidation;
+
+        /**
+         * Should this task store its result?
+         */
+        private boolean shouldStoreResult;
+
+        public Request(String jobId) {
+            this.jobId = ExceptionsHelper.requireNonNull(jobId, Job.ID);
+        }
+
+        public Request(StreamInput in) throws IOException {
+            super(in);
+            jobId = in.readString();
+            skipJobStateValidation = in.readBoolean();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeString(jobId);
+            out.writeBoolean(skipJobStateValidation);
+        }
+
+        public void setSkipJobStateValidation(boolean skipJobStateValidation) {
+            this.skipJobStateValidation = skipJobStateValidation;
+        }
+
+        public boolean isSkipJobStateValidation() {
+            return skipJobStateValidation;
+        }
+
+        /**
+         * Should this task store its result after it has finished?
+         */
+        public void setShouldStoreResult(boolean shouldStoreResult) {
+            this.shouldStoreResult = shouldStoreResult;
+        }
+
+        @Override
+        public boolean getShouldStoreResult() {
+            return shouldStoreResult;
+        }
+
+        @Override
+        public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+            return new CancellableTask(id, type, action, MlTasks.JOB_TASK_ID_PREFIX + jobId, parentTaskId, headers);
+        }
+
+        @Override
+        public ActionRequestValidationException validate() {
+            return null;
+        }
+
+        public String getJobId() {
+            return jobId;
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(jobId, skipJobStateValidation);
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || o.getClass() != getClass()) return false;
+            Request that = (Request) o;
+            return Objects.equals(jobId, that.jobId) && skipJobStateValidation == that.skipJobStateValidation;
+        }
+    }
+}

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

@@ -0,0 +1,111 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.ml.job.config;
+
+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.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ParseField;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.tasks.TaskId;
+
+import java.io.IOException;
+import java.util.Locale;
+import java.util.Objects;
+
+public class Blocked implements ToXContentObject, Writeable {
+
+    public enum Reason {
+        NONE, DELETE, RESET, REVERT;
+
+        public static Reason fromString(String value) {
+            return Reason.valueOf(value.toUpperCase(Locale.ROOT));
+        }
+
+        @Override
+        public String toString() {
+            return name().toLowerCase(Locale.ROOT);
+        }
+    }
+
+    public static final ParseField REASON = new ParseField("reason");
+    public static final ParseField TASK_ID = new ParseField("task_id");
+
+    public static final ConstructingObjectParser<Blocked, Void> LENIENT_PARSER = createParser(true);
+    public static final ConstructingObjectParser<Blocked, Void> STRICT_PARSER = createParser(false);
+
+    private static ConstructingObjectParser<Blocked, Void> createParser(boolean ignoreUnknownFields) {
+        ConstructingObjectParser<Blocked, Void> parser = new ConstructingObjectParser<>("blocked", ignoreUnknownFields,
+            a -> new Blocked((Reason) a[0], (TaskId) a[1]));
+        parser.declareString(ConstructingObjectParser.constructorArg(), Reason::fromString, REASON);
+        parser.declareString(ConstructingObjectParser.optionalConstructorArg(), TaskId::new, TASK_ID);
+        return parser;
+    }
+
+    private final Reason reason;
+
+    @Nullable
+    private final TaskId taskId;
+
+    public Blocked(Reason reason, @Nullable TaskId taskId) {
+        this.reason = Objects.requireNonNull(reason);
+        this.taskId = taskId;
+    }
+
+    public Blocked(StreamInput in) throws IOException {
+        this.reason = in.readEnum(Reason.class);
+        this.taskId = in.readOptionalWriteable(TaskId::readFromStream);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeEnum(reason);
+        out.writeOptionalWriteable(taskId);
+    }
+
+    public Reason getReason() {
+        return reason;
+    }
+
+    @Nullable
+    public TaskId getTaskId() {
+        return taskId;
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        builder.field(REASON.getPreferredName(), reason);
+        if (taskId != null) {
+            builder.field(TASK_ID.getPreferredName(), taskId.toString());
+        }
+        builder.endObject();
+        return builder;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(reason, taskId);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        Blocked that = (Blocked) o;
+        return Objects.equals(reason, that.reason) && Objects.equals(taskId, that.taskId);
+    }
+
+    public static Blocked none() {
+        return new Blocked(Reason.NONE, null);
+    }
+}