Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Transform] Make validate calls cancellable #110951

Merged
merged 4 commits into from
Jul 18, 2024
Merged

[Transform] Make validate calls cancellable #110951

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
27666b6
[Transform] Make validate calls cancellable
prwhelan Jul 16, 2024
6a23c7d
Update docs/changelog/110951.yaml
prwhelan Jul 16, 2024
c7b1251
Create CancellableTasks and NodeClients
prwhelan Jul 17, 2024
8491115
Merge branch 'main' into fix/1273
elasticmachine Jul 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions docs/changelog/110951.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
pr: 110951
summary: Make validate calls cancellable
prwhelan marked this conversation as resolved.
Show resolved Hide resolved
area: Transform
type: bug
issues: []
6 changes: 5 additions & 1 deletion ...m/src/main/java/org/elasticsearch/xpack/transform/action/TransportPutTransformAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.support.master.AcknowledgedTransportMasterNodeAction;
import org.elasticsearch.client.internal.Client;
import org.elasticsearch.client.internal.ParentTaskAssigningClient;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.block.ClusterBlockException;
import org.elasticsearch.cluster.block.ClusterBlockLevel;
Expand All @@ -25,6 +26,7 @@
import org.elasticsearch.common.util.concurrent.EsExecutors;
import org.elasticsearch.persistent.PersistentTasksCustomMetadata;
import org.elasticsearch.tasks.Task;
import org.elasticsearch.tasks.TaskId;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.TransportService;
import org.elasticsearch.xpack.core.ClientHelper;
Expand Down Expand Up @@ -110,9 +112,11 @@ protected void masterOperation(Task task, Request request, ClusterState clusterS
);

// <2> Validate source and destination indices

var parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
ActionListener<Void> checkPrivilegesListener = validateTransformListener.delegateFailureAndWrap(
(l, aVoid) -> ClientHelper.executeAsyncWithOrigin(
client,
new ParentTaskAssigningClient(client, parentTaskId),
ClientHelper.TRANSFORM_ORIGIN,
ValidateTransformAction.INSTANCE,
new ValidateTransformAction.Request(config, request.isDeferValidation(), request.ackTimeout()),
Expand Down
18 changes: 11 additions & 7 deletions ...src/main/java/org/elasticsearch/xpack/transform/action/TransportStartTransformAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.master.TransportMasterNodeAction;
import org.elasticsearch.client.internal.Client;
import org.elasticsearch.client.internal.ParentTaskAssigningClient;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.block.ClusterBlockException;
import org.elasticsearch.cluster.block.ClusterBlockLevel;
Expand All @@ -31,6 +32,7 @@
import org.elasticsearch.persistent.PersistentTasksService;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.tasks.Task;
import org.elasticsearch.tasks.TaskId;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.TransportService;
import org.elasticsearch.xpack.core.ClientHelper;
Expand Down Expand Up @@ -126,23 +128,25 @@ protected TransportStartTransformAction(

@Override
protected void masterOperation(
Task ignoredTask,
Task task,
StartTransformAction.Request request,
ClusterState state,
ActionListener<StartTransformAction.Response> listener
) {
TransformNodes.warnIfNoTransformNodes(state);

final SetOnce<TransformTaskParams> transformTaskParamsHolder = new SetOnce<>();
final SetOnce<TransformConfig> transformConfigHolder = new SetOnce<>();
var transformTaskParamsHolder = new SetOnce<TransformTaskParams>();
var transformConfigHolder = new SetOnce<TransformConfig>();
var parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
var parentClient = new ParentTaskAssigningClient(client, parentTaskId);

// <5> Wait for the allocated task's state to STARTED
ActionListener<PersistentTasksCustomMetadata.PersistentTask<TransformTaskParams>> newPersistentTaskActionListener = ActionListener
.wrap(task -> {
.wrap(t -> {
TransformTaskParams transformTask = transformTaskParamsHolder.get();
assert transformTask != null;
waitForTransformTaskStarted(
task.getId(),
t.getId(),
transformTask,
request.ackTimeout(),
ActionListener.wrap(taskStarted -> listener.onResponse(new StartTransformAction.Response(true)), listener::onFailure)
Expand Down Expand Up @@ -196,7 +200,7 @@ protected void masterOperation(
return;
}
TransformIndex.createDestinationIndex(
client,
parentClient,
auditor,
indexNameExpressionResolver,
state,
Expand Down Expand Up @@ -257,7 +261,7 @@ protected void masterOperation(
)
);
ClientHelper.executeAsyncWithOrigin(
client,
parentClient,
ClientHelper.TRANSFORM_ORIGIN,
ValidateTransformAction.INSTANCE,
new ValidateTransformAction.Request(config, false, request.ackTimeout()),
Expand Down
14 changes: 8 additions & 6 deletions .../main/java/org/elasticsearch/xpack/transform/action/TransportValidateTransformAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.HandledTransportAction;
import org.elasticsearch.client.internal.Client;
import org.elasticsearch.client.internal.ParentTaskAssigningClient;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.cluster.node.DiscoveryNode;
Expand All @@ -23,15 +24,14 @@
import org.elasticsearch.license.License;
import org.elasticsearch.license.RemoteClusterLicenseChecker;
import org.elasticsearch.tasks.Task;
import org.elasticsearch.tasks.TaskId;
import org.elasticsearch.transport.TransportService;
import org.elasticsearch.xpack.core.common.validation.SourceDestValidator;
import org.elasticsearch.xpack.core.transform.TransformDeprecations;
import org.elasticsearch.xpack.core.transform.TransformMessages;
import org.elasticsearch.xpack.core.transform.action.ValidateTransformAction;
import org.elasticsearch.xpack.core.transform.action.ValidateTransformAction.Request;
import org.elasticsearch.xpack.core.transform.action.ValidateTransformAction.Response;
import org.elasticsearch.xpack.core.transform.transforms.TransformConfig;
import org.elasticsearch.xpack.transform.transforms.Function;
import org.elasticsearch.xpack.transform.transforms.FunctionFactory;
import org.elasticsearch.xpack.transform.transforms.TransformNodes;
import org.elasticsearch.xpack.transform.utils.SourceDestValidations;
Expand Down Expand Up @@ -99,8 +99,10 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li

TransformNodes.warnIfNoTransformNodes(clusterState);

final TransformConfig config = request.getConfig();
final Function function = FunctionFactory.create(config);
var config = request.getConfig();
var function = FunctionFactory.create(config);
var parentTaskId = new TaskId(clusterService.localNode().getId(), task.getId());
var parentClient = new ParentTaskAssigningClient(client, parentTaskId);

if (config.getVersion() == null || config.getVersion().before(TransformDeprecations.MIN_TRANSFORM_VERSION)) {
listener.onFailure(
Expand Down Expand Up @@ -130,7 +132,7 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
if (request.isDeferValidation()) {
deduceMappingsListener.onResponse(emptyMap());
} else {
function.deduceMappings(client, config.getHeaders(), config.getId(), config.getSource(), deduceMappingsListener);
function.deduceMappings(parentClient, config.getHeaders(), config.getId(), config.getSource(), deduceMappingsListener);
}
}, listener::onFailure);

Expand All @@ -139,7 +141,7 @@ protected void doExecute(Task task, Request request, ActionListener<Response> li
if (request.isDeferValidation()) {
validateQueryListener.onResponse(true);
} else {
function.validateQuery(client, config.getHeaders(), config.getSource(), request.ackTimeout(), validateQueryListener);
function.validateQuery(parentClient, config.getHeaders(), config.getSource(), request.ackTimeout(), validateQueryListener);
}
}, listener::onFailure);

Expand Down