From c60caeffb8726e2a1d8d96b7d2cbe5aad70aa2ce Mon Sep 17 00:00:00 2001 From: Megha Sai Kavikondala Date: Wed, 8 Dec 2021 23:21:29 +0000 Subject: [PATCH 1/2] Adding the cancelled field for telling if the task is cancelled. Signed-off-by: Megha Sai Kavikondala --- .../org/opensearch/client/tasks/TaskInfo.java | 14 +++++ .../core/tasks/GetTaskResponseTests.java | 15 ++++- .../tasks/CancelTasksResponseTests.java | 6 +- .../TransportRethrottleActionTests.java | 2 + .../admin/cluster/node/tasks/TasksIT.java | 14 ++++- .../main/java/org/opensearch/tasks/Task.java | 1 + .../java/org/opensearch/tasks/TaskInfo.java | 59 ++++++++++++++++++- .../opensearch/tasks/TaskResultsService.java | 8 ++- .../admin/cluster/node/tasks/TaskTests.java | 7 +++ .../tasks/ListTasksResponseTests.java | 2 + .../org/opensearch/tasks/TaskInfoTests.java | 25 +++++++- 11 files changed, 146 insertions(+), 7 deletions(-) diff --git a/client/rest-high-level/src/main/java/org/opensearch/client/tasks/TaskInfo.java b/client/rest-high-level/src/main/java/org/opensearch/client/tasks/TaskInfo.java index 062fbe56e4ed9..de8374b283ea6 100644 --- a/client/rest-high-level/src/main/java/org/opensearch/client/tasks/TaskInfo.java +++ b/client/rest-high-level/src/main/java/org/opensearch/client/tasks/TaskInfo.java @@ -53,6 +53,7 @@ public class TaskInfo { private long startTime; private long runningTimeNanos; private boolean cancellable; + private boolean cancelled; private TaskId parentTaskId; private final Map status = new HashMap<>(); private final Map headers = new HashMap<>(); @@ -117,6 +118,14 @@ void setCancellable(boolean cancellable) { this.cancellable = cancellable; } + public boolean isCancelled() { + return cancelled; + } + + void setCancelled(boolean cancelled) { + this.cancelled = cancelled; + } + public TaskId getParentTaskId() { return parentTaskId; } @@ -158,6 +167,7 @@ private void noOpParse(Object s) {} parser.declareLong(TaskInfo::setStartTime, new ParseField("start_time_in_millis")); parser.declareLong(TaskInfo::setRunningTimeNanos, new ParseField("running_time_in_nanos")); parser.declareBoolean(TaskInfo::setCancellable, new ParseField("cancellable")); + parser.declareBoolean(TaskInfo::setCancelled, new ParseField("cancelled")); parser.declareString(TaskInfo::setParentTaskId, new ParseField("parent_task_id")); parser.declareObject(TaskInfo::setHeaders, (p, c) -> p.mapStrings(), new ParseField("headers")); PARSER = (XContentParser p, Void v, String name) -> parser.parse(p, new TaskInfo(new TaskId(name)), null); @@ -171,6 +181,7 @@ public boolean equals(Object o) { return getStartTime() == taskInfo.getStartTime() && getRunningTimeNanos() == taskInfo.getRunningTimeNanos() && isCancellable() == taskInfo.isCancellable() + && isCancelled() == taskInfo.isCancelled() && Objects.equals(getTaskId(), taskInfo.getTaskId()) && Objects.equals(getType(), taskInfo.getType()) && Objects.equals(getAction(), taskInfo.getAction()) @@ -190,6 +201,7 @@ public int hashCode() { getStartTime(), getRunningTimeNanos(), isCancellable(), + isCancelled(), getParentTaskId(), status, getHeaders() @@ -216,6 +228,8 @@ public String toString() { + runningTimeNanos + ", cancellable=" + cancellable + + ", cancelled=" + + cancelled + ", parentTaskId=" + parentTaskId + ", status=" diff --git a/client/rest-high-level/src/test/java/org/opensearch/client/core/tasks/GetTaskResponseTests.java b/client/rest-high-level/src/test/java/org/opensearch/client/core/tasks/GetTaskResponseTests.java index a14e1169d09fc..403e295303784 100644 --- a/client/rest-high-level/src/test/java/org/opensearch/client/core/tasks/GetTaskResponseTests.java +++ b/client/rest-high-level/src/test/java/org/opensearch/client/core/tasks/GetTaskResponseTests.java @@ -90,11 +90,24 @@ static TaskInfo randomTaskInfo() { long startTime = randomLong(); long runningTimeNanos = randomLong(); boolean cancellable = randomBoolean(); + boolean cancelled = cancellable == true ? randomBoolean() : false; TaskId parentTaskId = randomBoolean() ? TaskId.EMPTY_TASK_ID : randomTaskId(); Map headers = randomBoolean() ? Collections.emptyMap() : Collections.singletonMap(randomAlphaOfLength(5), randomAlphaOfLength(5)); - return new TaskInfo(taskId, type, action, description, status, startTime, runningTimeNanos, cancellable, parentTaskId, headers); + return new TaskInfo( + taskId, + type, + action, + description, + status, + startTime, + runningTimeNanos, + cancellable, + cancelled, + parentTaskId, + headers + ); } private static TaskId randomTaskId() { diff --git a/client/rest-high-level/src/test/java/org/opensearch/client/tasks/CancelTasksResponseTests.java b/client/rest-high-level/src/test/java/org/opensearch/client/tasks/CancelTasksResponseTests.java index 102ebb5fcd390..552a3712eea40 100644 --- a/client/rest-high-level/src/test/java/org/opensearch/client/tasks/CancelTasksResponseTests.java +++ b/client/rest-high-level/src/test/java/org/opensearch/client/tasks/CancelTasksResponseTests.java @@ -82,6 +82,8 @@ protected CancelTasksResponseTests.ByNodeCancelTasksResponse createServerTestIns } for (int i = 0; i < 4; i++) { + boolean cancellable = randomBoolean(); + boolean cancelled = cancellable == true ? randomBoolean() : false; tasks.add( new org.opensearch.tasks.TaskInfo( new TaskId(NODE_ID, (long) i), @@ -91,7 +93,8 @@ protected CancelTasksResponseTests.ByNodeCancelTasksResponse createServerTestIns new FakeTaskStatus(randomAlphaOfLength(4), randomInt()), randomLongBetween(1, 3), randomIntBetween(5, 10), - false, + cancellable, + cancelled, new TaskId("node1", randomLong()), Collections.singletonMap("x-header-of", "some-value") ) @@ -128,6 +131,7 @@ protected void assertInstances( assertEquals(ti.getStartTime(), taskInfo.getStartTime()); assertEquals(ti.getRunningTimeNanos(), taskInfo.getRunningTimeNanos()); assertEquals(ti.isCancellable(), taskInfo.isCancellable()); + assertEquals(ti.isCancelled(), taskInfo.isCancelled()); assertEquals(ti.getParentTaskId().getNodeId(), taskInfo.getParentTaskId().getNodeId()); assertEquals(ti.getParentTaskId().getId(), taskInfo.getParentTaskId().getId()); FakeTaskStatus status = (FakeTaskStatus) ti.getStatus(); diff --git a/modules/reindex/src/test/java/org/opensearch/index/reindex/TransportRethrottleActionTests.java b/modules/reindex/src/test/java/org/opensearch/index/reindex/TransportRethrottleActionTests.java index 4e6d3401a2f14..7c7c8d1a2c4ce 100644 --- a/modules/reindex/src/test/java/org/opensearch/index/reindex/TransportRethrottleActionTests.java +++ b/modules/reindex/src/test/java/org/opensearch/index/reindex/TransportRethrottleActionTests.java @@ -129,6 +129,7 @@ public void testRethrottleSuccessfulResponse() { 0, 0, true, + false, new TaskId("test", task.getId()), Collections.emptyMap() ) @@ -164,6 +165,7 @@ public void testRethrottleWithSomeSucceeded() { 0, 0, true, + false, new TaskId("test", task.getId()), Collections.emptyMap() ) diff --git a/server/src/internalClusterTest/java/org/opensearch/action/admin/cluster/node/tasks/TasksIT.java b/server/src/internalClusterTest/java/org/opensearch/action/admin/cluster/node/tasks/TasksIT.java index 02034ae05dc03..4dd68b4cb2421 100644 --- a/server/src/internalClusterTest/java/org/opensearch/action/admin/cluster/node/tasks/TasksIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/action/admin/cluster/node/tasks/TasksIT.java @@ -901,7 +901,19 @@ public void testNodeNotFoundButTaskFound() throws Exception { TaskResultsService resultsService = internalCluster().getInstance(TaskResultsService.class); resultsService.storeResult( new TaskResult( - new TaskInfo(new TaskId("fake", 1), "test", "test", "", null, 0, 0, false, TaskId.EMPTY_TASK_ID, Collections.emptyMap()), + new TaskInfo( + new TaskId("fake", 1), + "test", + "test", + "", + null, + 0, + 0, + false, + false, + TaskId.EMPTY_TASK_ID, + Collections.emptyMap() + ), new RuntimeException("test") ), new ActionListener() { diff --git a/server/src/main/java/org/opensearch/tasks/Task.java b/server/src/main/java/org/opensearch/tasks/Task.java index 8646a97da5cfe..ad9d5c3f04411 100644 --- a/server/src/main/java/org/opensearch/tasks/Task.java +++ b/server/src/main/java/org/opensearch/tasks/Task.java @@ -130,6 +130,7 @@ protected final TaskInfo taskInfo(String localNodeId, String description, Status startTime, System.nanoTime() - startTimeNanos, this instanceof CancellableTask, + this instanceof CancellableTask && ((CancellableTask) this).isCancelled(), parentTask, headers ); diff --git a/server/src/main/java/org/opensearch/tasks/TaskInfo.java b/server/src/main/java/org/opensearch/tasks/TaskInfo.java index 03afa763efd65..2348e53618d80 100644 --- a/server/src/main/java/org/opensearch/tasks/TaskInfo.java +++ b/server/src/main/java/org/opensearch/tasks/TaskInfo.java @@ -32,6 +32,7 @@ package org.opensearch.tasks; +import org.opensearch.Version; import org.opensearch.LegacyESVersion; import org.opensearch.common.ParseField; import org.opensearch.common.Strings; @@ -64,6 +65,9 @@ * snapshot information about currently running tasks. */ public final class TaskInfo implements Writeable, ToXContentFragment { + + static final String INCLUDE_CANCELLED_PARAM = "include_cancelled"; + private final TaskId taskId; private final String type; @@ -80,6 +84,8 @@ public final class TaskInfo implements Writeable, ToXContentFragment { private final boolean cancellable; + private final boolean cancelled; + private final TaskId parentTaskId; private final Map headers; @@ -93,9 +99,11 @@ public TaskInfo( long startTime, long runningTimeNanos, boolean cancellable, + boolean cancelled, TaskId parentTaskId, Map headers ) { + assert cancellable || cancelled == false : "uncancellable task cannot be cancelled"; this.taskId = taskId; this.type = type; this.action = action; @@ -104,6 +112,7 @@ public TaskInfo( this.startTime = startTime; this.runningTimeNanos = runningTimeNanos; this.cancellable = cancellable; + this.cancelled = cancelled; this.parentTaskId = parentTaskId; this.headers = headers; } @@ -120,6 +129,12 @@ public TaskInfo(StreamInput in) throws IOException { startTime = in.readLong(); runningTimeNanos = in.readLong(); cancellable = in.readBoolean(); + if (in.getVersion().onOrAfter(Version.V_2_0_0)) { + cancelled = in.readBoolean(); + } else { + cancelled = false; + } + assert cancellable || cancelled == false : "uncancellable task cannot be cancelled"; parentTaskId = TaskId.readFromStream(in); if (in.getVersion().onOrAfter(LegacyESVersion.V_6_2_0)) { headers = in.readMap(StreamInput::readString, StreamInput::readString); @@ -138,6 +153,9 @@ public void writeTo(StreamOutput out) throws IOException { out.writeLong(startTime); out.writeLong(runningTimeNanos); out.writeBoolean(cancellable); + if (out.getVersion().onOrAfter(Version.V_2_0_0)) { + out.writeBoolean(cancelled); + } parentTaskId.writeTo(out); if (out.getVersion().onOrAfter(LegacyESVersion.V_6_2_0)) { out.writeMap(headers, StreamOutput::writeString, StreamOutput::writeString); @@ -193,6 +211,13 @@ public boolean isCancellable() { return cancellable; } + /** + * Returns true if the task has been cancelled + */ + public boolean isCancelled() { + return cancelled; + } + /** * Returns the parent task id */ @@ -225,6 +250,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws } builder.field("running_time_in_nanos", runningTimeNanos); builder.field("cancellable", cancellable); + if (params.paramAsBoolean(INCLUDE_CANCELLED_PARAM, true) && cancellable) { + builder.field("cancelled", cancelled); + } if (parentTaskId.isSet()) { builder.field("parent_task_id", parentTaskId.toString()); } @@ -250,6 +278,7 @@ public static TaskInfo fromXContent(XContentParser parser) { long startTime = (Long) a[i++]; long runningTimeNanos = (Long) a[i++]; boolean cancellable = (Boolean) a[i++]; + boolean cancelled = a[i++] == Boolean.TRUE; String parentTaskIdString = (String) a[i++]; @SuppressWarnings("unchecked") Map headers = (Map) a[i++]; @@ -259,7 +288,19 @@ public static TaskInfo fromXContent(XContentParser parser) { } RawTaskStatus status = statusBytes == null ? null : new RawTaskStatus(statusBytes); TaskId parentTaskId = parentTaskIdString == null ? TaskId.EMPTY_TASK_ID : new TaskId(parentTaskIdString); - return new TaskInfo(id, type, action, description, status, startTime, runningTimeNanos, cancellable, parentTaskId, headers); + return new TaskInfo( + id, + type, + action, + description, + status, + startTime, + runningTimeNanos, + cancellable, + cancelled, + parentTaskId, + headers + ); }); static { // Note for the future: this has to be backwards and forwards compatible with all changes to the task storage format @@ -273,6 +314,7 @@ public static TaskInfo fromXContent(XContentParser parser) { PARSER.declareLong(constructorArg(), new ParseField("start_time_in_millis")); PARSER.declareLong(constructorArg(), new ParseField("running_time_in_nanos")); PARSER.declareBoolean(constructorArg(), new ParseField("cancellable")); + PARSER.declareBoolean(optionalConstructorArg(), new ParseField("cancelled")); PARSER.declareString(optionalConstructorArg(), new ParseField("parent_task_id")); PARSER.declareObject(optionalConstructorArg(), (p, c) -> p.mapStrings(), new ParseField("headers")); } @@ -297,12 +339,25 @@ public boolean equals(Object obj) { && Objects.equals(runningTimeNanos, other.runningTimeNanos) && Objects.equals(parentTaskId, other.parentTaskId) && Objects.equals(cancellable, other.cancellable) + && Objects.equals(cancelled, other.cancelled) && Objects.equals(status, other.status) && Objects.equals(headers, other.headers); } @Override public int hashCode() { - return Objects.hash(taskId, type, action, description, startTime, runningTimeNanos, parentTaskId, cancellable, status, headers); + return Objects.hash( + taskId, + type, + action, + description, + startTime, + runningTimeNanos, + parentTaskId, + cancellable, + cancelled, + status, + headers + ); } } diff --git a/server/src/main/java/org/opensearch/tasks/TaskResultsService.java b/server/src/main/java/org/opensearch/tasks/TaskResultsService.java index 8767ce95d3352..904600a1a3dd3 100644 --- a/server/src/main/java/org/opensearch/tasks/TaskResultsService.java +++ b/server/src/main/java/org/opensearch/tasks/TaskResultsService.java @@ -67,9 +67,12 @@ import java.nio.charset.StandardCharsets; import java.util.Iterator; import java.util.Map; +import java.util.stream.Collectors; +import java.util.Arrays; import static org.opensearch.action.admin.cluster.node.tasks.get.GetTaskAction.TASKS_ORIGIN; import static org.opensearch.common.unit.TimeValue.timeValueMillis; +import static org.opensearch.tasks.TaskInfo.INCLUDE_CANCELLED_PARAM; /** * Service that can store task results. @@ -78,6 +81,9 @@ public class TaskResultsService { private static final Logger logger = LogManager.getLogger(TaskResultsService.class); + private static Map mapParams = Arrays.stream(new String[][] { { INCLUDE_CANCELLED_PARAM, "false" } }) + .collect(Collectors.toMap(keyMapper -> keyMapper[0], valueMapper -> valueMapper[1])); + public static final String TASK_INDEX = ".tasks"; public static final String TASK_TYPE = "task"; @@ -171,7 +177,7 @@ private int getTaskResultMappingVersion(IndexMetadata metadata) { private void doStoreResult(TaskResult taskResult, ActionListener listener) { IndexRequestBuilder index = client.prepareIndex(TASK_INDEX, TASK_TYPE, taskResult.getTask().getTaskId().toString()); try (XContentBuilder builder = XContentFactory.contentBuilder(Requests.INDEX_CONTENT_TYPE)) { - taskResult.toXContent(builder, ToXContent.EMPTY_PARAMS); + taskResult.toXContent(builder, new ToXContent.MapParams(mapParams)); index.setSource(builder); } catch (IOException e) { throw new OpenSearchException("Couldn't convert task result to XContent for [{}]", e, taskResult.getTask()); diff --git a/server/src/test/java/org/opensearch/action/admin/cluster/node/tasks/TaskTests.java b/server/src/test/java/org/opensearch/action/admin/cluster/node/tasks/TaskTests.java index a2f8f0a5f7a44..50c34a7b01c06 100644 --- a/server/src/test/java/org/opensearch/action/admin/cluster/node/tasks/TaskTests.java +++ b/server/src/test/java/org/opensearch/action/admin/cluster/node/tasks/TaskTests.java @@ -49,6 +49,7 @@ public void testTaskInfoToString() { long startTime = randomNonNegativeLong(); long runningTime = randomNonNegativeLong(); boolean cancellable = randomBoolean(); + boolean cancelled = cancellable == true ? randomBoolean() : false; TaskInfo taskInfo = new TaskInfo( new TaskId(nodeId, taskId), "test_type", @@ -58,6 +59,7 @@ public void testTaskInfoToString() { startTime, runningTime, cancellable, + cancelled, TaskId.EMPTY_TASK_ID, Collections.singletonMap("foo", "bar") ); @@ -70,6 +72,11 @@ public void testTaskInfoToString() { assertEquals(((Number) map.get("start_time_in_millis")).longValue(), startTime); assertEquals(((Number) map.get("running_time_in_nanos")).longValue(), runningTime); assertEquals(map.get("cancellable"), cancellable); + if (cancellable) { + assertEquals(map.get("cancelled"), cancelled); + } else { + assertFalse(map.containsKey("cancelled")); + } assertEquals(map.get("headers"), Collections.singletonMap("foo", "bar")); } diff --git a/server/src/test/java/org/opensearch/tasks/ListTasksResponseTests.java b/server/src/test/java/org/opensearch/tasks/ListTasksResponseTests.java index 450dd522ca891..4d5feb46de1d0 100644 --- a/server/src/test/java/org/opensearch/tasks/ListTasksResponseTests.java +++ b/server/src/test/java/org/opensearch/tasks/ListTasksResponseTests.java @@ -70,6 +70,7 @@ public void testNonEmptyToString() { 0, 1, true, + false, new TaskId("node1", 0), Collections.singletonMap("foo", "bar") ); @@ -88,6 +89,7 @@ public void testNonEmptyToString() { + " \"running_time\" : \"1nanos\",\n" + " \"running_time_in_nanos\" : 1,\n" + " \"cancellable\" : true,\n" + + " \"cancelled\" : false,\n" + " \"parent_task_id\" : \"node1:0\",\n" + " \"headers\" : {\n" + " \"foo\" : \"bar\"\n" diff --git a/server/src/test/java/org/opensearch/tasks/TaskInfoTests.java b/server/src/test/java/org/opensearch/tasks/TaskInfoTests.java index b9a0d05149bb8..89b690d81a4ea 100644 --- a/server/src/test/java/org/opensearch/tasks/TaskInfoTests.java +++ b/server/src/test/java/org/opensearch/tasks/TaskInfoTests.java @@ -95,6 +95,7 @@ protected TaskInfo mutateInstance(TaskInfo info) { info.getStartTime(), info.getRunningTimeNanos(), info.isCancellable(), + info.isCancelled(), info.getParentTaskId(), info.getHeaders() ); @@ -108,6 +109,7 @@ protected TaskInfo mutateInstance(TaskInfo info) { info.getStartTime(), info.getRunningTimeNanos(), info.isCancellable(), + info.isCancelled(), info.getParentTaskId(), info.getHeaders() ); @@ -121,6 +123,7 @@ protected TaskInfo mutateInstance(TaskInfo info) { info.getStartTime(), info.getRunningTimeNanos(), info.isCancellable(), + info.isCancelled(), info.getParentTaskId(), info.getHeaders() ); @@ -134,6 +137,7 @@ protected TaskInfo mutateInstance(TaskInfo info) { info.getStartTime(), info.getRunningTimeNanos(), info.isCancellable(), + info.isCancelled(), info.getParentTaskId(), info.getHeaders() ); @@ -148,6 +152,7 @@ protected TaskInfo mutateInstance(TaskInfo info) { info.getStartTime(), info.getRunningTimeNanos(), info.isCancellable(), + info.isCancelled(), info.getParentTaskId(), info.getHeaders() ); @@ -161,6 +166,7 @@ protected TaskInfo mutateInstance(TaskInfo info) { info.getStartTime() + between(1, 100), info.getRunningTimeNanos(), info.isCancellable(), + info.isCancelled(), info.getParentTaskId(), info.getHeaders() ); @@ -174,6 +180,7 @@ protected TaskInfo mutateInstance(TaskInfo info) { info.getStartTime(), info.getRunningTimeNanos() + between(1, 100), info.isCancellable(), + info.isCancelled(), info.getParentTaskId(), info.getHeaders() ); @@ -187,6 +194,7 @@ protected TaskInfo mutateInstance(TaskInfo info) { info.getStartTime(), info.getRunningTimeNanos(), info.isCancellable() == false, + false, info.getParentTaskId(), info.getHeaders() ); @@ -201,6 +209,7 @@ protected TaskInfo mutateInstance(TaskInfo info) { info.getStartTime(), info.getRunningTimeNanos(), info.isCancellable(), + info.isCancelled(), parentId, info.getHeaders() ); @@ -221,6 +230,7 @@ protected TaskInfo mutateInstance(TaskInfo info) { info.getStartTime(), info.getRunningTimeNanos(), info.isCancellable(), + info.isCancelled(), info.getParentTaskId(), headers ); @@ -238,11 +248,24 @@ static TaskInfo randomTaskInfo() { long startTime = randomLong(); long runningTimeNanos = randomLong(); boolean cancellable = randomBoolean(); + boolean cancelled = cancellable == true ? randomBoolean() : false; TaskId parentTaskId = randomBoolean() ? TaskId.EMPTY_TASK_ID : randomTaskId(); Map headers = randomBoolean() ? Collections.emptyMap() : Collections.singletonMap(randomAlphaOfLength(5), randomAlphaOfLength(5)); - return new TaskInfo(taskId, type, action, description, status, startTime, runningTimeNanos, cancellable, parentTaskId, headers); + return new TaskInfo( + taskId, + type, + action, + description, + status, + startTime, + runningTimeNanos, + cancellable, + cancelled, + parentTaskId, + headers + ); } private static TaskId randomTaskId() { From c0f838439041b18b511ec6c5b42bac033ff1753f Mon Sep 17 00:00:00 2001 From: Megha Sai Kavikondala Date: Fri, 10 Dec 2021 21:43:27 +0000 Subject: [PATCH 2/2] Changes made to add the cancelled field when the task is completed. Signed-off-by: Megha Sai Kavikondala --- .../java/org/opensearch/tasks/TaskInfo.java | 7 +- .../opensearch/tasks/TaskResultsService.java | 8 +- .../opensearch/tasks/task-index-mapping.json | 3 + .../admin/cluster/node/tasks/TaskTests.java | 75 +++++++++++++++++-- 4 files changed, 73 insertions(+), 20 deletions(-) diff --git a/server/src/main/java/org/opensearch/tasks/TaskInfo.java b/server/src/main/java/org/opensearch/tasks/TaskInfo.java index 2348e53618d80..a4b4f3e2090d7 100644 --- a/server/src/main/java/org/opensearch/tasks/TaskInfo.java +++ b/server/src/main/java/org/opensearch/tasks/TaskInfo.java @@ -65,9 +65,6 @@ * snapshot information about currently running tasks. */ public final class TaskInfo implements Writeable, ToXContentFragment { - - static final String INCLUDE_CANCELLED_PARAM = "include_cancelled"; - private final TaskId taskId; private final String type; @@ -250,9 +247,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws } builder.field("running_time_in_nanos", runningTimeNanos); builder.field("cancellable", cancellable); - if (params.paramAsBoolean(INCLUDE_CANCELLED_PARAM, true) && cancellable) { - builder.field("cancelled", cancelled); - } + builder.field("cancelled", cancelled); if (parentTaskId.isSet()) { builder.field("parent_task_id", parentTaskId.toString()); } diff --git a/server/src/main/java/org/opensearch/tasks/TaskResultsService.java b/server/src/main/java/org/opensearch/tasks/TaskResultsService.java index 904600a1a3dd3..8767ce95d3352 100644 --- a/server/src/main/java/org/opensearch/tasks/TaskResultsService.java +++ b/server/src/main/java/org/opensearch/tasks/TaskResultsService.java @@ -67,12 +67,9 @@ import java.nio.charset.StandardCharsets; import java.util.Iterator; import java.util.Map; -import java.util.stream.Collectors; -import java.util.Arrays; import static org.opensearch.action.admin.cluster.node.tasks.get.GetTaskAction.TASKS_ORIGIN; import static org.opensearch.common.unit.TimeValue.timeValueMillis; -import static org.opensearch.tasks.TaskInfo.INCLUDE_CANCELLED_PARAM; /** * Service that can store task results. @@ -81,9 +78,6 @@ public class TaskResultsService { private static final Logger logger = LogManager.getLogger(TaskResultsService.class); - private static Map mapParams = Arrays.stream(new String[][] { { INCLUDE_CANCELLED_PARAM, "false" } }) - .collect(Collectors.toMap(keyMapper -> keyMapper[0], valueMapper -> valueMapper[1])); - public static final String TASK_INDEX = ".tasks"; public static final String TASK_TYPE = "task"; @@ -177,7 +171,7 @@ private int getTaskResultMappingVersion(IndexMetadata metadata) { private void doStoreResult(TaskResult taskResult, ActionListener listener) { IndexRequestBuilder index = client.prepareIndex(TASK_INDEX, TASK_TYPE, taskResult.getTask().getTaskId().toString()); try (XContentBuilder builder = XContentFactory.contentBuilder(Requests.INDEX_CONTENT_TYPE)) { - taskResult.toXContent(builder, new ToXContent.MapParams(mapParams)); + taskResult.toXContent(builder, ToXContent.EMPTY_PARAMS); index.setSource(builder); } catch (IOException e) { throw new OpenSearchException("Couldn't convert task result to XContent for [{}]", e, taskResult.getTask()); diff --git a/server/src/main/resources/org/opensearch/tasks/task-index-mapping.json b/server/src/main/resources/org/opensearch/tasks/task-index-mapping.json index 40730fc02886b..76b07bf3570f2 100644 --- a/server/src/main/resources/org/opensearch/tasks/task-index-mapping.json +++ b/server/src/main/resources/org/opensearch/tasks/task-index-mapping.json @@ -16,6 +16,9 @@ "cancellable": { "type": "boolean" }, + "cancelled": { + "type": "boolean" + }, "id": { "type": "long" }, diff --git a/server/src/test/java/org/opensearch/action/admin/cluster/node/tasks/TaskTests.java b/server/src/test/java/org/opensearch/action/admin/cluster/node/tasks/TaskTests.java index 50c34a7b01c06..0b01593e24633 100644 --- a/server/src/test/java/org/opensearch/action/admin/cluster/node/tasks/TaskTests.java +++ b/server/src/test/java/org/opensearch/action/admin/cluster/node/tasks/TaskTests.java @@ -48,8 +48,8 @@ public void testTaskInfoToString() { long taskId = randomIntBetween(0, 100000); long startTime = randomNonNegativeLong(); long runningTime = randomNonNegativeLong(); - boolean cancellable = randomBoolean(); - boolean cancelled = cancellable == true ? randomBoolean() : false; + boolean cancellable = false; + boolean cancelled = false; TaskInfo taskInfo = new TaskInfo( new TaskId(nodeId, taskId), "test_type", @@ -72,12 +72,73 @@ public void testTaskInfoToString() { assertEquals(((Number) map.get("start_time_in_millis")).longValue(), startTime); assertEquals(((Number) map.get("running_time_in_nanos")).longValue(), runningTime); assertEquals(map.get("cancellable"), cancellable); - if (cancellable) { - assertEquals(map.get("cancelled"), cancelled); - } else { - assertFalse(map.containsKey("cancelled")); - } + assertEquals(map.get("cancelled"), cancelled); assertEquals(map.get("headers"), Collections.singletonMap("foo", "bar")); } + public void testCancellableOption() { + String nodeId = randomAlphaOfLength(10); + long taskId = randomIntBetween(0, 100000); + long startTime = randomNonNegativeLong(); + long runningTime = randomNonNegativeLong(); + boolean cancellable = true; + boolean cancelled = true; + TaskInfo taskInfo = new TaskInfo( + new TaskId(nodeId, taskId), + "test_type", + "test_action", + "test_description", + null, + startTime, + runningTime, + cancellable, + cancelled, + TaskId.EMPTY_TASK_ID, + Collections.singletonMap("foo", "bar") + ); + String taskInfoString = taskInfo.toString(); + Map map = XContentHelper.convertToMap(new BytesArray(taskInfoString.getBytes(StandardCharsets.UTF_8)), true).v2(); + assertEquals(((Number) map.get("id")).longValue(), taskId); + assertEquals(map.get("type"), "test_type"); + assertEquals(map.get("action"), "test_action"); + assertEquals(map.get("description"), "test_description"); + assertEquals(((Number) map.get("start_time_in_millis")).longValue(), startTime); + assertEquals(((Number) map.get("running_time_in_nanos")).longValue(), runningTime); + assertEquals(map.get("cancellable"), cancellable); + assertEquals(map.get("cancelled"), cancelled); + assertEquals(map.get("headers"), Collections.singletonMap("foo", "bar")); + } + + public void testNonCancellableOption() { + String nodeId = randomAlphaOfLength(10); + long taskId = randomIntBetween(0, 100000); + long startTime = randomNonNegativeLong(); + long runningTime = randomNonNegativeLong(); + boolean cancellable = true; + boolean cancelled = false; + TaskInfo taskInfo = new TaskInfo( + new TaskId(nodeId, taskId), + "test_type", + "test_action", + "test_description", + null, + startTime, + runningTime, + cancellable, + cancelled, + TaskId.EMPTY_TASK_ID, + Collections.singletonMap("foo", "bar") + ); + String taskInfoString = taskInfo.toString(); + Map map = XContentHelper.convertToMap(new BytesArray(taskInfoString.getBytes(StandardCharsets.UTF_8)), true).v2(); + assertEquals(((Number) map.get("id")).longValue(), taskId); + assertEquals(map.get("type"), "test_type"); + assertEquals(map.get("action"), "test_action"); + assertEquals(map.get("description"), "test_description"); + assertEquals(((Number) map.get("start_time_in_millis")).longValue(), startTime); + assertEquals(((Number) map.get("running_time_in_nanos")).longValue(), runningTime); + assertEquals(map.get("cancellable"), cancellable); + assertEquals(map.get("cancelled"), cancelled); + assertEquals(map.get("headers"), Collections.singletonMap("foo", "bar")); + } }