ML: add migrate anomalies assistant (#36643)

* ML: add migrate anomalies assistant * adjusting failure handling for reindex * Fixing request and tests * Adding tests to blacklist * adjusting test * test fix: posting data directly to the job instead of relying on datafeed * adjusting API usage * adding Todos and adjusting endpoint * Adding types to reindexRequest * removing unreliable "live" data test * adding index refresh to test * adding index refresh to test * adding index refresh to yaml test * fixing bad exists call * removing todo * Addressing remove comments * Adjusting rest endpoint name * making service have its own logger * adjusting validity check for newindex names * fixing typos * fixing renaming
elastic · Jan 10, 2019 · dcf9d3c · dcf9d3c
1 parent 9ec27fd
commit dcf9d3c
Show file tree

Hide file tree

Showing 22 changed files with 1,826 additions and 111 deletions.
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackClientPlugin.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackClientPlugin.java
@@ -104,6 +104,7 @@
 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.MlUpgradeAction;
 import org.elasticsearch.xpack.core.ml.action.RevertModelSnapshotAction;
 import org.elasticsearch.xpack.core.ml.action.StartDatafeedAction;
 import org.elasticsearch.xpack.core.ml.action.StopDatafeedAction;
@@ -289,6 +290,7 @@ public List<GenericAction> getClientActions() {
                 PostCalendarEventsAction.INSTANCE,
                 PersistJobAction.INSTANCE,
                 FindFileStructureAction.INSTANCE,
+                MlUpgradeAction.INSTANCE,
                 // security
                 ClearRealmCacheAction.INSTANCE,
                 ClearRolesCacheAction.INSTANCE,

diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/MlUpgradeAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/MlUpgradeAction.java
@@ -0,0 +1,160 @@
+/*
+ * 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.IndicesOptions;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.action.support.master.MasterNodeReadOperationRequestBuilder;
+import org.elasticsearch.action.support.master.MasterNodeReadRequest;
+import org.elasticsearch.client.ElasticsearchClient;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ObjectParser;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.tasks.CancellableTask;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
+import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndexFields;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Objects;
+
+
+public class MlUpgradeAction extends Action<AcknowledgedResponse> {
+    public static final MlUpgradeAction INSTANCE = new MlUpgradeAction();
+    public static final String NAME = "cluster:admin/xpack/ml/upgrade";
+
+    private MlUpgradeAction() {
+        super(NAME);
+    }
+
+    @Override
+    public AcknowledgedResponse newResponse() {
+        return new AcknowledgedResponse();
+    }
+
+    public static class Request extends MasterNodeReadRequest<Request> implements ToXContentObject {
+
+        private static final ParseField REINDEX_BATCH_SIZE = new ParseField("reindex_batch_size");
+
+        public static ObjectParser<Request, Void> PARSER = new ObjectParser<>("ml_upgrade", true, Request::new);
+        static {
+            PARSER.declareInt(Request::setReindexBatchSize, REINDEX_BATCH_SIZE);
+        }
+
+        static final String INDEX = AnomalyDetectorsIndexFields.RESULTS_INDEX_PREFIX + "*";
+        private int reindexBatchSize = 1000;
+
+        /**
+         * Should this task store its result?
+         */
+        private boolean shouldStoreResult;
+
+        // for serialization
+        public Request() {
+        }
+
+        public Request(StreamInput in) throws IOException {
+            super(in);
+            reindexBatchSize = in.readInt();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeInt(reindexBatchSize);
+        }
+
+        public String[] indices() {
+            return new String[]{INDEX};
+        }
+
+        public IndicesOptions indicesOptions() {
+            return IndicesOptions.strictExpandOpenAndForbidClosed();
+        }
+
+        /**
+         * Should this task store its result after it has finished?
+         */
+        public Request setShouldStoreResult(boolean shouldStoreResult) {
+            this.shouldStoreResult = shouldStoreResult;
+            return this;
+        }
+
+        @Override
+        public boolean getShouldStoreResult() {
+            return shouldStoreResult;
+        }
+
+        public Request setReindexBatchSize(int reindexBatchSize) {
+            this.reindexBatchSize = reindexBatchSize;
+            return this;
+        }
+
+        public int getReindexBatchSize() {
+            return reindexBatchSize;
+        }
+
+        @Override
+        public ActionRequestValidationException validate() {
+            if (reindexBatchSize <= 0) {
+                ActionRequestValidationException validationException = new ActionRequestValidationException();
+                validationException.addValidationError("["+ REINDEX_BATCH_SIZE.getPreferredName()+"] must be greater than 0.");
+                return validationException;
+            }
+            return null;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) {
+                return true;
+            }
+            if (o == null || getClass() != o.getClass()) {
+                return false;
+            }
+
+            Request request = (Request) o;
+            return Objects.equals(reindexBatchSize, request.reindexBatchSize);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(reindexBatchSize);
+        }
+
+        @Override
+        public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+            return new CancellableTask(id, type, action, "ml-upgrade", parentTaskId, headers) {
+                @Override
+                public boolean shouldCancelChildrenOnCancellation() {
+                    return true;
+                }
+            };
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.startObject();
+            builder.field(REINDEX_BATCH_SIZE.getPreferredName(), reindexBatchSize);
+            builder.endObject();
+            return builder;
+        }
+    }
+
+    public static class RequestBuilder extends MasterNodeReadOperationRequestBuilder<Request, AcknowledgedResponse, RequestBuilder> {
+
+        public RequestBuilder(ElasticsearchClient client) {
+            super(client, INSTANCE, new Request());
+        }
+    }
+
+}
diff --git a/...ugin/core/src/test/java/org/elasticsearch/xpack/core/ml/action/MlUpgradeRequestTests.java b/...ugin/core/src/test/java/org/elasticsearch/xpack/core/ml/action/MlUpgradeRequestTests.java
@@ -0,0 +1,28 @@
+/*
+ * 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.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
+
+
+public class MlUpgradeRequestTests extends AbstractWireSerializingTestCase<MlUpgradeAction.Request> {
+
+    @Override
+    protected MlUpgradeAction.Request createTestInstance() {
+        MlUpgradeAction.Request request = new MlUpgradeAction.Request();
+        if (randomBoolean()) {
+            request.setReindexBatchSize(randomIntBetween(1, 10_000));
+        }
+        return request;
+    }
+
+    @Override
+    protected Writeable.Reader<MlUpgradeAction.Request> instanceReader() {
+        return MlUpgradeAction.Request::new;
+    }
+
+}
diff --git a/x-pack/plugin/ml/qa/ml-with-security/build.gradle b/x-pack/plugin/ml/qa/ml-with-security/build.gradle
@@ -93,7 +93,9 @@ integTestRunner {
     'ml/validate/Test job config that is invalid only because of the job ID',
     'ml/validate_detector/Test invalid detector',
     'ml/delete_forecast/Test delete on _all forecasts not allow no forecasts',
-    'ml/delete_forecast/Test delete forecast on missing forecast'
+    'ml/delete_forecast/Test delete forecast on missing forecast',
+    'ml/ml_upgrade/Upgrade results when there is nothing to upgrade',
+    'ml/ml_upgrade/Upgrade results when there is nothing to upgrade not waiting for results'
   ].join(',')
 }
 

diff --git a/...src/test/java/org/elasticsearch/xpack/ml/integration/MlNativeAutodetectIntegTestCase.java b/...src/test/java/org/elasticsearch/xpack/ml/integration/MlNativeAutodetectIntegTestCase.java
@@ -26,6 +26,7 @@
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.index.reindex.ReindexPlugin;
 import org.elasticsearch.persistent.PersistentTaskParams;
 import org.elasticsearch.persistent.PersistentTaskState;
 import org.elasticsearch.plugins.Plugin;
@@ -120,7 +121,7 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
 
     @Override
     protected Collection<Class<? extends Plugin>> transportClientPlugins() {
-        return Arrays.asList(XPackClientPlugin.class, Netty4Plugin.class);
+        return Arrays.asList(XPackClientPlugin.class, Netty4Plugin.class, ReindexPlugin.class);
     }
 
     @Override