Add wait_for_completion parameter to resize&open&forcemerge APIs (ope…

…nsearch-project#6228)

Signed-off-by: Gao Binlong <gbinlong@amazon.com>

CHANGELOG.md

@@ -8,6 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Support for HTTP/2 (server-side) ([#3847](https://github.com/opensearch-project/OpenSearch/pull/3847))
 - Add getter for path field in NestedQueryBuilder ([#4636](https://github.com/opensearch-project/OpenSearch/pull/4636))
 - Allow mmap to use new JDK-19 preview APIs in Apache Lucene 9.4+ ([#5151](https://github.com/opensearch-project/OpenSearch/pull/5151))
+- Add wait_for_completion parameter to resize&open&forcemerge APIs
 
 ### Dependencies
 - Bumps `log4j-core` from 2.18.0 to 2.19.0

modules/reindex/src/main/java/org/opensearch/index/reindex/AbstractBaseReindexRestHandler.java

@@ -37,13 +37,9 @@
 import org.opensearch.action.support.ActiveShardCount;
 import org.opensearch.client.node.NodeClient;
 import org.opensearch.common.io.stream.NamedWriteableRegistry;
-import org.opensearch.core.xcontent.XContentBuilder;
 import org.opensearch.rest.BaseRestHandler;
-import org.opensearch.rest.BytesRestResponse;
 import org.opensearch.rest.RestRequest;
-import org.opensearch.rest.RestStatus;
 import org.opensearch.tasks.LoggingTaskListener;
-import org.opensearch.tasks.Task;
 
 import java.io.IOException;
 import java.util.HashMap;
@@ -124,17 +120,6 @@ protected Request setCommonOptions(RestRequest restRequest, Request request) {
         return request;
     }
 
-    private RestChannelConsumer sendTask(String localNodeId, Task task) {
-        return channel -> {
-            try (XContentBuilder builder = channel.newBuilder()) {
-                builder.startObject();
-                builder.field("task", localNodeId + ":" + task.getId());
-                builder.endObject();
-                channel.sendResponse(new BytesRestResponse(RestStatus.OK, builder));
-            }
-        };
-    }
-
     private static Integer parseSlices(RestRequest request) {
         String slicesString = request.param("slices");
         if (slicesString == null) {

rest-api-spec/src/main/resources/rest-api-spec/api/indices.clone.json

@@ -46,6 +46,14 @@
       "wait_for_active_shards": {
         "type" : "string",
         "description" : "Set the number of active shards to wait for on the cloned index before the operation returns."
+      },
+      "wait_for_completion": {
+        "type" : "boolean",
+        "description" : "If false, the request will return a task immediately and the operation will run in background. Defaults to true."
+      },
+      "task_execution_timeout": {
+        "type" : "time",
+        "description" : "Explicit task execution timeout, only useful when wait_for_completion is false, defaults to 1h."
       }
     },
     "body": {

rest-api-spec/src/main/resources/rest-api-spec/api/indices.forcemerge.json

@@ -59,6 +59,10 @@
       "only_expunge_deletes":{
         "type":"boolean",
         "description":"Specify whether the operation should only expunge deleted documents"
+      },
+      "wait_for_completion": {
+        "type" : "boolean",
+        "description" : "If false, the request will return a task immediately and the operation will run in background. Defaults to true."
       }
     }
   }

rest-api-spec/src/main/resources/rest-api-spec/api/indices.open.json

@@ -53,6 +53,14 @@
       "wait_for_active_shards":{
         "type":"string",
         "description":"Sets the number of active shards to wait for before the operation returns."
+      },
+      "wait_for_completion": {
+        "type" : "boolean",
+        "description" : "If false, the request will return a task immediately and the operation will run in background. Defaults to true."
+      },
+      "task_execution_timeout": {
+        "type" : "time",
+        "description" : "Explicit task execution timeout, only useful when wait_for_completion is false, defaults to 1h."
       }
     }
   }

rest-api-spec/src/main/resources/rest-api-spec/api/indices.shrink.json

@@ -50,6 +50,14 @@
       "wait_for_active_shards": {
         "type" : "string",
         "description" : "Set the number of active shards to wait for on the shrunken index before the operation returns."
+      },
+      "wait_for_completion": {
+        "type" : "boolean",
+        "description" : "If false, the request will return a task immediately and the operation will run in background. Defaults to true."
+      },
+      "task_execution_timeout": {
+        "type" : "time",
+        "description" : "Explicit task execution timeout, only useful when wait_for_completion is false, defaults to 1h."
       }
     },
     "body":{

rest-api-spec/src/main/resources/rest-api-spec/api/indices.split.json

@@ -50,6 +50,14 @@
       "wait_for_active_shards": {
         "type" : "string",
         "description" : "Set the number of active shards to wait for on the shrunken index before the operation returns."
+      },
+      "wait_for_completion": {
+        "type" : "boolean",
+        "description" : "If false, the request will return a task immediately and the operation will run in background. Defaults to true."
+      },
+      "task_execution_timeout": {
+        "type" : "time",
+        "description" : "Explicit task execution timeout, only useful when wait_for_completion is false, defaults to 1h."
       }
     },
     "body":{

rest-api-spec/src/main/resources/rest-api-spec/test/indices.clone/40_wait_for_completion.yml

@@ -0,0 +1,77 @@
+---
+"Clone index with wait_for_completion":
+  # clone index with wait_for_completion parameter, when the parameter is set to false, the API
+  # will return a task immediately and the clone operation will run in background.
+
+  - skip:
+      version: " - 2.9.99"
+      reason: "only available in 3.0+"
+      features: allowed_warnings
+
+  - do:
+      nodes.info:
+        node_id: data:true
+  - set:
+      nodes._arbitrary_key_: node_id
+
+  - do:
+      indices.create:
+        index: source
+        wait_for_active_shards: 1
+        body:
+          settings:
+            # ensure everything is allocated on the same data node
+            index.routing.allocation.include._id: $node_id
+            index.number_of_shards: 1
+            index.number_of_replicas: 0
+  - do:
+      index:
+        index: source
+        id:    "1"
+        body:  { "foo": "hello world" }
+
+  - do:
+      get:
+        index: source
+        id:    "1"
+
+  - match: { _index:   source }
+  - match: { _id:      "1"     }
+  - match: { _source:  { foo: "hello world" } }
+
+  # make it read-only
+  - do:
+      indices.put_settings:
+        index: source
+        body:
+          index.blocks.write: true
+          index.number_of_replicas: 0
+
+  - do:
+      cluster.health:
+        wait_for_status: green
+        index: source
+
+  # clone with wait_for_completion
+  - do:
+      allowed_warnings:
+        - "Parameter [master_timeout] is deprecated and will be removed in 3.0. To support inclusive language, please use [cluster_manager_timeout] instead."
+      indices.clone:
+        index: "source"
+        target: "new_cloned_index"
+        wait_for_active_shards: 1
+        master_timeout: 10s
+        wait_for_completion: false
+        body:
+          settings:
+            index.number_of_shards: 1
+            "index.number_of_replicas": 0
+
+  - match: { task: /^.+$/ }
+  - set: { task: taskId }
+
+  - do:
+      tasks.get:
+        task_id: $taskId
+  - match: { task.action: "indices:admin/resize" }
+  - match: { task.description: "clone from [source] to [new_cloned_index]" }

rest-api-spec/src/main/resources/rest-api-spec/test/indices.forcemerge/20_wait_for_completion.yml

@@ -0,0 +1,27 @@
+---
+"Force merge index with wait_for_completion":
+  # force merge index with wait_for_completion parameter, when the parameter is set to false, the API
+  # will return a task immediately and the merge process will run in background.
+
+  - skip:
+      version: " - 2.9.99"
+      reason: "only available in 3.0+"
+      features: allowed_warnings
+
+  - do:
+      indices.create:
+        index: test_index
+
+  - do:
+      indices.forcemerge:
+        index: test_index
+        wait_for_completion: false
+        max_num_segments: 1
+  - match: { task: /^.+$/ }
+  - set: { task: taskId }
+
+  - do:
+      tasks.get:
+        task_id: $taskId
+  - match: { task.action: "indices:admin/forcemerge" }
+  - match: { task.description: "Force-merge indices [test_index], maxSegments[1], onlyExpungeDeletes[false], flush[true]" }

rest-api-spec/src/main/resources/rest-api-spec/test/indices.open/30_wait_for_completion.yml

@@ -0,0 +1,36 @@
+---
+"Open index with wait_for_completion":
+  # open index with wait_for_completion parameter, when the parameter is set to false, the API
+  # will return a task immediately and the open operation will run in background.
+
+  - skip:
+      version: " - 2.9.99"
+      reason: "only available in 3.0+"
+      features: allowed_warnings
+
+  - do:
+      indices.create:
+        index: test_index
+        body:
+          settings:
+            number_of_replicas: 0
+            number_of_shards: 1
+
+  - do:
+      indices.close:
+        index: test_index
+  - is_true: acknowledged
+
+  - do:
+      indices.open:
+        index: test_index
+        wait_for_active_shards: all
+        wait_for_completion: false
+  - match: { task: /^.+$/ }
+  - set: { task: taskId }
+
+  - do:
+      tasks.get:
+        task_id: $taskId
+  - match: { task.action: "indices:admin/open" }
+  - match: { task.description: "open indices [test_index]" }

rest-api-spec/src/main/resources/rest-api-spec/test/indices.shrink/50_wait_for_completion.yml

@@ -0,0 +1,77 @@
+---
+"Shrink index with wait_for_completion":
+  # shrink index with wait_for_completion parameter, when the parameter is set to false, the API
+  # will return a task immediately and the shrink operation will run in background.
+
+  - skip:
+      version: " - 2.9.99"
+      reason: "only available in 3.0+"
+      features: allowed_warnings
+
+  - do:
+      nodes.info:
+        node_id: data:true
+  - set:
+      nodes._arbitrary_key_: node_id
+
+  - do:
+      indices.create:
+        index: source
+        wait_for_active_shards: 1
+        body:
+          settings:
+            # ensure everything is allocated on the same data node
+            index.routing.allocation.include._id: $node_id
+            index.number_of_shards: 3
+            index.number_of_replicas: 0
+  - do:
+      index:
+        index: source
+        id:    "1"
+        body:  { "foo": "hello world" }
+
+  - do:
+      get:
+        index: source
+        id:    "1"
+
+  - match: { _index:   source }
+  - match: { _id:      "1"     }
+  - match: { _source:  { foo: "hello world" } }
+
+  # make it read-only
+  - do:
+      indices.put_settings:
+        index: source
+        body:
+          index.blocks.write: true
+          index.number_of_replicas: 0
+
+  - do:
+      cluster.health:
+        wait_for_status: green
+        index: source
+
+  # shrink with wait_for_completion
+  - do:
+      allowed_warnings:
+        - "Parameter [master_timeout] is deprecated and will be removed in 3.0. To support inclusive language, please use [cluster_manager_timeout] instead."
+      indices.shrink:
+        index: "source"
+        target: "new_shrunken_index"
+        wait_for_active_shards: 1
+        master_timeout: 10s
+        wait_for_completion: false
+        body:
+          settings:
+            index.number_of_shards: 1
+            "index.number_of_replicas": 0
+
+  - match: { task: /^.+$/ }
+  - set: { task: taskId }
+
+  - do:
+      tasks.get:
+        task_id: $taskId
+  - match: { task.action: "indices:admin/resize" }
+  - match: { task.description: "shrink from [source] to [new_shrunken_index]" }