Skip to content

Commit 350c2b9

Browse files
dimitris-athanasioudimitris-athanasiou
authored
[ML] Provide a way to revert an AD job to an empty snapshot (#65431)
This commit adds a way to revert an anomaly detection job to the empty snapshot. Combining this with `delete_intervening_results` to `true`, the user can now reset the job and start over. The API call looks like this: POST _ml/anomaly_detectors/<job_id>/model_snapshots/empty/_revert
1 parent 699af9d commit 350c2b9

File tree

9 files changed

+131
-17
lines changed

9 files changed

+131
-17
lines changed

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/job/config/JobUpdate.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
import org.elasticsearch.common.xcontent.ObjectParser;
1818
import org.elasticsearch.common.xcontent.ToXContentObject;
1919
import org.elasticsearch.common.xcontent.XContentBuilder;
20+
import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.ModelSnapshot;
2021
import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
2122

2223
import java.io.IOException;
@@ -469,7 +470,7 @@ public Job mergeWithJob(Job source, ByteSizeValue maxModelMemoryLimit) {
469470
builder.setCustomSettings(customSettings);
470471
}
471472
if (modelSnapshotId != null) {
472-
builder.setModelSnapshotId(modelSnapshotId);
473+
builder.setModelSnapshotId(ModelSnapshot.isTheEmptySnapshot(modelSnapshotId) ? null : modelSnapshotId);
473474
}
474475
if (modelSnapshotMinVersion != null) {
475476
builder.setModelSnapshotMinVersion(modelSnapshotMinVersion);

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/job/process/autodetect/state/ModelSnapshot.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@ private static ObjectParser<Builder, Void> createParser(boolean ignoreUnknownFie
8080
return parser;
8181
}
8282

83+
private static String EMPTY_SNAPSHOT_ID = "empty";
8384

8485
private final String jobId;
8586

@@ -285,6 +286,14 @@ public List<String> stateDocumentIds() {
285286
return stateDocumentIds;
286287
}
287288

289+
public boolean isTheEmptySnapshot() {
290+
return isTheEmptySnapshot(snapshotId);
291+
}
292+
293+
public static boolean isTheEmptySnapshot(String snapshotId) {
294+
return EMPTY_SNAPSHOT_ID.equals(snapshotId);
295+
}
296+
288297
public static String documentIdPrefix(String jobId) {
289298
return jobId + "_" + TYPE + "_";
290299
}
@@ -435,4 +444,9 @@ public ModelSnapshot build() {
435444
latestRecordTimeStamp, latestResultTimeStamp, quantiles, retain);
436445
}
437446
}
447+
448+
public static ModelSnapshot emptySnapshot(String jobId) {
449+
return new ModelSnapshot(jobId, Version.CURRENT, new Date(), "empty snapshot", EMPTY_SNAPSHOT_ID, 0,
450+
new ModelSizeStats.Builder(jobId).build(), null, null, null, false);
451+
}
438452
}

‎x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/job/config/JobUpdateTests.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import org.elasticsearch.common.xcontent.XContentParser;
1616
import org.elasticsearch.test.AbstractSerializingTestCase;
1717
import org.elasticsearch.test.VersionUtils;
18+
import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.ModelSnapshot;
1819

1920
import java.util.ArrayList;
2021
import java.util.Arrays;
@@ -26,7 +27,9 @@
2627
import java.util.Set;
2728

2829
import static org.hamcrest.Matchers.equalTo;
30+
import static org.hamcrest.Matchers.is;
2931
import static org.hamcrest.Matchers.not;
32+
import static org.hamcrest.Matchers.nullValue;
3033
import static org.mockito.Mockito.mock;
3134

3235
public class JobUpdateTests extends AbstractSerializingTestCase<JobUpdate> {
@@ -369,4 +372,23 @@ public void testUpdate_withAnalysisLimitsPreviouslyUndefined() {
369372

370373
updateAboveMaxLimit.mergeWithJob(jobBuilder.build(), new ByteSizeValue(10000L, ByteSizeUnit.MB));
371374
}
375+
376+
public void testUpdate_givenEmptySnapshot() {
377+
Job.Builder jobBuilder = new Job.Builder("my_job");
378+
Detector.Builder d1 = new Detector.Builder("count", null);
379+
AnalysisConfig.Builder ac = new AnalysisConfig.Builder(Collections.singletonList(d1.build()));
380+
jobBuilder.setAnalysisConfig(ac);
381+
jobBuilder.setDataDescription(new DataDescription.Builder());
382+
jobBuilder.setCreateTime(new Date());
383+
jobBuilder.setModelSnapshotId("some_snapshot_id");
384+
Job job = jobBuilder.build();
385+
assertThat(job.getModelSnapshotId(), equalTo("some_snapshot_id"));
386+
387+
JobUpdate update = new JobUpdate.Builder(job.getId())
388+
.setModelSnapshotId(ModelSnapshot.emptySnapshot(job.getId()).getSnapshotId())
389+
.build();
390+
391+
Job updatedJob = update.mergeWithJob(job, ByteSizeValue.ofMb(100));
392+
assertThat(updatedJob.getModelSnapshotId(), is(nullValue()));
393+
}
372394
}

‎x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/job/process/autodetect/state/ModelSnapshotTests.java‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@
1818

1919
import static org.hamcrest.Matchers.containsString;
2020
import static org.hamcrest.Matchers.equalTo;
21+
import static org.hamcrest.Matchers.is;
22+
import static org.hamcrest.Matchers.nullValue;
2123

2224
public class ModelSnapshotTests extends AbstractSerializingTestCase<ModelSnapshot> {
2325
private static final Date DEFAULT_TIMESTAMP = new Date();
@@ -155,7 +157,7 @@ public static ModelSnapshot createRandomized() {
155157
modelSnapshot.setMinVersion(Version.CURRENT);
156158
modelSnapshot.setTimestamp(new Date(TimeValue.parseTimeValue(randomTimeValue(), "test").millis()));
157159
modelSnapshot.setDescription(randomAlphaOfLengthBetween(1, 20));
158-
modelSnapshot.setSnapshotId(randomAlphaOfLengthBetween(1, 20));
160+
modelSnapshot.setSnapshotId(randomAlphaOfLength(10));
159161
modelSnapshot.setSnapshotDocCount(randomInt());
160162
modelSnapshot.setModelSizeStats(ModelSizeStatsTests.createRandomized());
161163
modelSnapshot.setLatestResultTimeStamp(
@@ -214,4 +216,18 @@ public void testLenientParser() throws IOException {
214216
ModelSnapshot.LENIENT_PARSER.apply(parser, null);
215217
}
216218
}
219+
220+
public void testEmptySnapshot() {
221+
ModelSnapshot modelSnapshot = ModelSnapshot.emptySnapshot("my_job");
222+
assertThat(modelSnapshot.getSnapshotId(), equalTo("empty"));
223+
assertThat(modelSnapshot.isTheEmptySnapshot(), is(true));
224+
assertThat(modelSnapshot.getMinVersion(), equalTo(Version.CURRENT));
225+
assertThat(modelSnapshot.getLatestRecordTimeStamp(), is(nullValue()));
226+
assertThat(modelSnapshot.getLatestResultTimeStamp(), is(nullValue()));
227+
}
228+
229+
public void testIsEmpty_GivenNonEmptySnapshot() {
230+
ModelSnapshot modelSnapshot = createRandomized();
231+
assertThat(modelSnapshot.isTheEmptySnapshot(), is(false));
232+
}
217233
}

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

Lines changed: 50 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,16 +18,19 @@
1818
import org.elasticsearch.index.query.QueryBuilders;
1919
import org.elasticsearch.rest.RestStatus;
2020
import org.elasticsearch.search.SearchHits;
21+
import org.elasticsearch.xpack.core.ml.action.RevertModelSnapshotAction;
2122
import org.elasticsearch.xpack.core.ml.annotations.Annotation;
2223
import org.elasticsearch.xpack.core.ml.annotations.Annotation.Event;
2324
import org.elasticsearch.xpack.core.ml.annotations.AnnotationIndex;
2425
import org.elasticsearch.xpack.core.ml.job.config.AnalysisConfig;
2526
import org.elasticsearch.xpack.core.ml.job.config.DataDescription;
2627
import org.elasticsearch.xpack.core.ml.job.config.Detector;
2728
import org.elasticsearch.xpack.core.ml.job.config.Job;
29+
import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.DataCounts;
2830
import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.ModelSizeStats;
2931
import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.ModelSnapshot;
3032
import org.elasticsearch.xpack.core.ml.job.process.autodetect.state.Quantiles;
33+
import org.elasticsearch.xpack.core.ml.job.results.AnomalyRecord;
3134
import org.elasticsearch.xpack.core.ml.job.results.Bucket;
3235
import org.elasticsearch.xpack.core.security.user.XPackUser;
3336
import org.junit.After;
@@ -49,6 +52,8 @@
4952
import static org.hamcrest.Matchers.greaterThan;
5053
import static org.hamcrest.Matchers.is;
5154
import static org.hamcrest.Matchers.not;
55+
import static org.hamcrest.Matchers.notNullValue;
56+
import static org.hamcrest.Matchers.nullValue;
5257

5358
/**
5459
* This test pushes data through a job in 2 runs creating
@@ -58,19 +63,60 @@
5863
public class RevertModelSnapshotIT extends MlNativeAutodetectIntegTestCase {
5964

6065
@After
61-
public void tearDownData() throws Exception {
66+
public void tearDownData() {
6267
cleanUp();
6368
}
6469

6570
public void testRevertModelSnapshot() throws Exception {
66-
test("revert-model-snapshot-it-job", false);
71+
testRunJobInTwoPartsAndRevertSnapshotAndRunToCompletion("revert-model-snapshot-it-job", false);
6772
}
6873

6974
public void testRevertModelSnapshot_DeleteInterveningResults() throws Exception {
70-
test("revert-model-snapshot-it-job-delete-intervening-results", true);
75+
testRunJobInTwoPartsAndRevertSnapshotAndRunToCompletion("revert-model-snapshot-it-job-delete-intervening-results", true);
7176
}
7277

73-
private void test(String jobId, boolean deleteInterveningResults) throws Exception {
78+
public void testRevertToEmptySnapshot() throws Exception {
79+
String jobId = "revert-to-empty-snapshot-test";
80+
81+
TimeValue bucketSpan = TimeValue.timeValueHours(1);
82+
long startTime = 1491004800000L;
83+
84+
String data = generateData(startTime, bucketSpan, 20, Arrays.asList("foo"),
85+
(bucketIndex, series) -> bucketIndex == 19 ? 100.0 : 10.0).stream().collect(Collectors.joining());
86+
87+
Job.Builder job = buildAndRegisterJob(jobId, bucketSpan);
88+
openJob(job.getId());
89+
postData(job.getId(), data);
90+
flushJob(job.getId(), true);
91+
closeJob(job.getId());
92+
93+
assertThat(getJob(jobId).get(0).getModelSnapshotId(), is(notNullValue()));
94+
List<Bucket> expectedBuckets = getBuckets(jobId);
95+
assertThat(expectedBuckets.size(), equalTo(20));
96+
List<AnomalyRecord> expectedRecords = getRecords(jobId);
97+
assertThat(expectedBuckets.isEmpty(), is(false));
98+
assertThat(expectedRecords.isEmpty(), is(false));
99+
100+
RevertModelSnapshotAction.Response revertResponse = revertModelSnapshot(jobId, "empty", true);
101+
assertThat(revertResponse.getModel().getSnapshotId(), equalTo("empty"));
102+
103+
assertThat(getJob(jobId).get(0).getModelSnapshotId(), is(nullValue()));
104+
assertThat(getBuckets(jobId).isEmpty(), is(true));
105+
assertThat(getRecords(jobId).isEmpty(), is(true));
106+
assertThat(getJobStats(jobId).get(0).getDataCounts().getLatestRecordTimeStamp(), is(nullValue()));
107+
108+
// Now run again and see we get same results
109+
openJob(job.getId());
110+
DataCounts dataCounts = postData(job.getId(), data);
111+
assertThat(dataCounts.getOutOfOrderTimeStampCount(), equalTo(0L));
112+
flushJob(job.getId(), true);
113+
closeJob(job.getId());
114+
115+
assertThat(getBuckets(jobId).size(), equalTo(expectedBuckets.size()));
116+
assertThat(getRecords(jobId), equalTo(expectedRecords));
117+
}
118+
119+
private void testRunJobInTwoPartsAndRevertSnapshotAndRunToCompletion(String jobId, boolean deleteInterveningResults) throws Exception {
74120
TimeValue bucketSpan = TimeValue.timeValueHours(1);
75121
long startTime = 1491004800000L;
76122

‎x-pack/plugin/ml/src/internalClusterTest/java/org/elasticsearch/xpack/ml/integration/AutodetectResultProcessorIT.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -758,7 +758,7 @@ private Optional<Quantiles> getQuantiles() throws Exception {
758758
AtomicReference<Exception> errorHolder = new AtomicReference<>();
759759
AtomicReference<Optional<Quantiles>> resultHolder = new AtomicReference<>();
760760
CountDownLatch latch = new CountDownLatch(1);
761-
jobResultsProvider.getAutodetectParams(JobTests.buildJobBuilder(JOB_ID).build(), params -> {
761+
jobResultsProvider.getAutodetectParams(JobTests.buildJobBuilder(JOB_ID).setModelSnapshotId("test_snapshot").build(), params -> {
762762
resultHolder.set(Optional.ofNullable(params.quantiles()));
763763
latch.countDown();
764764
}, e -> {

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

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -127,22 +127,31 @@ private void getModelSnapshot(RevertModelSnapshotAction.Request request, JobResu
127127
Consumer<Exception> errorHandler) {
128128
logger.info("Reverting to snapshot '" + request.getSnapshotId() + "'");
129129

130+
if (ModelSnapshot.isTheEmptySnapshot(request.getSnapshotId())) {
131+
handler.accept(ModelSnapshot.emptySnapshot(request.getJobId()));
132+
return;
133+
}
134+
130135
provider.getModelSnapshot(request.getJobId(), request.getSnapshotId(), modelSnapshot -> {
131136
if (modelSnapshot == null) {
132-
throw new ResourceNotFoundException(Messages.getMessage(Messages.REST_NO_SUCH_MODEL_SNAPSHOT, request.getSnapshotId(),
133-
request.getJobId()));
137+
throw missingSnapshotException(request);
134138
}
135139
handler.accept(modelSnapshot.result);
136140
}, errorHandler);
137141
}
138142

143+
private static ResourceNotFoundException missingSnapshotException(RevertModelSnapshotAction.Request request) {
144+
return new ResourceNotFoundException(Messages.getMessage(Messages.REST_NO_SUCH_MODEL_SNAPSHOT, request.getSnapshotId(),
145+
request.getJobId()));
146+
}
147+
139148
private ActionListener<RevertModelSnapshotAction.Response> wrapDeleteOldAnnotationsListener(
140149
ActionListener<RevertModelSnapshotAction.Response> listener,
141150
ModelSnapshot modelSnapshot,
142151
String jobId) {
143152

144153
return ActionListener.wrap(response -> {
145-
Date deleteAfter = modelSnapshot.getLatestResultTimeStamp();
154+
Date deleteAfter = modelSnapshot.getLatestResultTimeStamp() == null ? new Date(0) : modelSnapshot.getLatestResultTimeStamp();
146155
logger.info("[{}] Removing intervening annotations after reverting model: deleting annotations after [{}]", jobId, deleteAfter);
147156

148157
JobDataDeleter dataDeleter = new JobDataDeleter(client, jobId);
@@ -176,7 +185,7 @@ private ActionListener<RevertModelSnapshotAction.Response> wrapDeleteOldDataList
176185
// wrap the listener with one that invokes the OldDataRemover on
177186
// acknowledged responses
178187
return ActionListener.wrap(response -> {
179-
Date deleteAfter = modelSnapshot.getLatestResultTimeStamp();
188+
Date deleteAfter = modelSnapshot.getLatestResultTimeStamp() == null ? new Date(0) : modelSnapshot.getLatestResultTimeStamp();
180189
logger.info("[{}] Removing intervening records after reverting model: deleting results after [{}]", jobId, deleteAfter);
181190

182191
JobDataDeleter dataDeleter = new JobDataDeleter(client, jobId);

‎x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/JobManager.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -428,7 +428,7 @@ private void validate(Job job, JobUpdate jobUpdate, ActionListener<Void> handler
428428
}
429429

430430
private void validateModelSnapshotIdUpdate(Job job, String modelSnapshotId, VoidChainTaskExecutor voidChainTaskExecutor) {
431-
if (modelSnapshotId != null) {
431+
if (modelSnapshotId != null && ModelSnapshot.isTheEmptySnapshot(modelSnapshotId) == false) {
432432
voidChainTaskExecutor.add(listener -> {
433433
jobResultsProvider.getModelSnapshot(job.getId(), modelSnapshotId, newModelSnapshot -> {
434434
if (newModelSnapshot == null) {
@@ -599,6 +599,11 @@ public void revertSnapshot(RevertModelSnapshotAction.Request request, ActionList
599599
// Step 3. After the model size stats is persisted, also persist the snapshot's quantiles and respond
600600
// -------
601601
CheckedConsumer<IndexResponse, Exception> modelSizeStatsResponseHandler = response -> {
602+
// In case we are reverting to the empty snapshot the quantiles will be null
603+
if (modelSnapshot.getQuantiles() == null) {
604+
actionListener.onResponse(new RevertModelSnapshotAction.Response(modelSnapshot));
605+
return;
606+
}
602607
jobResultsPersister.persistQuantiles(modelSnapshot.getQuantiles(), WriteRequest.RefreshPolicy.IMMEDIATE,
603608
ActionListener.wrap(quantilesResponse -> {
604609
// The quantiles can be large, and totally dominate the output -

‎x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobResultsProvider.java‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -585,11 +585,12 @@ public void getAutodetectParams(Job job, String snapshotId, Consumer<AutodetectP
585585
MultiSearchRequestBuilder msearch = client.prepareMultiSearch()
586586
.add(createLatestDataCountsSearch(resultsIndex, jobId))
587587
.add(createLatestModelSizeStatsSearch(resultsIndex))
588-
.add(createLatestTimingStatsSearch(resultsIndex, jobId))
589-
// These next two document IDs never need to be the legacy ones due to the rule
590-
// that you cannot open a 5.4 job in a subsequent version of the product
591-
.add(createDocIdSearch(resultsIndex, ModelSnapshot.documentId(jobId, snapshotId)))
592-
.add(createDocIdSearch(stateIndex, Quantiles.documentId(jobId)));
588+
.add(createLatestTimingStatsSearch(resultsIndex, jobId));
589+
590+
if (snapshotId != null) {
591+
msearch.add(createDocIdSearch(resultsIndex, ModelSnapshot.documentId(jobId, snapshotId)));
592+
msearch.add(createDocIdSearch(stateIndex, Quantiles.documentId(jobId)));
593+
}
593594

594595
for (String filterId : job.getAnalysisConfig().extractReferencedFilters()) {
595596
msearch.add(createDocIdSearch(MlMetaIndex.indexName(), MlFilter.documentId(filterId)));

0 commit comments

Comments
 (0)