Skip to content

Commit 95b7158

Browse files
edsavageedsavage
committed
[ML] Add "close_job" parameter to the stop datafeed API
When the stop datafeed API is called with close_job=true the job associated with the datafeed will be closed first, which in turn will result in the datafeed being closed. Closes #138010
1 parent e6c5dcc commit 95b7158

File tree

6 files changed

+179
-9
lines changed

6 files changed

+179
-9
lines changed

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/StopDatafeedAction.java‎

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ public static class Request extends BaseTasksRequest<Request> implements ToXCont
4242
public static final ParseField TIMEOUT = new ParseField("timeout");
4343
public static final ParseField FORCE = new ParseField("force");
4444
public static final ParseField ALLOW_NO_MATCH = new ParseField("allow_no_match");
45+
public static final ParseField CLOSE_JOB = new ParseField("close_job");
4546

4647
public static final ObjectParser<Request, Void> PARSER = new ObjectParser<>(NAME, Request::new);
4748
static {
@@ -52,6 +53,7 @@ public static class Request extends BaseTasksRequest<Request> implements ToXCont
5253
);
5354
PARSER.declareBoolean(Request::setForce, FORCE);
5455
PARSER.declareBoolean(Request::setAllowNoMatch, ALLOW_NO_MATCH);
56+
PARSER.declareBoolean(Request::setCloseJob, CLOSE_JOB);
5557
}
5658

5759
public static Request parseRequest(String datafeedId, XContentParser parser) {
@@ -67,6 +69,7 @@ public static Request parseRequest(String datafeedId, XContentParser parser) {
6769
private TimeValue stopTimeout = DEFAULT_TIMEOUT;
6870
private boolean force = false;
6971
private boolean allowNoMatch = true;
72+
private boolean closeJob = false;
7073

7174
public Request(String datafeedId) {
7275
this.datafeedId = ExceptionsHelper.requireNonNull(datafeedId, DatafeedConfig.ID.getPreferredName());
@@ -81,6 +84,7 @@ public Request(StreamInput in) throws IOException {
8184
stopTimeout = in.readTimeValue();
8285
force = in.readBoolean();
8386
allowNoMatch = in.readBoolean();
87+
closeJob = in.readBoolean();
8488
}
8589

8690
public String getDatafeedId() {
@@ -124,6 +128,15 @@ public Request setAllowNoMatch(boolean allowNoMatch) {
124128
return this;
125129
}
126130

131+
public boolean closeJob() {
132+
return closeJob;
133+
}
134+
135+
public Request setCloseJob(boolean closeJob) {
136+
this.closeJob = closeJob;
137+
return this;
138+
}
139+
127140
@Override
128141
public boolean match(Task task) {
129142
for (String id : resolvedStartedDatafeedIds) {
@@ -148,11 +161,12 @@ public void writeTo(StreamOutput out) throws IOException {
148161
out.writeTimeValue(stopTimeout);
149162
out.writeBoolean(force);
150163
out.writeBoolean(allowNoMatch);
164+
out.writeBoolean(closeJob);
151165
}
152166

153167
@Override
154168
public int hashCode() {
155-
return Objects.hash(datafeedId, stopTimeout, force, allowNoMatch);
169+
return Objects.hash(datafeedId, stopTimeout, force, allowNoMatch, closeJob);
156170
}
157171

158172
@Override
@@ -162,6 +176,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
162176
builder.field(TIMEOUT.getPreferredName(), stopTimeout.getStringRep());
163177
builder.field(FORCE.getPreferredName(), force);
164178
builder.field(ALLOW_NO_MATCH.getPreferredName(), allowNoMatch);
179+
builder.field(CLOSE_JOB.getPreferredName(), closeJob);
165180
builder.endObject();
166181
return builder;
167182
}
@@ -178,7 +193,8 @@ public boolean equals(Object obj) {
178193
return Objects.equals(datafeedId, other.datafeedId)
179194
&& Objects.equals(stopTimeout, other.stopTimeout)
180195
&& Objects.equals(force, other.force)
181-
&& Objects.equals(allowNoMatch, other.allowNoMatch);
196+
&& Objects.equals(allowNoMatch, other.allowNoMatch)
197+
&& Objects.equals(closeJob, other.closeJob);
182198
}
183199
}
184200

‎x-pack/plugin/ml/qa/native-multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/ml/integration/DatafeedJobsIT.java‎

Lines changed: 84 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -487,19 +487,101 @@ public void testRealtime() throws Exception {
487487
startRealtime(jobId);
488488

489489
try {
490-
StopDatafeedAction.Response stopJobResponse = stopDatafeed(datafeedId);
491-
assertTrue(stopJobResponse.isStopped());
490+
StopDatafeedAction.Response stopDatafeedResponse = stopDatafeed(datafeedId);
491+
assertTrue(stopDatafeedResponse.isStopped());
492+
} catch (Exception e) {
493+
HotThreads.logLocalHotThreads(logger, Level.INFO, "hot threads at failure", ReferenceDocs.LOGGING);
494+
throw e;
495+
}
496+
assertBusy(() -> {
497+
GetDatafeedsStatsAction.Request request = new GetDatafeedsStatsAction.Request(datafeedId);
498+
GetDatafeedsStatsAction.Response response = client().execute(GetDatafeedsStatsAction.INSTANCE, request).actionGet();
499+
assertThat(response.getResponse().results().get(0).getDatafeedState(), equalTo(DatafeedState.STOPPED));
500+
});
501+
502+
// The job should _not_ have closed automatically
503+
assertBusy(() -> {
504+
GetJobsStatsAction.Request request = new GetJobsStatsAction.Request(jobId);
505+
GetJobsStatsAction.Response response = client().execute(GetJobsStatsAction.INSTANCE, request).actionGet();
506+
assertThat(response.getResponse().results().get(0).getState(), equalTo(JobState.OPENED));
507+
});
508+
}
509+
510+
private void doTestRealtime_GivenCloseJobParameter(String jobId, boolean closeJobParameter, JobState jobState) throws Exception {
511+
String datafeedId = jobId + "-datafeed";
512+
startRealtime(jobId);
513+
514+
try {
515+
StopDatafeedAction.Response stopDatafeedResponse = stopDatafeed(datafeedId, true);
516+
assertTrue(stopDatafeedResponse.isStopped());
492517
} catch (Exception e) {
493518
HotThreads.logLocalHotThreads(logger, Level.INFO, "hot threads at failure", ReferenceDocs.LOGGING);
494519
throw e;
495520
}
521+
522+
// The job should have closed automatically
523+
assertBusy(() -> {
524+
GetJobsStatsAction.Request request = new GetJobsStatsAction.Request(jobId);
525+
GetJobsStatsAction.Response response = client().execute(GetJobsStatsAction.INSTANCE, request).actionGet();
526+
assertThat(response.getResponse().results().get(0).getState(), equalTo(JobState.CLOSED));
527+
});
528+
496529
assertBusy(() -> {
497530
GetDatafeedsStatsAction.Request request = new GetDatafeedsStatsAction.Request(datafeedId);
498531
GetDatafeedsStatsAction.Response response = client().execute(GetDatafeedsStatsAction.INSTANCE, request).actionGet();
499532
assertThat(response.getResponse().results().get(0).getDatafeedState(), equalTo(DatafeedState.STOPPED));
500533
});
501534
}
502535

536+
public void testRealtime_GivenCloseJobParameterIsTrue() throws Exception {
537+
doTestRealtime_GivenCloseJobParameter("realtime-job-close-job-true", true, JobState.CLOSED);
538+
}
539+
540+
public void testRealtime_GivenCloseJobParameterIsFalse() throws Exception {
541+
doTestRealtime_GivenCloseJobParameter("realtime-job-close-job-false", false, JobState.OPENED);
542+
}
543+
544+
private void doTestStopLookback_GivenCloseJobParameter(String jobId, boolean closeJobParameter, JobState jobState) throws Exception {
545+
String datafeedId = jobId + "-datafeed";
546+
547+
client().admin().indices().prepareCreate("data").setMapping("time", "type=date").get();
548+
long numDocs = randomIntBetween(1024, 2048);
549+
long now = System.currentTimeMillis();
550+
long oneWeekAgo = now - 604800000;
551+
long twoWeeksAgo = oneWeekAgo - 604800000;
552+
indexDocs(logger, "data", numDocs, twoWeeksAgo, oneWeekAgo);
553+
554+
Job.Builder job = createScheduledJob(jobId);
555+
putJob(job);
556+
openJob(job.getId());
557+
assertBusy(() -> assertEquals(getJobStats(job.getId()).get(0).getState(), JobState.OPENED));
558+
559+
DatafeedConfig.Builder datafeedConfigBuilder = createDatafeedBuilder(datafeedId, jobId, Collections.singletonList("data"));
560+
// Use lots of chunks to maximise the chance that we can stop the lookback before it completes
561+
datafeedConfigBuilder.setChunkingConfig(ChunkingConfig.newManual(new TimeValue(1, TimeUnit.SECONDS)));
562+
DatafeedConfig datafeedConfig = datafeedConfigBuilder.build();
563+
putDatafeed(datafeedConfig);
564+
startDatafeed(datafeedConfig.getId(), 0L, now);
565+
assertBusy(() -> assertThat(getDataCounts(job.getId()).getProcessedRecordCount(), greaterThan(0L)), 60, TimeUnit.SECONDS);
566+
567+
// Stop the datafeed with the given close_job parameter
568+
StopDatafeedAction.Response stopDatafeedResponse = stopDatafeed(datafeedId, closeJobParameter);
569+
assertTrue(stopDatafeedResponse.isStopped());
570+
571+
// Check the job state is as expected
572+
assertBusy(() -> assertEquals(jobState, getJobStats(jobId).get(0).getState()));
573+
}
574+
575+
public void testStopLookback_GivenCloseJobParameterIsTrue() throws Exception {
576+
// Stop the datafeed with close_job=true
577+
doTestStopLookback_GivenCloseJobParameter("lookback-stop-close-job-true", true, JobState.CLOSED);
578+
}
579+
580+
public void testStopLookback_GivenCloseJobParameterIsFalse() throws Exception {
581+
// Stop the datafeed with close_job=false
582+
doTestStopLookback_GivenCloseJobParameter("lookback-stop-close-job-false", false, JobState.OPENED);
583+
}
584+
503585
public void testCloseJobStopsRealtimeDatafeed() throws Exception {
504586
String jobId = "realtime-close-job";
505587
String datafeedId = jobId + "-datafeed";

‎x-pack/plugin/ml/qa/native-multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/ml/integration/MlNativeAutodetectIntegTestCase.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -141,6 +141,12 @@ protected StopDatafeedAction.Response stopDatafeed(String datafeedId) {
141141
return client().execute(StopDatafeedAction.INSTANCE, request).actionGet();
142142
}
143143

144+
protected StopDatafeedAction.Response stopDatafeed(String datafeedId, boolean closeJob) {
145+
StopDatafeedAction.Request request = new StopDatafeedAction.Request(datafeedId);
146+
request.setCloseJob(closeJob);
147+
return client().execute(StopDatafeedAction.INSTANCE, request).actionGet();
148+
}
149+
144150
protected PutDatafeedAction.Response updateDatafeed(DatafeedUpdate update) {
145151
UpdateDatafeedAction.Request request = new UpdateDatafeedAction.Request(update);
146152
return client().execute(UpdateDatafeedAction.INSTANCE, request).actionGet();

‎x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportStopDatafeedAction.java‎

Lines changed: 67 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import org.elasticsearch.action.FailedNodeException;
1616
import org.elasticsearch.action.TaskOperationFailure;
1717
import org.elasticsearch.action.support.ActionFilters;
18+
import org.elasticsearch.action.support.master.AcknowledgedResponse;
1819
import org.elasticsearch.action.support.tasks.TransportTasksAction;
1920
import org.elasticsearch.client.internal.Client;
2021
import org.elasticsearch.client.internal.OriginSettingClient;
@@ -27,7 +28,9 @@
2728
import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
2829
import org.elasticsearch.common.util.concurrent.EsExecutors;
2930
import org.elasticsearch.core.FixForMultiProject;
31+
import org.elasticsearch.core.Predicates;
3032
import org.elasticsearch.core.TimeValue;
33+
import org.elasticsearch.core.Tuple;
3134
import org.elasticsearch.discovery.MasterNotDiscoveredException;
3235
import org.elasticsearch.injection.guice.Inject;
3336
import org.elasticsearch.persistent.PersistentTasksClusterService;
@@ -39,6 +42,7 @@
3942
import org.elasticsearch.transport.TransportResponseHandler;
4043
import org.elasticsearch.transport.TransportService;
4144
import org.elasticsearch.xpack.core.ml.MlTasks;
45+
import org.elasticsearch.xpack.core.ml.action.CloseJobAction;
4246
import org.elasticsearch.xpack.core.ml.action.StartDatafeedAction;
4347
import org.elasticsearch.xpack.core.ml.action.StopDatafeedAction;
4448
import org.elasticsearch.xpack.core.ml.datafeed.DatafeedState;
@@ -48,6 +52,7 @@
4852
import org.elasticsearch.xpack.ml.MachineLearning;
4953
import org.elasticsearch.xpack.ml.datafeed.persistence.DatafeedConfigProvider;
5054
import org.elasticsearch.xpack.ml.notifications.AnomalyDetectionAuditor;
55+
import org.elasticsearch.xpack.ml.utils.TypedChainTaskExecutor;
5156

5257
import java.util.ArrayList;
5358
import java.util.Collection;
@@ -58,8 +63,10 @@
5863
import java.util.concurrent.atomic.AtomicInteger;
5964
import java.util.stream.Collectors;
6065

66+
import static java.util.stream.Collectors.toList;
6167
import static org.elasticsearch.core.Strings.format;
6268
import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN;
69+
import static org.elasticsearch.xpack.core.ClientHelper.executeAsyncWithOrigin;
6370
import static org.elasticsearch.xpack.ml.utils.ExceptionCollectionHandling.exceptionArrayToStatusException;
6471

6572
public class TransportStopDatafeedAction extends TransportTasksAction<
@@ -106,7 +113,7 @@ public TransportStopDatafeedAction(
106113
}
107114

108115
/**
109-
* Sort the datafeed IDs the their task state and add to one
116+
* Sort the datafeed IDs by their task state and add to one
110117
* of the list arguments depending on the state.
111118
*
112119
* @param expandedDatafeedIds The expanded set of IDs
@@ -211,7 +218,64 @@ private void doExecute(
211218
if (request.isForce()) {
212219
forceStopDatafeed(request, listener, tasks, nodes, notStoppedDatafeeds);
213220
} else {
214-
normalStopDatafeed(task, request, listener, tasks, nodes, startedDatafeeds, stoppingDatafeeds, attempt);
221+
final List<String> startedDatafeedsJobs = new ArrayList<>();
222+
for (String datafeedId : startedDatafeeds) {
223+
PersistentTasksCustomMetadata.PersistentTask<?> datafeedTask = MlTasks.getDatafeedTask(datafeedId, tasks);
224+
if (datafeedTask != null
225+
&& PersistentTasksClusterService.needsReassignment(datafeedTask.getAssignment(), nodes) == false) {
226+
startedDatafeedsJobs.add(((StartDatafeedAction.DatafeedParams) datafeedTask.getParams()).getJobId());
227+
}
228+
}
229+
if (request.closeJob() && startedDatafeedsJobs.isEmpty() == false) {
230+
// If the "close_job" parameter was set to "true" on the stop datafeed request we attempt to first close the
231+
// jobs associated with the datafeeds. This will in turn attempt to stop the jobs' datafeeds (this time with the
232+
// "close_job" flag set to false, to avoid recursion)
233+
ActionListener<List<Tuple<String, AcknowledgedResponse>>> closeJobActionListener = listener
234+
.delegateFailureAndWrap((delegate, jobsResponses) -> {
235+
List<String> jobIds = jobsResponses.stream()
236+
.filter(t -> t.v2().isAcknowledged() == false)
237+
.map(Tuple::v1)
238+
.collect(toList());
239+
if (jobIds.isEmpty()) {
240+
logger.debug("Successfully closed jobs (and associated datafeeds)");
241+
} else {
242+
logger.warn("Failed to close jobs (and associated datafeeds): {}", jobIds);
243+
}
244+
delegate.onResponse(new StopDatafeedAction.Response(true));
245+
});
246+
247+
TypedChainTaskExecutor<Tuple<String, AcknowledgedResponse>> chainTaskExecutor = new TypedChainTaskExecutor<>(
248+
EsExecutors.DIRECT_EXECUTOR_SERVICE,
249+
Predicates.always(),
250+
Predicates.always()
251+
);
252+
for (String jobId : startedDatafeedsJobs) {
253+
chainTaskExecutor.add(
254+
al -> executeAsyncWithOrigin(
255+
client,
256+
ML_ORIGIN,
257+
CloseJobAction.INSTANCE,
258+
new CloseJobAction.Request(jobId),
259+
listener.delegateFailureAndWrap(
260+
(l, response) -> l.onResponse(new StopDatafeedAction.Response(response.isClosed()))
261+
)
262+
)
263+
);
264+
}
265+
chainTaskExecutor.execute(closeJobActionListener);
266+
} else {
267+
normalStopDatafeed(
268+
task,
269+
request,
270+
listener,
271+
tasks,
272+
nodes,
273+
startedDatafeeds,
274+
stoppingDatafeeds,
275+
startedDatafeedsJobs,
276+
attempt
277+
);
278+
}
215279
}
216280
}, listener::onFailure)
217281
);
@@ -226,10 +290,10 @@ private void normalStopDatafeed(
226290
DiscoveryNodes nodes,
227291
List<String> startedDatafeeds,
228292
List<String> stoppingDatafeeds,
293+
List<String> startedDatafeedsJobs,
229294
int attempt
230295
) {
231296
final Set<String> executorNodes = new HashSet<>();
232-
final List<String> startedDatafeedsJobs = new ArrayList<>();
233297
final List<String> resolvedStartedDatafeeds = new ArrayList<>();
234298
final List<PersistentTasksCustomMetadata.PersistentTask<?>> allDataFeedsToWaitFor = new ArrayList<>();
235299
for (String datafeedId : startedDatafeeds) {
@@ -240,7 +304,6 @@ private void normalStopDatafeed(
240304
assert datafeedTask != null : msg;
241305
logger.error(msg);
242306
} else if (PersistentTasksClusterService.needsReassignment(datafeedTask.getAssignment(), nodes) == false) {
243-
startedDatafeedsJobs.add(((StartDatafeedAction.DatafeedParams) datafeedTask.getParams()).getJobId());
244307
resolvedStartedDatafeeds.add(datafeedId);
245308
executorNodes.add(datafeedTask.getExecutorNode());
246309
allDataFeedsToWaitFor.add(datafeedTask);

‎x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/persistence/DatafeedConfigProvider.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,7 @@
8181
* This class implements CRUD operation for the
8282
* datafeed configuration document
8383
*
84-
* The number of datafeeds returned in a search it limited to
84+
* The number of datafeeds returned in a search is limited to
8585
* {@link MlConfigIndex#CONFIG_INDEX_MAX_RESULTS_WINDOW}.
8686
* In most cases we expect 10s or 100s of datafeeds to be defined and
8787
* a search for all datafeeds should return all.

‎x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/rest/datafeeds/RestStopDatafeedAction.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,9 @@ protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient
5959
request.setForce(restRequest.paramAsBoolean(Request.FORCE.getPreferredName(), request.isForce()));
6060
}
6161
request.setAllowNoMatch(restRequest.paramAsBoolean(Request.ALLOW_NO_MATCH.getPreferredName(), request.allowNoMatch()));
62+
if (restRequest.hasParam(Request.CLOSE_JOB.getPreferredName())) {
63+
request.setCloseJob(restRequest.paramAsBoolean(Request.CLOSE_JOB.getPreferredName(), request.closeJob()));
64+
}
6265
}
6366
return channel -> client.execute(StopDatafeedAction.INSTANCE, request, new RestBuilderListener<Response>(channel) {
6467

0 commit comments

Comments
 (0)