Skip to content

Commit d6af669

Browse files
imotovimotov
committed
Combine node name and task id into single string task id
This commit changes the URL for task operations from `/_tasks/{nodeId}/{taskId}` to `/_tasks/{taskId}`, where `{taskId}` has a form of nodeid:id
1 parent 7140d18 commit d6af669

File tree

27 files changed

+354
-308
lines changed

27 files changed

+354
-308
lines changed

‎core/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/cancel/CancelTasksRequest.java‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -36,14 +36,6 @@ public class CancelTasksRequest extends BaseTasksRequest<CancelTasksRequest> {
3636

3737
private String reason = DEFAULT_REASON;
3838

39-
/**
40-
* Cancel tasks on the specified nodes. If none are passed, all cancellable tasks on
41-
* all nodes will be cancelled.
42-
*/
43-
public CancelTasksRequest(String... nodesIds) {
44-
super(nodesIds);
45-
}
46-
4739
@Override
4840
public void readFrom(StreamInput in) throws IOException {
4941
super.readFrom(in);
@@ -54,7 +46,6 @@ public void readFrom(StreamInput in) throws IOException {
5446
public void writeTo(StreamOutput out) throws IOException {
5547
super.writeTo(out);
5648
out.writeString(reason);
57-
5849
}
5950

6051
@Override

‎core/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/cancel/TransportCancelTasksAction.java‎

Lines changed: 34 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,6 @@
2424
import org.elasticsearch.action.TaskOperationFailure;
2525
import org.elasticsearch.action.admin.cluster.node.tasks.list.TaskInfo;
2626
import org.elasticsearch.action.support.ActionFilters;
27-
import org.elasticsearch.action.support.tasks.BaseTasksRequest;
2827
import org.elasticsearch.action.support.tasks.TransportTasksAction;
2928
import org.elasticsearch.cluster.ClusterName;
3029
import org.elasticsearch.cluster.ClusterService;
@@ -36,6 +35,7 @@
3635
import org.elasticsearch.common.io.stream.StreamOutput;
3736
import org.elasticsearch.common.settings.Settings;
3837
import org.elasticsearch.tasks.CancellableTask;
38+
import org.elasticsearch.tasks.TaskId;
3939
import org.elasticsearch.threadpool.ThreadPool;
4040
import org.elasticsearch.transport.EmptyTransportResponseHandler;
4141
import org.elasticsearch.transport.TransportChannel;
@@ -84,17 +84,17 @@ protected TaskInfo readTaskResponse(StreamInput in) throws IOException {
8484
}
8585

8686
protected void processTasks(CancelTasksRequest request, Consumer<CancellableTask> operation) {
87-
if (request.taskId() != BaseTasksRequest.ALL_TASKS) {
87+
if (request.taskId().isSet() == false) {
8888
// we are only checking one task, we can optimize it
89-
CancellableTask task = taskManager.getCancellableTask(request.taskId());
89+
CancellableTask task = taskManager.getCancellableTask(request.taskId().getId());
9090
if (task != null) {
9191
if (request.match(task)) {
9292
operation.accept(task);
9393
} else {
9494
throw new IllegalArgumentException("task [" + request.taskId() + "] doesn't support this operation");
9595
}
9696
} else {
97-
if (taskManager.getTask(request.taskId()) != null) {
97+
if (taskManager.getTask(request.taskId().getId()) != null) {
9898
// The task exists, but doesn't support cancellation
9999
throw new IllegalArgumentException("task [" + request.taskId() + "] doesn't support cancellation");
100100
} else {
@@ -135,11 +135,14 @@ protected boolean accumulateExceptions() {
135135
}
136136

137137
private void setBanOnNodes(String reason, CancellableTask task, Set<String> nodes, BanLock banLock) {
138-
sendSetBanRequest(nodes, new BanParentTaskRequest(clusterService.localNode().getId(), task.getId(), reason), banLock);
138+
sendSetBanRequest(nodes,
139+
BanParentTaskRequest.createSetBanParentTaskRequest(new TaskId(clusterService.localNode().getId(), task.getId()), reason),
140+
banLock);
139141
}
140142

141143
private void removeBanOnNodes(CancellableTask task, Set<String> nodes) {
142-
sendRemoveBanRequest(nodes, new BanParentTaskRequest(clusterService.localNode().getId(), task.getId()));
144+
sendRemoveBanRequest(nodes,
145+
BanParentTaskRequest.createRemoveBanParentTaskRequest(new TaskId(clusterService.localNode().getId(), task.getId())));
143146
}
144147

145148
private void sendSetBanRequest(Set<String> nodes, BanParentTaskRequest request, BanLock banLock) {
@@ -148,8 +151,8 @@ private void sendSetBanRequest(Set<String> nodes, BanParentTaskRequest request,
148151
DiscoveryNode discoveryNode = clusterState.getNodes().get(node);
149152
if (discoveryNode != null) {
150153
// Check if node still in the cluster
151-
logger.debug("Sending ban for tasks with the parent [{}:{}] to the node [{}], ban [{}]", request.parentNodeId, request
152-
.parentTaskId, node, request.ban);
154+
logger.debug("Sending ban for tasks with the parent [{}] to the node [{}], ban [{}]", request.parentTaskId, node,
155+
request.ban);
153156
transportService.sendRequest(discoveryNode, BAN_PARENT_ACTION_NAME, request,
154157
new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {
155158
@Override
@@ -164,8 +167,8 @@ public void handleException(TransportException exp) {
164167
});
165168
} else {
166169
banLock.onBanSet();
167-
logger.debug("Cannot send ban for tasks with the parent [{}:{}] to the node [{}] - the node no longer in the cluster",
168-
request.parentNodeId, request.parentTaskId, node);
170+
logger.debug("Cannot send ban for tasks with the parent [{}] to the node [{}] - the node no longer in the cluster",
171+
request.parentTaskId, node);
169172
}
170173
}
171174
}
@@ -176,13 +179,12 @@ private void sendRemoveBanRequest(Set<String> nodes, BanParentTaskRequest reques
176179
DiscoveryNode discoveryNode = clusterState.getNodes().get(node);
177180
if (discoveryNode != null) {
178181
// Check if node still in the cluster
179-
logger.debug("Sending remove ban for tasks with the parent [{}:{}] to the node [{}]", request.parentNodeId,
180-
request.parentTaskId, node);
182+
logger.debug("Sending remove ban for tasks with the parent [{}] to the node [{}]", request.parentTaskId, node);
181183
transportService.sendRequest(discoveryNode, BAN_PARENT_ACTION_NAME, request, EmptyTransportResponseHandler
182184
.INSTANCE_SAME);
183185
} else {
184-
logger.debug("Cannot send remove ban request for tasks with the parent [{}:{}] to the node [{}] - the node no longer in " +
185-
"the cluster", request.parentNodeId, request.parentTaskId, node);
186+
logger.debug("Cannot send remove ban request for tasks with the parent [{}] to the node [{}] - the node no longer in " +
187+
"the cluster", request.parentTaskId, node);
186188
}
187189
}
188190
}
@@ -218,23 +220,27 @@ public void finish() {
218220

219221
private static class BanParentTaskRequest extends TransportRequest {
220222

221-
private String parentNodeId;
222-
223-
private long parentTaskId;
223+
private TaskId parentTaskId;
224224

225225
private boolean ban;
226226

227227
private String reason;
228228

229-
BanParentTaskRequest(String parentNodeId, long parentTaskId, String reason) {
230-
this.parentNodeId = parentNodeId;
229+
static BanParentTaskRequest createSetBanParentTaskRequest(TaskId parentTaskId, String reason) {
230+
return new BanParentTaskRequest(parentTaskId, reason);
231+
}
232+
233+
static BanParentTaskRequest createRemoveBanParentTaskRequest(TaskId parentTaskId) {
234+
return new BanParentTaskRequest(parentTaskId);
235+
}
236+
237+
private BanParentTaskRequest(TaskId parentTaskId, String reason) {
231238
this.parentTaskId = parentTaskId;
232239
this.ban = true;
233240
this.reason = reason;
234241
}
235242

236-
BanParentTaskRequest(String parentNodeId, long parentTaskId) {
237-
this.parentNodeId = parentNodeId;
243+
private BanParentTaskRequest(TaskId parentTaskId) {
238244
this.parentTaskId = parentTaskId;
239245
this.ban = false;
240246
}
@@ -245,8 +251,7 @@ public BanParentTaskRequest() {
245251
@Override
246252
public void readFrom(StreamInput in) throws IOException {
247253
super.readFrom(in);
248-
parentNodeId = in.readString();
249-
parentTaskId = in.readLong();
254+
parentTaskId = new TaskId(in);
250255
ban = in.readBoolean();
251256
if (ban) {
252257
reason = in.readString();
@@ -256,8 +261,7 @@ public void readFrom(StreamInput in) throws IOException {
256261
@Override
257262
public void writeTo(StreamOutput out) throws IOException {
258263
super.writeTo(out);
259-
out.writeString(parentNodeId);
260-
out.writeLong(parentTaskId);
264+
parentTaskId.writeTo(out);
261265
out.writeBoolean(ban);
262266
if (ban) {
263267
out.writeString(reason);
@@ -269,13 +273,13 @@ class BanParentRequestHandler implements TransportRequestHandler<BanParentTaskRe
269273
@Override
270274
public void messageReceived(final BanParentTaskRequest request, final TransportChannel channel) throws Exception {
271275
if (request.ban) {
272-
logger.debug("Received ban for the parent [{}:{}] on the node [{}], reason: [{}]", request.parentNodeId, request
273-
.parentTaskId, clusterService.localNode().getId(), request.reason);
274-
taskManager.setBan(request.parentNodeId, request.parentTaskId, request.reason);
276+
logger.debug("Received ban for the parent [{}] on the node [{}], reason: [{}]", request.parentTaskId,
277+
clusterService.localNode().getId(), request.reason);
278+
taskManager.setBan(request.parentTaskId, request.reason);
275279
} else {
276-
logger.debug("Removing ban for the parent [{}:{}] on the node [{}]", request.parentNodeId, request.parentTaskId,
280+
logger.debug("Removing ban for the parent [{}] on the node [{}]", request.parentTaskId,
277281
clusterService.localNode().getId());
278-
taskManager.removeBan(request.parentNodeId, request.parentTaskId);
282+
taskManager.removeBan(request.parentTaskId);
279283
}
280284
channel.sendResponse(TransportResponse.Empty.INSTANCE);
281285
}

‎core/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/ListTasksRequest.java‎

Lines changed: 1 addition & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -32,14 +32,6 @@ public class ListTasksRequest extends BaseTasksRequest<ListTasksRequest> {
3232

3333
private boolean detailed = false;
3434

35-
/**
36-
* Get information from nodes based on the nodes ids specified. If none are passed, information
37-
* for all nodes will be returned.
38-
*/
39-
public ListTasksRequest(String... nodesIds) {
40-
super(nodesIds);
41-
}
42-
4335
/**
4436
* Should the detailed task information be returned.
4537
*/
@@ -48,7 +40,7 @@ public boolean detailed() {
4840
}
4941

5042
/**
51-
* Should the node settings be returned.
43+
* Should the detailed task information be returned.
5244
*/
5345
public ListTasksRequest detailed(boolean detailed) {
5446
this.detailed = detailed;

‎core/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/ListTasksResponse.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -138,11 +138,13 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
138138
}
139139
builder.endObject();
140140
}
141-
builder.startArray("tasks");
141+
builder.startObject("tasks");
142142
for(TaskInfo task : entry.getValue()) {
143+
builder.startObject(task.getTaskId().toString(), XContentBuilder.FieldCaseConversion.NONE);
143144
task.toXContent(builder, params);
145+
builder.endObject();
144146
}
145-
builder.endArray();
147+
builder.endObject();
146148
builder.endObject();
147149
}
148150
builder.endObject();

‎core/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/list/TaskInfo.java‎

Lines changed: 20 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.elasticsearch.common.xcontent.ToXContent;
2727
import org.elasticsearch.common.xcontent.XContentBuilder;
2828
import org.elasticsearch.tasks.Task;
29+
import org.elasticsearch.tasks.TaskId;
2930

3031
import java.io.IOException;
3132

@@ -41,7 +42,7 @@ public class TaskInfo implements Writeable<TaskInfo>, ToXContent {
4142

4243
private final DiscoveryNode node;
4344

44-
private final long id;
45+
private final TaskId taskId;
4546

4647
private final String type;
4748

@@ -51,28 +52,21 @@ public class TaskInfo implements Writeable<TaskInfo>, ToXContent {
5152

5253
private final Task.Status status;
5354

54-
private final String parentNode;
55+
private final TaskId parentTaskId;
5556

56-
private final long parentId;
57-
58-
public TaskInfo(DiscoveryNode node, long id, String type, String action, String description, Task.Status status) {
59-
this(node, id, type, action, description, status, null, -1L);
60-
}
61-
62-
public TaskInfo(DiscoveryNode node, long id, String type, String action, String description, Task.Status status, String parentNode, long parentId) {
57+
public TaskInfo(DiscoveryNode node, long id, String type, String action, String description, Task.Status status, TaskId parentTaskId) {
6358
this.node = node;
64-
this.id = id;
59+
this.taskId = new TaskId(node.getId(), id);
6560
this.type = type;
6661
this.action = action;
6762
this.description = description;
6863
this.status = status;
69-
this.parentNode = parentNode;
70-
this.parentId = parentId;
64+
this.parentTaskId = parentTaskId;
7165
}
7266

7367
public TaskInfo(StreamInput in) throws IOException {
7468
node = DiscoveryNode.readNode(in);
75-
id = in.readLong();
69+
taskId = new TaskId(node.getId(), in.readLong());
7670
type = in.readString();
7771
action = in.readString();
7872
description = in.readOptionalString();
@@ -81,16 +75,19 @@ public TaskInfo(StreamInput in) throws IOException {
8175
} else {
8276
status = null;
8377
}
84-
parentNode = in.readOptionalString();
85-
parentId = in.readLong();
78+
parentTaskId = new TaskId(in);
79+
}
80+
81+
public TaskId getTaskId() {
82+
return taskId;
8683
}
8784

8885
public DiscoveryNode getNode() {
8986
return node;
9087
}
9188

9289
public long getId() {
93-
return id;
90+
return taskId.getId();
9491
}
9592

9693
public String getType() {
@@ -113,12 +110,8 @@ public Task.Status getStatus() {
113110
return status;
114111
}
115112

116-
public String getParentNode() {
117-
return parentNode;
118-
}
119-
120-
public long getParentId() {
121-
return parentId;
113+
public TaskId getParentTaskId() {
114+
return parentTaskId;
122115
}
123116

124117
@Override
@@ -129,7 +122,7 @@ public TaskInfo readFrom(StreamInput in) throws IOException {
129122
@Override
130123
public void writeTo(StreamOutput out) throws IOException {
131124
node.writeTo(out);
132-
out.writeLong(id);
125+
out.writeLong(taskId.getId());
133126
out.writeString(type);
134127
out.writeString(action);
135128
out.writeOptionalString(description);
@@ -139,15 +132,13 @@ public void writeTo(StreamOutput out) throws IOException {
139132
} else {
140133
out.writeBoolean(false);
141134
}
142-
out.writeOptionalString(parentNode);
143-
out.writeLong(parentId);
135+
parentTaskId.writeTo(out);
144136
}
145137

146138
@Override
147139
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
148-
builder.startObject();
149140
builder.field("node", node.getId());
150-
builder.field("id", id);
141+
builder.field("id", taskId.getId());
151142
builder.field("type", type);
152143
builder.field("action", action);
153144
if (status != null) {
@@ -156,11 +147,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
156147
if (description != null) {
157148
builder.field("description", description);
158149
}
159-
if (parentNode != null) {
160-
builder.field("parent_node", parentNode);
161-
builder.field("parent_id", parentId);
150+
if (parentTaskId.isSet() == false) {
151+
builder.field("parent_task_id", parentTaskId.toString());
162152
}
163-
builder.endObject();
164153
return builder;
165154
}
166155
}

0 commit comments

Comments
 (0)