[ML] Add missing job_id filter to Anomaly Detection data deleter (elastic#138160)

When reverting to a previous model snapshot intervening results are removed.
A bug in the delete-by-query query meant that results were deleted for all jobs
in this period not just the job being deleted. The fix here reinstates the job_id 
filter.

Author: davidkyle

docs/changelog/138160.yaml

@@ -0,0 +1,5 @@
+pr: 138160
+summary: Add missing `job_id` filter to Anomaly Detection data deleter
+area: Machine Learning
+type: bug
+issues: []

muted-tests.yml

@@ -267,12 +267,6 @@ tests:
 - class: org.elasticsearch.xpack.sql.qa.mixed_node.SqlCompatIT
   method: testNullsOrderWithMissingOrderSupportQueryingNewNode
   issue: https://github.com/elastic/elasticsearch/issues/132249
-- class: org.elasticsearch.xpack.ml.integration.RevertModelSnapshotIT
-  method: testRevertModelSnapshot_DeleteInterveningResults
-  issue: https://github.com/elastic/elasticsearch/issues/132349
-- class: org.elasticsearch.xpack.ml.integration.RevertModelSnapshotIT
-  method: testRevertModelSnapshot
-  issue: https://github.com/elastic/elasticsearch/issues/132733
 - class: org.elasticsearch.packaging.test.ArchiveGenerateInitialCredentialsTests
   method: test40VerifyAutogeneratedCredentials
   issue: https://github.com/elastic/elasticsearch/issues/132877

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

@@ -303,6 +303,28 @@ protected void assertThatNumberOfAnnotationsIsEqualTo(int expectedNumberOfAnnota
         });
     }
 
+    protected void assertThatNumberOfAnnotationsIsEqualTo(String jobId, int expectedNumberOfAnnotations) throws Exception {
+        // Refresh the annotations index so that recently indexed annotation docs are visible.
+        indicesAdmin().prepareRefresh(AnnotationIndex.LATEST_INDEX_NAME)
+            .setIndicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN_CLOSED_HIDDEN)
+            .get();
+
+        SearchRequest searchRequest = client().prepareSearch(AnnotationIndex.READ_ALIAS_NAME)
+            .setQuery(QueryBuilders.termQuery("job_id", jobId))
+            .setIndicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN_CLOSED_HIDDEN)
+            .request();
+
+        assertCheckedResponse(client().search(searchRequest), searchResponse -> {
+            List<Annotation> annotations = new ArrayList<>();
+            for (SearchHit hit : searchResponse.getHits().getHits()) {
+                try (XContentParser parser = createParser(jsonXContent, hit.getSourceRef())) {
+                    annotations.add(Annotation.fromXContent(parser, null));
+                }
+            }
+            assertThat("Annotations were: " + annotations, annotations, hasSize(expectedNumberOfAnnotations));
+        });
+    }
+
     protected List<Annotation> getAnnotations() throws Exception {
         List<Annotation> annotations = new ArrayList<>();
         // Refresh the annotations index so that recently indexed annotation docs are visible.

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

@@ -65,6 +65,8 @@
  */
 public class RevertModelSnapshotIT extends MlNativeAutodetectIntegTestCase {
 
+    private static final long DATA_START_TIME = 1761955200000L;
+
     @After
     public void tearDownData() {
         cleanUp();
@@ -75,7 +77,32 @@ public void testRevertModelSnapshot() throws Exception {
     }
 
     public void testRevertModelSnapshot_DeleteInterveningResults() throws Exception {
+        // Create and run a unrelated job to chech it is not affected by reverting a different job
+        String jobId = "revert-snapshot-delete-intervening-unrelated-job";
+
+        TimeValue bucketSpan = TimeValue.timeValueHours(1);
+        long startTime = DATA_START_TIME - (bucketSpan.getMillis() * 2);
+        String data = String.join("", generateData(startTime, bucketSpan, 23, List.of("foo"), (bucketIndex, series) -> 10.0));
+
+        Job.Builder job = buildAndRegisterJob(jobId, bucketSpan);
+        openJob(job.getId());
+        postData(job.getId(), data);
+        flushJob(job.getId(), true);
+        closeJob(job.getId());
+
+        String snapShotId = getJob(jobId).get(0).getModelSnapshotId();
+        assertThat(snapShotId, is(notNullValue()));
+        List<Bucket> buckets = getBuckets(jobId);
+        assertThat(buckets.size(), greaterThan(0));
+
+        // Run another job and revert to an previous snapshot
         testRunJobInTwoPartsAndRevertSnapshotAndRunToCompletion("revert-model-snapshot-it-job-delete-intervening-results", true);
+
+        // Check snapshot Id and buckets have not changed
+        assertThat(getJob(jobId).getFirst().getModelSnapshotId(), is(snapShotId));
+        List<Bucket> bucketsAfterRevert = getBuckets(jobId);
+        assertThat(bucketsAfterRevert.size(), is(buckets.size()));
+        assertThat(bucketsAfterRevert, is(buckets));
     }
 
     public void testRevertToEmptySnapshot() throws Exception {
@@ -126,13 +153,13 @@ public void testRevertToEmptySnapshot() throws Exception {
 
     private void testRunJobInTwoPartsAndRevertSnapshotAndRunToCompletion(String jobId, boolean deleteInterveningResults) throws Exception {
         TimeValue bucketSpan = TimeValue.timeValueHours(1);
-        long startTime = 1491004800000L;
 
         Job.Builder job = buildAndRegisterJob(jobId, bucketSpan);
         openJob(job.getId());
         postData(
             job.getId(),
-            generateData(startTime, bucketSpan, 10, Arrays.asList("foo"), (bucketIndex, series) -> bucketIndex == 5 ? 100.0 : 10.0).stream()
+            generateData(DATA_START_TIME, bucketSpan, 10, Arrays.asList("foo"), (bucketIndex, series) -> bucketIndex == 5 ? 100.0 : 10.0)
+                .stream()
                 .collect(Collectors.joining())
         );
         flushJob(job.getId(), true);
@@ -156,7 +183,7 @@ private void testRunJobInTwoPartsAndRevertSnapshotAndRunToCompletion(String jobI
         postData(
             job.getId(),
             generateData(
-                startTime + 10 * bucketSpan.getMillis(),
+                DATA_START_TIME + 10 * bucketSpan.getMillis(),
                 bucketSpan,
                 10,
                 Arrays.asList("foo", "bar"),
@@ -187,15 +214,15 @@ private void testRunJobInTwoPartsAndRevertSnapshotAndRunToCompletion(String jobI
         ModelSnapshot revertSnapshot = modelSnapshots.get(1);
 
         // Check there are 2 annotations (one per model snapshot)
-        assertThatNumberOfAnnotationsIsEqualTo(2);
+        assertThatNumberOfAnnotationsIsEqualTo(jobId, 2);
 
         // Add 3 new annotations...
         Instant lastResultTimestamp = revertSnapshot.getLatestResultTimeStamp().toInstant();
         client().index(randomAnnotationIndexRequest(job.getId(), lastResultTimestamp.plusSeconds(10), Event.DELAYED_DATA)).actionGet();
         client().index(randomAnnotationIndexRequest(job.getId(), lastResultTimestamp.plusSeconds(20), Event.MODEL_CHANGE)).actionGet();
         client().index(randomAnnotationIndexRequest(job.getId(), lastResultTimestamp.minusSeconds(10), Event.MODEL_CHANGE)).actionGet();
         // ... and check there are 5 annotations in total now
-        assertThatNumberOfAnnotationsIsEqualTo(5);
+        assertThatNumberOfAnnotationsIsEqualTo(jobId, 5);
 
         GetJobsStatsAction.Response.JobStats statsBeforeRevert = getJobStats(jobId).get(0);
         Instant timeBeforeRevert = Instant.now();
@@ -219,7 +246,7 @@ private void testRunJobInTwoPartsAndRevertSnapshotAndRunToCompletion(String jobI
         assertThat(getQuantiles(job.getId()).getTimestamp(), equalTo(revertSnapshot.getLatestResultTimeStamp()));
 
         // Check annotations with event type from {delayed_data, model_change} have been removed if deleteInterveningResults flag is set
-        assertThatNumberOfAnnotationsIsEqualTo(deleteInterveningResults ? 3 : 5);
+        assertThatNumberOfAnnotationsIsEqualTo(jobId, deleteInterveningResults ? 3 : 5);
 
         // Reverting should not have deleted any forecast docs
         assertThat(countForecastDocs(job.getId(), forecastId), is(numForecastDocs));
@@ -229,7 +256,7 @@ private void testRunJobInTwoPartsAndRevertSnapshotAndRunToCompletion(String jobI
         postData(
             job.getId(),
             generateData(
-                startTime + 10 * bucketSpan.getMillis(),
+                DATA_START_TIME + 10 * bucketSpan.getMillis(),
                 bucketSpan,
                 10,
                 Arrays.asList("foo", "bar"),

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

@@ -251,6 +251,7 @@ private <T> String[] removeReadOnlyIndices(
      */
     public void deleteResultsFromTime(long cutoffEpochMs, ActionListener<Boolean> listener) {
         QueryBuilder query = QueryBuilders.boolQuery()
+            .filter(QueryBuilders.termQuery(Job.ID.getPreferredName(), jobId))
             .filter(
                 QueryBuilders.termsQuery(
                     Result.RESULT_TYPE.getPreferredName(),
@@ -262,6 +263,7 @@ public void deleteResultsFromTime(long cutoffEpochMs, ActionListener<Boolean> li
                 )
             )
             .filter(QueryBuilders.rangeQuery(Result.TIMESTAMP.getPreferredName()).gte(cutoffEpochMs));
+
         String[] indicesToQuery = removeReadOnlyIndices(
             List.of(AnomalyDetectorsIndex.jobResultsAliasedName(jobId)),
             listener,

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/job/persistence/JobDataDeleterTests.java

@@ -58,7 +58,6 @@ public void setUpTests() {
 
     @After
     public void verifyNoMoreInteractionsWithClient() {
-        verify(client, times(2)).threadPool();
         verifyNoMoreInteractions(client);
     }
 
@@ -85,6 +84,7 @@ public void testDeleteAllAnnotations() {
                 assertThat(dbqQueryString, containsString("_xpack"));
             }
         });
+        verify(client, times(2)).threadPool();
     }
 
     public void testDeleteAnnotations_TimestampFiltering() {
@@ -115,6 +115,7 @@ public void testDeleteAnnotations_TimestampFiltering() {
                 assertThat(dbqQueryString, containsString("_xpack"));
             }
         });
+        verify(client, times(2)).threadPool();
     }
 
     public void testDeleteAnnotations_EventFiltering() {
@@ -145,6 +146,22 @@ public void testDeleteAnnotations_EventFiltering() {
                 assertThat(dbqQueryString, containsString("_xpack"));
             }
         });
+        verify(client, times(2)).threadPool();
+    }
+
+    public void testDeleteResultsFromTime() {
+        MockWritableIndexExpander.create(true);
+        long fromEpochMs = randomNonNegativeLong();
+        JobDataDeleter jobDataDeleter = new JobDataDeleter(client, JOB_ID, randomBoolean());
+        jobDataDeleter.deleteResultsFromTime(fromEpochMs, ActionTestUtils.assertNoFailureListener(deleteResponse -> {}));
+
+        verify(client).execute(eq(DeleteByQueryAction.INSTANCE), deleteRequestCaptor.capture(), any());
+
+        DeleteByQueryRequest deleteRequest = deleteRequestCaptor.getValue();
+        assertThat(deleteRequest.indices(), is(arrayContaining(".ml-anomalies-my-job-id")));
+        String dbqQueryString = Strings.toString(deleteRequest.getSearchRequest().source().query());
+        assertThat(dbqQueryString, containsString("{\"term\":{\"job_id\":{\"value\":\"my-job-id\"}}"));
+        verify(client, times(1)).threadPool();
     }
 
     public void testDeleteDatafeedTimingStats() {
@@ -162,6 +179,7 @@ public void testDeleteDatafeedTimingStats() {
             DeleteByQueryRequest deleteRequest = deleteRequestCaptor.getValue();
             assertThat(deleteRequest.indices(), is(arrayContaining(AnomalyDetectorsIndex.jobResultsAliasedName(JOB_ID))));
         });
+        verify(client, times(2)).threadPool();
     }
 
     public void testDeleteDatafeedTimingStats_WhenIndexReadOnly_ShouldNotDeleteAnything() {
@@ -178,5 +196,6 @@ public void testDeleteDatafeedTimingStats_WhenIndexReadOnly_ShouldNotDeleteAnyth
                 client.threadPool();
             }
         });
+        verify(client, times(2)).threadPool();
     }
 }