[CCR] added rest specs and simple rest test for follow and unfollow apis (#30123)

martijnvg · martijnvg · commit 8a43e7014fd8 · 2018-05-07T14:19:31.000+02:00
[CCR] added rest specs and simple rest test for follow and unfollow apis, also

Added an acknowledge field in follow and unfollow api responses. Currently these api return an empty response and fixed bug in unfollow api that didn't cleanup node tasks properly.
diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/FollowExistingIndexAction.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/FollowExistingIndexAction.java
@@ -10,10 +10,10 @@
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestBuilder;
 import org.elasticsearch.action.ActionRequestValidationException;
-import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.client.ElasticsearchClient;
 import org.elasticsearch.cluster.ClusterState;
@@ -139,8 +139,26 @@ public static class RequestBuilder extends ActionRequestBuilder<Request, Respons
         }
     }
 
-    public static class Response extends ActionResponse {
+    public static class Response extends AcknowledgedResponse {
 
+        Response() {
+        }
+
+        Response(boolean acknowledged) {
+            super(acknowledged);
+        }
+
+        @Override
+        public void readFrom(StreamInput in) throws IOException {
+            super.readFrom(in);
+            readAcknowledged(in);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            writeAcknowledged(out);
+        }
     }
 
     public static class TransportAction extends HandledTransportAction<Request, Response> {
@@ -261,7 +279,7 @@ void finalizeResponse() {
 
                                         if (error == null) {
                                             // include task ids?
-                                            handler.onResponse(new Response());
+                                            handler.onResponse(new Response(true));
                                         } else {
                                             // TODO: cancel all started tasks
                                             handler.onFailure(error);
diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowNodeTask.java
@@ -29,6 +29,11 @@ public class ShardFollowNodeTask extends AllocatedPersistentTask {
         super(id, type, action, description, parentTask, headers);
     }
 
+    @Override
+    protected void onCancelled() {
+        markAsCompleted();
+    }
+
     void updateProcessedGlobalCheckpoint(long processedGlobalCheckpoint) {
         this.processedGlobalCheckpoint.set(processedGlobalCheckpoint);
     }
diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/UnfollowIndexAction.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/UnfollowIndexAction.java
@@ -10,10 +10,10 @@
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestBuilder;
 import org.elasticsearch.action.ActionRequestValidationException;
-import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.client.ElasticsearchClient;
 import org.elasticsearch.cluster.metadata.IndexMetaData;
@@ -81,13 +81,32 @@ public void writeTo(StreamOutput out) throws IOException {
         }
     }
 
-    public static class Response extends ActionResponse {
+    public static class Response extends AcknowledgedResponse {
+
+        Response(boolean acknowledged) {
+            super(acknowledged);
+        }
+
+        Response() {
+        }
+
+        @Override
+        public void readFrom(StreamInput in) throws IOException {
+            super.readFrom(in);
+            readAcknowledged(in);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            writeAcknowledged(out);
+        }
 
     }
 
     public static class RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder> {
 
-        public RequestBuilder(ElasticsearchClient client) {
+        RequestBuilder(ElasticsearchClient client) {
             super(client, INSTANCE, new Request());
         }
     }
@@ -147,7 +166,7 @@ void finalizeResponse() {
 
                                 if (error == null) {
                                     // include task ids?
-                                    listener.onResponse(new Response());
+                                    listener.onResponse(new Response(true));
                                 } else {
                                     // TODO: cancel all started tasks
                                     listener.onFailure(error);
diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/rest/RestFollowExistingIndexAction.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/rest/RestFollowExistingIndexAction.java
@@ -7,21 +7,16 @@
 
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.BytesRestResponse;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
-import org.elasticsearch.rest.RestResponse;
-import org.elasticsearch.rest.RestStatus;
-import org.elasticsearch.rest.action.RestBuilderListener;
+import org.elasticsearch.rest.action.RestToXContentListener;
 import org.elasticsearch.xpack.ccr.action.ShardFollowTask;
 
 import java.io.IOException;
 
 import static org.elasticsearch.xpack.ccr.action.FollowExistingIndexAction.INSTANCE;
 import static org.elasticsearch.xpack.ccr.action.FollowExistingIndexAction.Request;
-import static org.elasticsearch.xpack.ccr.action.FollowExistingIndexAction.Response;
 
 // TODO: change to confirm with API design
 public class RestFollowExistingIndexAction extends BaseRestHandler {
@@ -52,13 +47,6 @@ protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient
             long value = Long.valueOf(restRequest.param(ShardFollowTask.PROCESSOR_MAX_TRANSLOG_BYTES_PER_REQUEST.getPreferredName()));
             request.setProcessorMaxTranslogBytes(value);
         }
-        return channel -> client.execute(INSTANCE, request, new RestBuilderListener<Response>(channel) {
-            @Override
-            public RestResponse buildResponse(Response response, XContentBuilder builder) throws Exception {
-                return new BytesRestResponse(RestStatus.OK, builder.startObject()
-                        .endObject());
-
-            }
-        });
+        return channel -> client.execute(INSTANCE, request, new RestToXContentListener<>(channel));
     }
-}
+}
diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/rest/RestUnfollowIndexAction.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/rest/RestUnfollowIndexAction.java
@@ -7,20 +7,15 @@
 
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.rest.BaseRestHandler;
-import org.elasticsearch.rest.BytesRestResponse;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
-import org.elasticsearch.rest.RestResponse;
-import org.elasticsearch.rest.RestStatus;
-import org.elasticsearch.rest.action.RestBuilderListener;
+import org.elasticsearch.rest.action.RestToXContentListener;
 
 import java.io.IOException;
 
 import static org.elasticsearch.xpack.ccr.action.UnfollowIndexAction.INSTANCE;
 import static org.elasticsearch.xpack.ccr.action.UnfollowIndexAction.Request;
-import static org.elasticsearch.xpack.ccr.action.UnfollowIndexAction.Response;
 
 // TODO: change to confirm with API design
 public class RestUnfollowIndexAction extends BaseRestHandler {
@@ -40,13 +35,6 @@ public String getName() {
     protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
         Request request = new Request();
         request.setFollowIndex(restRequest.param("follow_index"));
-        return channel -> client.execute(INSTANCE, request, new RestBuilderListener<Response>(channel) {
-            @Override
-            public RestResponse buildResponse(Response response, XContentBuilder builder) throws Exception {
-                return new BytesRestResponse(RestStatus.OK, builder.startObject()
-                        .endObject());
-
-            }
-        });
+        return channel -> client.execute(INSTANCE, request, new RestToXContentListener<>(channel));
     }
 }
diff --git a/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/ShardChangesIT.java b/x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/ShardChangesIT.java
@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.ccr;
 
+import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksAction;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksRequest;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksResponse;
 import org.elasticsearch.action.admin.indices.stats.ShardStats;
@@ -203,6 +204,17 @@ public void testFollowIndex() throws Exception {
             final ClusterState clusterState = client().admin().cluster().prepareState().get().getState();
             final PersistentTasksCustomMetaData tasks = clusterState.getMetaData().custom(PersistentTasksCustomMetaData.TYPE);
             assertThat(tasks.tasks().size(), equalTo(0));
+
+            ListTasksRequest listTasksRequest = new ListTasksRequest();
+            listTasksRequest.setDetailed(true);
+            ListTasksResponse listTasksResponse = client().admin().cluster().listTasks(listTasksRequest).get();
+            int numNodeTasks = 0;
+            for (TaskInfo taskInfo : listTasksResponse.getTasks()) {
+                if (taskInfo.getAction().startsWith(ListTasksAction.NAME) == false) {
+                    numNodeTasks++;
+                }
+            }
+            assertThat(numNodeTasks, equalTo(0));
         });
     }
 
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/api/xpack.ccr.follow_existing_index.json b/x-pack/plugin/src/test/resources/rest-api-spec/api/xpack.ccr.follow_existing_index.json
@@ -0,0 +1,24 @@
+{
+  "xpack.ccr.follow_existing_index": {
+    "documentation": "http://www.elastic.co/guide/en/elasticsearch/reference/current",
+    "methods": [ "POST" ],
+    "url": {
+      "path": "/_xpack/ccr/{follow_index}/_follow",
+      "paths": [ "/_xpack/ccr/{follow_index}/_follow" ],
+      "parts": {
+        "follow_index": {
+          "type": "string",
+          "required": true,
+          "description": "The name of the index that follows to leader index."
+        }
+      },
+      "params": {
+        "leader_index": {
+          "type": "string",
+          "required": true,
+          "description": "The name of the index to read the changes from."
+        }
+      }
+    }
+  }
+}
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/api/xpack.ccr.unfollow_index.json b/x-pack/plugin/src/test/resources/rest-api-spec/api/xpack.ccr.unfollow_index.json
@@ -0,0 +1,19 @@
+{
+  "xpack.ccr.unfollow_index": {
+    "documentation": "http://www.elastic.co/guide/en/elasticsearch/reference/current",
+    "methods": [ "POST" ],
+    "url": {
+      "path": "/_xpack/ccr/{follow_index}/_unfollow",
+      "paths": [ "/_xpack/ccr/{follow_index}/_unfollow" ],
+      "parts": {
+        "follow_index": {
+          "type": "string",
+          "required": true,
+          "description": "The name of the follow index that should stop following its leader index."
+        }
+      },
+      "params": {
+      }
+    }
+  }
+}
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/test/ccr/follow_and_unfollow.yml b/x-pack/plugin/src/test/resources/rest-api-spec/test/ccr/follow_and_unfollow.yml
@@ -0,0 +1,34 @@
+---
+"Test follow and unfollow an existing index":
+  - do:
+      indices.create:
+        index: foo
+        body:
+          mappings:
+            doc:
+              properties:
+                field:
+                  type: keyword
+  - is_true: acknowledged
+
+  - do:
+      indices.create:
+        index: bar
+        body:
+          mappings:
+            doc:
+              properties:
+                field:
+                  type: keyword
+  - is_true: acknowledged
+
+  - do:
+      xpack.ccr.follow_existing_index:
+        leader_index: foo
+        follow_index: bar
+  - is_true: acknowledged
+
+  - do:
+      xpack.ccr.unfollow_index:
+        follow_index: bar
+  - is_true: acknowledged

Original file line number	Diff line number	Diff line change
`@@ -29,6 +29,11 @@ public class ShardFollowNodeTask extends AllocatedPersistentTask {`
`29`	`29`	`super(id, type, action, description, parentTask, headers);`
`30`	`30`	`}`
`31`	`31`
	`32`	`+ @Override`
	`33`	`+ protected void onCancelled() {`
	`34`	`+ markAsCompleted();`
	`35`	`+ }`
	`36`	`+`
`32`	`37`	`void updateProcessedGlobalCheckpoint(long processedGlobalCheckpoint) {`
`33`	`38`	`this.processedGlobalCheckpoint.set(processedGlobalCheckpoint);`
`34`	`39`	`}`