[ML] Add daily task to manage .ml-state indices

docs/changelog/137653.yaml

@@ -0,0 +1,5 @@
+pr: 137653
+summary: Add daily task to manage .ml-state indices
+area: Machine Learning
+type: enhancement
+issues: []

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

@@ -19,6 +19,7 @@
 
 import java.util.Locale;
 import java.util.Map;
+import java.util.Optional;
 
 import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN;
 import static org.elasticsearch.xpack.core.ClientHelper.executeAsyncWithOrigin;
@@ -30,6 +31,7 @@ public final class AnomalyDetectorsIndex {
 
     private static final String RESULTS_MAPPINGS_VERSION_VARIABLE = "xpack.ml.version";
     private static final String RESOURCE_PATH = "/ml/anomalydetection/";
+    private static final String WRITE_ALIAS_PREFIX = ".write-";
     public static final int RESULTS_INDEX_MAPPINGS_VERSION = 1;
 
     private AnomalyDetectorsIndex() {}
@@ -57,11 +59,16 @@ public static String jobResultsAliasedName(String jobId) {
      * @param jobResultsAliasedName The alias
      * @return The job Id
      */
-    public static String jobIdFromAlias(String jobResultsAliasedName) {
+    public static Optional<String> jobIdFromAlias(String jobResultsAliasedName) {
         if (jobResultsAliasedName.length() < AnomalyDetectorsIndexFields.RESULTS_INDEX_PREFIX.length()) {
-            return null;
+            return Optional.empty();
         }
-        return jobResultsAliasedName.substring(AnomalyDetectorsIndexFields.RESULTS_INDEX_PREFIX.length());
+
+        var jobId = jobResultsAliasedName.substring(AnomalyDetectorsIndexFields.RESULTS_INDEX_PREFIX.length());
+        if (jobId.startsWith(WRITE_ALIAS_PREFIX)) {
+            jobId = jobId.substring(WRITE_ALIAS_PREFIX.length());
+        }
+        return Optional.of(jobId);
     }
 
     /**

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

@@ -509,10 +509,15 @@ public void testLatestIndexMatchingBaseName_CollidingIndexNames() {
 
     public void testBuildIndexAliasesRequest() {
         var anomaliesIndex = ".ml-anomalies-sharedindex";
+        var newIndex = anomaliesIndex + "-000001";
+
         var jobs = List.of("job1", "job2");
-        IndexMetadata.Builder indexMetadata = createSharedResultsIndex(anomaliesIndex, IndexVersion.current(), jobs);
+        IndexMetadata.Builder oldIndexMetadata = createSharedResultsIndex(anomaliesIndex, IndexVersion.current(), jobs);
+        IndexMetadata.Builder newIndexMetadata = createEmptySharedResultsIndex(newIndex, IndexVersion.current());
+
         Metadata.Builder metadata = Metadata.builder();
-        metadata.put(indexMetadata);
+        metadata.put(oldIndexMetadata);
+        metadata.put(newIndexMetadata);
         ClusterState.Builder csBuilder = ClusterState.builder(new ClusterName("_name"));
         csBuilder.metadata(metadata);
 
@@ -522,8 +527,13 @@ public void testBuildIndexAliasesRequest() {
             TEST_REQUEST_TIMEOUT
         );
 
-        var newIndex = anomaliesIndex + "-000001";
-        var request = MlIndexAndAlias.addIndexAliasesRequests(aliasRequestBuilder, anomaliesIndex, newIndex, csBuilder.build());
+        String[] currentIndices = { anomaliesIndex };
+        var request = MlIndexAndAlias.addResultsIndexRolloverAliasActions(
+            aliasRequestBuilder,
+            newIndex,
+            csBuilder.build(),
+            Arrays.asList(currentIndices)
+        );
         var actions = request.request().getAliasActions();
         assertThat(actions, hasSize(6));
 
@@ -535,6 +545,7 @@ public void testBuildIndexAliasesRequest() {
                 newIndex,
                 IndicesAliasesRequest.AliasActions.Type.ADD
             );
+
             assertThat(actions.stream().filter(expected::matches).count(), equalTo(1L));
 
             expected = new AliasActionMatcher(
@@ -544,12 +555,13 @@ public void testBuildIndexAliasesRequest() {
             );
             assertThat(actions.stream().filter(expected::matches).count(), equalTo(1L));
 
-            expected = new AliasActionMatcher(
+            // This alias action request ensures that every index has a read alias, even if the old index was missing one.
+            var expected1 = new AliasActionMultiIndicesMatcher(
                 AnomalyDetectorsIndex.jobResultsAliasedName(job),
-                newIndex,
+                new String[] { anomaliesIndex, newIndex },
                 IndicesAliasesRequest.AliasActions.Type.ADD
             );
-            assertThat(actions.stream().filter(expected::matches).count(), equalTo(1L));
+            assertThat(actions.stream().filter(expected1::matches).count(), equalTo(1L));
         }
     }
 
@@ -561,7 +573,15 @@ boolean matches(IndicesAliasesRequest.AliasActions aliasAction) {
         }
     }
 
-    private IndexMetadata.Builder createSharedResultsIndex(String indexName, IndexVersion indexVersion, List<String> jobs) {
+    private record AliasActionMultiIndicesMatcher(String aliasName, String[] indices, IndicesAliasesRequest.AliasActions.Type actionType) {
+        boolean matches(IndicesAliasesRequest.AliasActions aliasAction) {
+            return aliasAction.actionType() == actionType
+                && aliasAction.aliases()[0].equals(aliasName)
+                && Arrays.stream(aliasAction.indices()).toList().equals(Arrays.stream(indices).toList());
+        }
+    }
+
+    private IndexMetadata.Builder createEmptySharedResultsIndex(String indexName, IndexVersion indexVersion) {
         IndexMetadata.Builder indexMetadata = IndexMetadata.builder(indexName);
         indexMetadata.settings(
             Settings.builder()
@@ -571,6 +591,13 @@ private IndexMetadata.Builder createSharedResultsIndex(String indexName, IndexVe
                 .put(IndexMetadata.SETTING_INDEX_UUID, "_uuid")
         );
 
+        return indexMetadata;
+    }
+
+    private IndexMetadata.Builder createSharedResultsIndex(String indexName, IndexVersion indexVersion, List<String> jobs) {
+
+        var indexMetadata = createEmptySharedResultsIndex(indexName, indexVersion);
+
         for (var jobId : jobs) {
             indexMetadata.putAlias(AliasMetadata.builder(AnomalyDetectorsIndex.jobResultsAliasedName(jobId)).isHidden(true).build());
             indexMetadata.putAlias(

x-pack/plugin/core/template-resources/src/main/resources/ml/anomalydetection/state_index_template.json

@@ -9,9 +9,7 @@
       "index" : {
         "auto_expand_replicas" : "0-1",
         "hidden": true
-      },
-      "index.lifecycle.name": "${xpack.ml.index.lifecycle.name}",
-      "index.lifecycle.rollover_alias": "${xpack.ml.index.lifecycle.rollover_alias}"
+      }
     },
     "mappings" : {
       "_meta": {

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

@@ -58,6 +58,7 @@ public void testTriggerDeleteJobsInStateDeletingWithoutDeletionTask() throws Int
             mock(IndexNameExpressionResolver.class),
             true,
             true,
+            true,
             true
         );