honeymoose/OpenSearch
1
0
Fork 0
mirror of https://github.com/honeymoose/OpenSearch.git synced 2025-02-10 15:05:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add trace log for task cancellation (#55940)

Browse Source 
Adding trace logs to the task cancellation and its tests 
to debug the test failure in #55875.

Relates ##55875
This commit is contained in:
Nhat Nguyen 2020-04-29 15:04:45 -04:00
parent 144e8ce092
commit edbaa19a5d
3 changed files with 25 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
server/src/main/java/org/elasticsearch/action/admin/cluster/node/tasks/cancel/TransportCancelTasksAction.java
View File

@ -113,13 +113,19 @@ public class TransportCancelTasksAction extends TransportTasksAction<Cancellable
} }
void cancelTaskAndDescendants(CancellableTask task, String reason, boolean waitForCompletion, ActionListener<Void> listener) { void cancelTaskAndDescendants(CancellableTask task, String reason, boolean waitForCompletion, ActionListener<Void> listener) {
final TaskId taskId = task.taskInfo(clusterService.localNode().getId(), false).getTaskId();
if (task.shouldCancelChildrenOnCancellation()) { if (task.shouldCancelChildrenOnCancellation()) {
logger.trace("cancelling task [{}] and its descendants", taskId);
StepListener<Void> completedListener = new StepListener<>(); StepListener<Void> completedListener = new StepListener<>();
GroupedActionListener<Void> groupedListener = new GroupedActionListener<>(ActionListener.map(completedListener, r -> null), 3); GroupedActionListener<Void> groupedListener = new GroupedActionListener<>(ActionListener.map(completedListener, r -> null), 3);
Collection<DiscoveryNode> childrenNodes = Collection<DiscoveryNode> childrenNodes = taskManager.startBanOnChildrenNodes(task.getId(), () -> {
taskManager.startBanOnChildrenNodes(task.getId(), () -> groupedListener.onResponse(null)); logger.trace("child tasks of parent [{}] are completed", taskId);
taskManager.cancel(task, reason, () -> groupedListener.onResponse(null)); groupedListener.onResponse(null);
});
taskManager.cancel(task, reason, () -> {
logger.trace("task [{}] is cancelled", taskId);
groupedListener.onResponse(null);
});
StepListener<Void> banOnNodesListener = new StepListener<>(); StepListener<Void> banOnNodesListener = new StepListener<>();
setBanOnNodes(reason, waitForCompletion, task, childrenNodes, banOnNodesListener); setBanOnNodes(reason, waitForCompletion, task, childrenNodes, banOnNodesListener);
banOnNodesListener.whenComplete(groupedListener::onResponse, groupedListener::onFailure); banOnNodesListener.whenComplete(groupedListener::onResponse, groupedListener::onFailure);
@ -137,7 +143,7 @@ public class TransportCancelTasksAction extends TransportTasksAction<Cancellable
banOnNodesListener.whenComplete(r -> listener.onResponse(null), listener::onFailure); banOnNodesListener.whenComplete(r -> listener.onResponse(null), listener::onFailure);
} }
} else { } else {
logger.trace("task {} doesn't have any children that should be cancelled", task.getId()); logger.trace("task [{}] doesn't have any children that should be cancelled", taskId);
if (waitForCompletion) { if (waitForCompletion) {
taskManager.cancel(task, reason, () -> listener.onResponse(null)); taskManager.cancel(task, reason, () -> listener.onResponse(null));
} else { } else {
@ -153,23 +159,24 @@ public class TransportCancelTasksAction extends TransportTasksAction<Cancellable
listener.onResponse(null); listener.onResponse(null);
return; return;
} }
logger.trace("cancelling task {} on child nodes {}", task.getId(), childNodes); final TaskId taskId = new TaskId(clusterService.localNode().getId(), task.getId());
logger.trace("cancelling child tasks of [{}] on child nodes {}", taskId, childNodes);
GroupedActionListener<Void> groupedListener = GroupedActionListener<Void> groupedListener =
new GroupedActionListener<>(ActionListener.map(listener, r -> null), childNodes.size()); new GroupedActionListener<>(ActionListener.map(listener, r -> null), childNodes.size());
final BanParentTaskRequest banRequest = BanParentTaskRequest.createSetBanParentTaskRequest( final BanParentTaskRequest banRequest = BanParentTaskRequest.createSetBanParentTaskRequest(taskId, reason, waitForCompletion);
new TaskId(clusterService.localNode().getId(), task.getId()), reason, waitForCompletion);
for (DiscoveryNode node : childNodes) { for (DiscoveryNode node : childNodes) {
transportService.sendRequest(node, BAN_PARENT_ACTION_NAME, banRequest, transportService.sendRequest(node, BAN_PARENT_ACTION_NAME, banRequest,
new EmptyTransportResponseHandler(ThreadPool.Names.SAME) { new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {
@Override @Override
public void handleResponse(TransportResponse.Empty response) { public void handleResponse(TransportResponse.Empty response) {
logger.trace("sent ban for tasks with the parent [{}] to the node [{}]", taskId, node);
groupedListener.onResponse(null); groupedListener.onResponse(null);
} }
@Override @Override
public void handleException(TransportException exp) { public void handleException(TransportException exp) {
assert ExceptionsHelper.unwrapCause(exp) instanceof ElasticsearchSecurityException == false; assert ExceptionsHelper.unwrapCause(exp) instanceof ElasticsearchSecurityException == false;
logger.warn("Cannot send ban for tasks with the parent [{}] to the node [{}]", banRequest.parentTaskId, node); logger.warn("Cannot send ban for tasks with the parent [{}] to the node [{}]", taskId, node);
groupedListener.onFailure(exp); groupedListener.onFailure(exp);
} }
}); });

6
server/src/main/java/org/elasticsearch/tasks/TaskManager.java
View File

@ -197,8 +197,12 @@ public class TaskManager implements ClusterStateApplier {
public Releasable registerChildNode(long taskId, DiscoveryNode node) { public Releasable registerChildNode(long taskId, DiscoveryNode node) {
final CancellableTaskHolder holder = cancellableTasks.get(taskId); final CancellableTaskHolder holder = cancellableTasks.get(taskId);
if (holder != null) { if (holder != null) {
logger.trace("register child node [{}] task [{}]", node, taskId);
holder.registerChildNode(node); holder.registerChildNode(node);
return Releasables.releaseOnce(() -> holder.unregisterChildNode(node)); return Releasables.releaseOnce(() -> {
logger.trace("unregister child node [{}] task [{}]", node, taskId);
holder.unregisterChildNode(node);
});
} }
return () -> {}; return () -> {};
} }

4
server/src/test/java/org/elasticsearch/action/admin/cluster/node/tasks/CancellableTasksIT.java
View File

@ -52,6 +52,7 @@ import org.elasticsearch.tasks.TaskId;
import org.elasticsearch.tasks.TaskInfo; import org.elasticsearch.tasks.TaskInfo;
import org.elasticsearch.tasks.TaskManager; import org.elasticsearch.tasks.TaskManager;
import org.elasticsearch.test.ESIntegTestCase; import org.elasticsearch.test.ESIntegTestCase;
import org.elasticsearch.test.junit.annotations.TestIssueLogging;
import org.elasticsearch.threadpool.ThreadPool; import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.TransportException; import org.elasticsearch.transport.TransportException;
import org.elasticsearch.transport.TransportResponseHandler; import org.elasticsearch.transport.TransportResponseHandler;
@ -79,6 +80,9 @@ import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasSize; import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.instanceOf;
@TestIssueLogging(
value = "org.elasticsearch.action.admin.cluster.node.tasks.cancel:TRACE,org.elasticsearch.tasks:TRACE",
issueUrl = "https://github.com/elastic/elasticsearch/issues/55875")
public class CancellableTasksIT extends ESIntegTestCase { public class CancellableTasksIT extends ESIntegTestCase {
static int idGenerator = 0; static int idGenerator = 0;