[ML] Check if the anomaly results index has been rolled over

docs/changelog/125404.yaml

@@ -0,0 +1,5 @@
+pr: 125404
+summary: Check if the anomaly results index has been rolled over
+area: Machine Learning
+type: upgrade
+issues: []

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/utils/MlIndexAndAlias.java

@@ -407,7 +407,9 @@ public static boolean hasIndexTemplate(ClusterState state, String templateName,
     }
 
     public static boolean has6DigitSuffix(String indexName) {
-        return HAS_SIX_DIGIT_SUFFIX.test(indexName);
+        String[] indexParts = indexName.split("-");
+        String suffix = indexParts[indexParts.length - 1];
+        return HAS_SIX_DIGIT_SUFFIX.test(suffix);
     }
 
     /**

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/utils/MlIndexAndAliasTests.java

@@ -382,6 +382,13 @@ public void testIndexIsReadWriteCompatibleInV9() {
         assertFalse(MlIndexAndAlias.indexIsReadWriteCompatibleInV9(IndexVersions.V_7_17_0));
     }
 
+    public void testHas6DigitSuffix() {
+        assertTrue(MlIndexAndAlias.has6DigitSuffix("index-000001"));
+        assertFalse(MlIndexAndAlias.has6DigitSuffix("index1"));
+        assertFalse(MlIndexAndAlias.has6DigitSuffix("index-foo"));
+        assertFalse(MlIndexAndAlias.has6DigitSuffix("index000001"));
+    }
+
     private void createIndexAndAliasIfNecessary(ClusterState clusterState) {
         MlIndexAndAlias.createIndexAndAliasIfNecessary(
             client,

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MlAnomaliesIndexUpdate.java

@@ -9,6 +9,7 @@
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.ResourceAlreadyExistsException;
 import org.elasticsearch.TransportVersion;
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionListener;
@@ -37,9 +38,12 @@
 import org.elasticsearch.xpack.core.ml.utils.MlStrings;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN;
+import static org.elasticsearch.xpack.core.ml.utils.MlIndexAndAlias.FIRST_INDEX_SIX_DIGIT_SUFFIX;
+import static org.elasticsearch.xpack.core.ml.utils.MlIndexAndAlias.has6DigitSuffix;
 
 /**
  * Rollover the various .ml-anomalies result indices
@@ -108,6 +112,14 @@ public void runUpdate(ClusterState latestState) {
                 continue;
             }
 
+            // Check if this index has already been rolled over
+            String latestIndex = latestIndexMatchingBaseName(index, expressionResolver, latestState);
+
+            if (index.equals(latestIndex) == false) {
+                logger.debug("index [{}] will not be rolled over as there is a later index [{}]", index, latestIndex);
+                continue;
+            }
+
             PlainActionFuture<Boolean> updated = new PlainActionFuture<>();
             rollAndUpdateAliases(latestState, index, updated);
             try {
@@ -137,7 +149,7 @@ public void runUpdate(ClusterState latestState) {
 
     private void rollAndUpdateAliases(ClusterState clusterState, String index, ActionListener<Boolean> listener) {
         // Create an alias specifically for rolling over.
-        // The ml-anomalies index has aliases for each job anyone
+        // The ml-anomalies index has aliases for each job, any
         // of which could be used but that means one alias is
         // treated differently.
         // Using a `.` in the alias name avoids any conflicts
@@ -163,9 +175,18 @@ private void rollAndUpdateAliases(ClusterState clusterState, String index, Actio
     }
 
     private void rollover(String alias, @Nullable String newIndexName, ActionListener<String> listener) {
-        client.admin().indices().rolloverIndex(new RolloverRequest(alias, newIndexName), listener.delegateFailure((l, response) -> {
-            l.onResponse(response.getNewIndex());
-        }));
+        client.admin()
+            .indices()
+            .rolloverIndex(
+                new RolloverRequest(alias, newIndexName),
+                ActionListener.wrap(response -> listener.onResponse(response.getNewIndex()), e -> {
+                    if (e instanceof ResourceAlreadyExistsException alreadyExistsException) {
+                        listener.onResponse(alreadyExistsException.getIndex().getName());
+                    } else {
+                        listener.onFailure(e);
+                    }
+                })
+            );
     }
 
     private void createAliasForRollover(String indexName, String aliasName, ActionListener<IndicesAliasesResponse> listener) {
@@ -232,4 +253,40 @@ static boolean isAnomaliesReadAlias(String aliasName) {
         // which is not a valid job id.
         return MlStrings.isValidId(jobIdPart);
     }
+
+    /**
+     * Strip any suffix from the index name and find any other indices
+     * that match the base name. Then return the latest index from the
+     * matching ones.
+     *
+     * @param index The index to check
+     * @param expressionResolver The expression resolver
+     * @param latestState The latest cluster state
+     * @return The latest index that matches the base name of the given index
+     */
+    static String latestIndexMatchingBaseName(String index, IndexNameExpressionResolver expressionResolver, ClusterState latestState) {
+        String baseIndexName = MlIndexAndAlias.has6DigitSuffix(index)
+            ? index.substring(0, index.length() - FIRST_INDEX_SIX_DIGIT_SUFFIX.length())
+            : index;
+
+        String[] matching = expressionResolver.concreteIndexNames(
+            latestState,
+            IndicesOptions.lenientExpandOpenHidden(),
+            baseIndexName + "*"
+        );
+
+        // This should never happen
+        if (matching.length == 0) {
+            return index;
+        }
+
+        // Exclude indices that start with the same base name but are a different index
+        // e.g. .ml-anomalies-foobar should not be included when the index name is
+        // .ml-anomalies-foo
+        String[] filtered = Arrays.stream(matching).filter(i -> {
+            return i.equals(index) || (has6DigitSuffix(i) && i.length() == baseIndexName.length() + FIRST_INDEX_SIX_DIGIT_SUFFIX.length());
+        }).toArray(String[]::new);
+
+        return MlIndexAndAlias.latestIndex(filtered);
+    }
 }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MlAutoUpdateService.java

@@ -66,7 +66,6 @@ public void clusterChanged(ClusterChangedEvent event) {
             .filter(action -> action.isAbleToRun(latestState))
             .filter(action -> currentlyUpdating.add(action.getName()))
             .toList();
-        // TODO run updates serially
         threadPool.executor(MachineLearning.UTILITY_THREAD_POOL_NAME)
             .execute(() -> toRun.forEach((action) -> this.runUpdate(action, latestState)));
     }

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MlAnomaliesIndexUpdateTests.java

@@ -32,6 +32,7 @@
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex;
+import org.elasticsearch.xpack.core.ml.utils.MlIndexAndAlias;
 
 import java.util.List;
 import java.util.Map;
@@ -179,6 +180,78 @@ public void testRunUpdate_LegacyIndex() {
         verifyNoMoreInteractions(client);
     }
 
+    public void testLatestIndexMatchingBaseName_isLatest() {
+        Metadata.Builder metadata = Metadata.builder();
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-foo", IndexVersion.current(), List.of("job1")));
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-bar", IndexVersion.current(), List.of("job2")));
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-bax", IndexVersion.current(), List.of("job3")));
+        ClusterState.Builder csBuilder = ClusterState.builder(new ClusterName("_name"));
+        csBuilder.metadata(metadata);
+
+        var latest = MlAnomaliesIndexUpdate.latestIndexMatchingBaseName(
+            ".ml-anomalies-custom-foo",
+            TestIndexNameExpressionResolver.newInstance(),
+            csBuilder.build()
+        );
+        assertEquals(".ml-anomalies-custom-foo", latest);
+    }
+
+    public void testLatestIndexMatchingBaseName_hasLater() {
+        Metadata.Builder metadata = Metadata.builder();
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-foo", IndexVersion.current(), List.of("job1")));
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-bar", IndexVersion.current(), List.of("job2")));
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-foo-000001", IndexVersion.current(), List.of("job3")));
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-foo-000002", IndexVersion.current(), List.of("job4")));
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-baz-000001", IndexVersion.current(), List.of("job5")));
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-baz-000002", IndexVersion.current(), List.of("job6")));
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-baz-000003", IndexVersion.current(), List.of("job7")));
+        ClusterState.Builder csBuilder = ClusterState.builder(new ClusterName("_name"));
+        csBuilder.metadata(metadata);
+        var state = csBuilder.build();
+
+        assertTrue(MlIndexAndAlias.has6DigitSuffix(".ml-anomalies-custom-foo-000002"));
+
+        var latest = MlAnomaliesIndexUpdate.latestIndexMatchingBaseName(
+            ".ml-anomalies-custom-foo",
+            TestIndexNameExpressionResolver.newInstance(),
+            state
+        );
+        assertEquals(".ml-anomalies-custom-foo-000002", latest);
+
+        latest = MlAnomaliesIndexUpdate.latestIndexMatchingBaseName(
+            ".ml-anomalies-custom-baz-000001",
+            TestIndexNameExpressionResolver.newInstance(),
+            state
+        );
+        assertEquals(".ml-anomalies-custom-baz-000003", latest);
+    }
+
+    public void testLatestIndexMatchingBaseName_CollidingIndexNames() {
+        Metadata.Builder metadata = Metadata.builder();
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-foo", IndexVersion.current(), List.of("job1")));
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-bar", IndexVersion.current(), List.of("job2")));
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-foodifferent000001", IndexVersion.current(), List.of("job3")));
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-foo-notthisone-000001", IndexVersion.current(), List.of("job4")));
+        metadata.put(createSharedResultsIndex(".ml-anomalies-custom-foo-notthisone-000002", IndexVersion.current(), List.of("job5")));
+        ClusterState.Builder csBuilder = ClusterState.builder(new ClusterName("_name"));
+        csBuilder.metadata(metadata);
+        var state = csBuilder.build();
+
+        var latest = MlAnomaliesIndexUpdate.latestIndexMatchingBaseName(
+            ".ml-anomalies-custom-foo",
+            TestIndexNameExpressionResolver.newInstance(),
+            state
+        );
+        assertEquals(".ml-anomalies-custom-foo", latest);
+
+        latest = MlAnomaliesIndexUpdate.latestIndexMatchingBaseName(
+            ".ml-anomalies-custom-foo-notthisone-000001",
+            TestIndexNameExpressionResolver.newInstance(),
+            state
+        );
+        assertEquals(".ml-anomalies-custom-foo-notthisone-000002", latest);
+    }
+
     private record AliasActionMatcher(String aliasName, String index, IndicesAliasesRequest.AliasActions.Type actionType) {
         boolean matches(IndicesAliasesRequest.AliasActions aliasAction) {
             return aliasAction.actionType() == actionType