HLRC: migration api - upgrade

client/rest-high-level/src/main/java/org/elasticsearch/client/MigrationClient.java

@@ -21,6 +21,11 @@
 
 import org.elasticsearch.client.migration.IndexUpgradeInfoRequest;
 import org.elasticsearch.client.migration.IndexUpgradeInfoResponse;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.client.migration.IndexUpgradeSubmissionResponse;
+import org.elasticsearch.index.reindex.BulkByScrollResponse;
+import org.elasticsearch.client.migration.IndexUpgradeRequest;
+
 
 import java.io.IOException;
 import java.util.Collections;
@@ -52,4 +57,20 @@ public IndexUpgradeInfoResponse getAssistance(IndexUpgradeInfoRequest request, R
         return restHighLevelClient.performRequestAndParseEntity(request, MigrationRequestConverters::getMigrationAssistance, options,
             IndexUpgradeInfoResponse::fromXContent, Collections.emptySet());
     }
+
+    public BulkByScrollResponse upgrade(IndexUpgradeRequest request, RequestOptions options) throws IOException {
+        return restHighLevelClient.performRequestAndParseEntity(request, MigrationRequestConverters::migrate, options,
+            BulkByScrollResponse::fromXContent, Collections.emptySet());
+    }
+
+    public IndexUpgradeSubmissionResponse submitUpgradeTask(IndexUpgradeRequest request, RequestOptions options) throws IOException {
+        return restHighLevelClient.performRequestAndParseEntity(request, MigrationRequestConverters::submitMigrateTask, options,
+            IndexUpgradeSubmissionResponse::fromXContent, Collections.emptySet());
+    }
+
+
+    public void upgradeAsync(IndexUpgradeRequest request, RequestOptions options, ActionListener<BulkByScrollResponse> listener)  {
+        restHighLevelClient.performRequestAsyncAndParseEntity(request, MigrationRequestConverters::migrate, options,
+            BulkByScrollResponse::fromXContent, listener, Collections.emptySet());
+    }
 }

client/rest-high-level/src/main/java/org/elasticsearch/client/MigrationRequestConverters.java

@@ -20,6 +20,8 @@
 package org.elasticsearch.client;
 
 import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.elasticsearch.client.migration.IndexUpgradeRequest;
 import org.elasticsearch.client.migration.IndexUpgradeInfoRequest;
 
 final class MigrationRequestConverters {
@@ -36,4 +38,25 @@ static Request getMigrationAssistance(IndexUpgradeInfoRequest indexUpgradeInfoRe
         parameters.withIndicesOptions(indexUpgradeInfoRequest.indicesOptions());
         return request;
     }
+
+    static Request migrate(IndexUpgradeRequest indexUpgradeRequest) {
+        return prepareMigrateRequest(indexUpgradeRequest, true);
+    }
+
+    static Request submitMigrateTask(IndexUpgradeRequest indexUpgradeRequest) {
+        return prepareMigrateRequest(indexUpgradeRequest, false);
+    }
+
+    private static Request prepareMigrateRequest(IndexUpgradeRequest indexUpgradeRequest, boolean waitForCompletion) {
+        RequestConverters.EndpointBuilder endpointBuilder = new RequestConverters.EndpointBuilder()
+            .addPathPartAsIs("_xpack/migration/upgrade")
+            .addCommaSeparatedPathParts(indexUpgradeRequest.indices());
+        String endpoint = endpointBuilder.build();
+        Request request = new Request(HttpPost.METHOD_NAME, endpoint);
+
+        RequestConverters.Params params = new RequestConverters.Params(request);
+        params.withWaitForCompletion(waitForCompletion);
+
+        return request;
+    }
 }

client/rest-high-level/src/main/java/org/elasticsearch/client/RequestConverters.java

@@ -886,10 +886,7 @@ Params withDetailed(boolean detailed) {
         }
 
         Params withWaitForCompletion(boolean waitForCompletion) {
-            if (waitForCompletion) {
-                return putParam("wait_for_completion", Boolean.TRUE.toString());
-            }
-            return this;
+            return putParam("wait_for_completion", Boolean.toString(waitForCompletion));
         }
 
         Params withNodes(String[] nodes) {

client/rest-high-level/src/main/java/org/elasticsearch/client/migration/IndexUpgradeInfoRequest.java

@@ -26,6 +26,10 @@
 import java.util.Arrays;
 import java.util.Objects;
 
+/**
+ * A request for retrieving upgrade information
+ * Part of Migration API
+ */
 public class IndexUpgradeInfoRequest extends TimedRequest implements IndicesRequest.Replaceable {
 
     private String[] indices = Strings.EMPTY_ARRAY;

client/rest-high-level/src/main/java/org/elasticsearch/client/migration/IndexUpgradeInfoResponse.java

@@ -28,6 +28,9 @@
 
 import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg;
 
+/**
+ * Response object that contains information about indices to be upgraded
+ */
 public class IndexUpgradeInfoResponse {
 
     private static final ParseField INDICES = new ParseField("indices");

client/rest-high-level/src/main/java/org/elasticsearch/client/migration/IndexUpgradeRequest.java

@@ -0,0 +1,96 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client.migration;
+
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.IndicesRequest;
+import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.action.support.master.MasterNodeReadRequest;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Objects;
+
+/**
+ * A request for performing Upgrade on Index
+ * Part of Migration API
+ */
+public class IndexUpgradeRequest extends MasterNodeReadRequest<IndexUpgradeRequest> implements IndicesRequest {
+
+    private String[] indices ;
+    private IndicesOptions indicesOptions = IndicesOptions.fromOptions(false, true, true, true);
+
+    public IndexUpgradeRequest(String index) {
+        indices = new String[]{index};
+    }
+
+    public IndexUpgradeRequest(StreamInput in) throws IOException {
+        super(in);
+        indices = in.readStringArray();
+        indicesOptions = IndicesOptions.readIndicesOptions(in);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeStringArray(indices);
+        indicesOptions.writeIndicesOptions(out);
+    }
+
+    @Override
+    public String[] indices() {
+        return indices;
+    }
+
+
+    @Override
+    public IndicesOptions indicesOptions() {
+        return indicesOptions;
+    }
+
+    public void indicesOptions(IndicesOptions indicesOptions) {
+        this.indicesOptions = indicesOptions;
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        return null;
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        throw new UnsupportedOperationException("usage of Streamable is to be replaced by Writeable");
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        IndexUpgradeRequest request = (IndexUpgradeRequest) o;
+        return Arrays.equals(indices, request.indices) &&
+                Objects.equals(indicesOptions.toString(), request.indicesOptions.toString());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(Arrays.hashCode(indices), indicesOptions.toString());
+    }
+}

client/rest-high-level/src/main/java/org/elasticsearch/client/migration/IndexUpgradeSubmissionResponse.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client.migration;
+
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ObjectParser;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.tasks.TaskId;
+
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * Response object that contains the taskID from submitted IndexUpgradeRequest
+ */
+public class IndexUpgradeSubmissionResponse {
+
+    private static final ParseField TASK = new ParseField("task");
+
+    public static final ConstructingObjectParser<IndexUpgradeSubmissionResponse, Void> PARSER = new ConstructingObjectParser<>(
+        "upgrade_submission_response",
+            true, a-> new IndexUpgradeSubmissionResponse( (TaskId) a[0]));
+
+    static {
+        PARSER.declareField(ConstructingObjectParser.optionalConstructorArg(), TaskId.parser(), TASK, ObjectParser.ValueType.STRING);
+    }
+
+    public static IndexUpgradeSubmissionResponse fromXContent(XContentParser parser) throws IOException {
+        return PARSER.parse(parser, null);
+    }
+
+    private final TaskId task;
+
+    IndexUpgradeSubmissionResponse(@Nullable TaskId task) {
+        this.task = task;
+    }
+
+    /**
+     * Get the task id
+     * @return the id of the upgrade task.
+     */
+    public TaskId getTask() {
+        return task;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(task);
+    }
+
+    @Override
+    public boolean equals(Object other) {
+        if (this == other) {
+            return true;
+        }
+
+        if (other == null || getClass() != other.getClass()) {
+            return false;
+        }
+
+        IndexUpgradeSubmissionResponse that = (IndexUpgradeSubmissionResponse) other;
+        return Objects.equals(task, that.task);
+    }
+}

client/rest-high-level/src/test/java/org/elasticsearch/client/MigrationIT.java

@@ -19,25 +19,73 @@
 
 package org.elasticsearch.client;
 
-import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
+import org.elasticsearch.ElasticsearchStatusException;
 import org.elasticsearch.client.migration.IndexUpgradeInfoRequest;
 import org.elasticsearch.client.migration.IndexUpgradeInfoResponse;
+import org.elasticsearch.client.migration.IndexUpgradeRequest;
+import org.elasticsearch.client.migration.IndexUpgradeSubmissionResponse;
+import org.elasticsearch.common.settings.Settings;
 
 import java.io.IOException;
+import java.util.function.BooleanSupplier;
+
+import static org.hamcrest.Matchers.containsString;
 
 public class MigrationIT extends ESRestHighLevelClientTestCase {
 
     public void testGetAssistance() throws IOException {
-        RestHighLevelClient client = highLevelClient();
         {
-            IndexUpgradeInfoResponse response = client.migration().getAssistance(new IndexUpgradeInfoRequest(), RequestOptions.DEFAULT);
+            IndexUpgradeInfoResponse response = highLevelClient().migration()
+                .getAssistance(new IndexUpgradeInfoRequest(), RequestOptions.DEFAULT);
             assertEquals(0, response.getActions().size());
         }
         {
-            client.indices().create(new CreateIndexRequest("test"), RequestOptions.DEFAULT);
-            IndexUpgradeInfoResponse response = client.migration().getAssistance(
+            createIndex("test", Settings.EMPTY);
+            IndexUpgradeInfoResponse response = highLevelClient().migration().getAssistance(
                 new IndexUpgradeInfoRequest("test"), RequestOptions.DEFAULT);
             assertEquals(0, response.getActions().size());
         }
     }
+
+    public void testUpgradeWhenIndexCannotBeUpgraded() throws IOException {
+        createIndex("test", Settings.EMPTY);
+
+        ThrowingRunnable execute = () -> execute(new IndexUpgradeRequest("test"),
+            highLevelClient().migration()::upgrade,
+            highLevelClient().migration()::upgradeAsync);
+
+        ElasticsearchStatusException responseException = expectThrows(ElasticsearchStatusException.class, execute);
+
+        assertThat(responseException.getDetailedMessage(), containsString("cannot be upgraded"));
+    }
+
+    public void testUpgradeWithTaskApi() throws IOException, InterruptedException {
+        createIndex("test", Settings.EMPTY);
+
+        IndexUpgradeRequest request = new IndexUpgradeRequest("test");
+
+        IndexUpgradeSubmissionResponse upgrade = highLevelClient().migration()
+            .submitUpgradeTask(request, RequestOptions.DEFAULT);
+
+        assertNotNull(upgrade.getTask());
+
+        BooleanSupplier hasUpgradeCompleted = checkCompletionStatus(upgrade);
+        awaitBusy(hasUpgradeCompleted);
+    }
+
+    /**
+     * Using low-level api as high-level-rest-client's getTaskById work is in progress.
+     * TODO revisit once that work is finished
+     */
+    private BooleanSupplier checkCompletionStatus(IndexUpgradeSubmissionResponse upgrade) {
+        return () -> {
+            try {
+                Response response = client().performRequest(new Request("GET", "/_tasks/" + upgrade.getTask().toString()));
+                return (boolean) entityAsMap(response).get("completed");
+            } catch (IOException e) {
+                fail(e.getMessage());
+                return false;
+            }
+        };
+    }
 }