[ML] Manage AD results indices (elastic#136065)

Add a rollover check for the AD results indices to the nightly ML maintenance task.

The results indices will be rolled over when they are of a size equal or greater to 50GB (This value can be adjusted in the cluster config)

The concrete AD reults indices now have a six digit suffix. This is necessary to keep track of rollover behaviour and to determine which index is the "latest" in the series.

We choose not to use ILM to manage the rollover as that is not available for Serverless.

Author: edsavage

docs/changelog/136065.yaml

@@ -0,0 +1,5 @@
+pr: 136065
+summary: Nightly maintenance for anomaly detection results indices to keep to manageable size.
+area: Machine Learning
+type: enhancement
+issues: []

docs/reference/elasticsearch/configuration-reference/machine-learning-settings.md

@@ -86,6 +86,9 @@ $$$xpack.ml.max_open_jobs$$$
 `xpack.ml.nightly_maintenance_requests_per_second`
 :   ([Dynamic](docs-content://deploy-manage/stack-settings.md#dynamic-cluster-setting)) The rate at which the nightly maintenance task deletes expired model snapshots and results. The setting is a proxy to the [`requests_per_second`](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-delete-by-query) parameter used in the delete by query requests and controls throttling. When the {{operator-feature}} is enabled, this setting can be updated only by operator users. Valid values must be greater than `0.0` or equal to `-1.0`, where `-1.0` means a default value is used. Defaults to `-1.0`
 
+`xpack.ml.results_index_rollover_max_size`
+:   ([Dynamic](docs-content://deploy-manage/stack-settings.md#dynamic-cluster-setting)) The maximum size the anomaly detection results indices can reach before being rolled over by the nightly maintenance task. When the {{operator-feature}} is enabled, this setting can be updated only by operator users. Valid values must be greater than or equal to `-1B`. A value of `-1B` means the indices will never be rolled over. A value of `0B` means the indices will always be rolled over, regardless of size. Defaults to `50GB`.
+
 `xpack.ml.node_concurrent_job_allocations`
 :   ([Dynamic](docs-content://deploy-manage/stack-settings.md#dynamic-cluster-setting)) The maximum number of jobs that can concurrently be in the `opening` state on each node. Typically, jobs spend a small amount of time in this state before they move to `open` state. Jobs that must restore large models when they are opening spend more time in the `opening` state. When the {{operator-feature}} is enabled, this setting can be updated only by operator users. Defaults to `2`.
 

x-pack/plugin/build.gradle

@@ -85,6 +85,12 @@ tasks.named("precommit").configure {
 }
 
 tasks.named("yamlRestCompatTestTransform").configure({ task ->
+  task.replaceIsTrue("\\.ml-anomalies-shared.mappings._meta.version", "\\.ml-anomalies-shared-000001.mappings._meta.version")
+  task.replaceKeyInMatch("\\.ml-anomalies-shared.mappings.new_field.mapping.new_field.type", "\\.ml-anomalies-shared-000001.mappings.new_field.mapping.new_field.type")
+  task.replaceValueTextByKeyValue("index", ".ml-anomalies-shared", ".ml-anomalies-shared-000001")
+  task.replaceValueTextByKeyValue("index", ".ml-anomalies-custom-all-test-1,.ml-anomalies-custom-all-test-2", ".ml-anomalies-custom-all-test-1-000001,.ml-anomalies-custom-all-test-2-000001")
+  task.replaceValueTextByKeyValue("index", ".ml-anomalies-custom-all-test-1", ".ml-anomalies-custom-all-test-1-000001")
+  task.replaceValueTextByKeyValue("index", ".ml-anomalies-custom-all-test-2", ".ml-anomalies-custom-all-test-2-000001")
   task.skipTest("esql/60_usage/Basic ESQL usage output (telemetry)", "The telemetry output changed. We dropped a column. That's safe.")
   task.skipTest("inference/inference_crud/Test get all", "Assertions on number of inference models break due to default configs")
   task.skipTest("esql/60_usage/Basic ESQL usage output (telemetry) snapshot version", "The number of functions is constantly increasing")

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

@@ -1349,6 +1349,7 @@ public Job build() {
                 // Conditional guards against multiple prepending due to updates instead of first creation
                 resultsIndexName = resultsIndexName.startsWith("custom-") ? resultsIndexName : "custom-" + resultsIndexName;
             }
+
             if (datafeedConfig != null) {
                 if (datafeedConfig.getId() == null) {
                     datafeedConfig.setId(id);

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

@@ -20,6 +20,7 @@
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
 import org.elasticsearch.action.admin.indices.create.CreateIndexRequestBuilder;
 import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
+import org.elasticsearch.action.admin.indices.rollover.RolloverRequest;
 import org.elasticsearch.action.admin.indices.template.put.TransportPutComposableIndexTemplateAction;
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.action.support.IndicesOptions;
@@ -30,17 +31,23 @@
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.IndexVersions;
+import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.indices.SystemIndexDescriptor;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
 import org.elasticsearch.xcontent.json.JsonXContent;
+import org.elasticsearch.xpack.core.ml.job.config.Job;
+import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndex;
+import org.elasticsearch.xpack.core.ml.job.persistence.AnomalyDetectorsIndexFields;
 import org.elasticsearch.xpack.core.template.IndexTemplateConfig;
 
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Comparator;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.function.Predicate;
 import java.util.regex.Pattern;
@@ -70,6 +77,10 @@ public final class MlIndexAndAlias {
 
     private static final Logger logger = LogManager.getLogger(MlIndexAndAlias.class);
     private static final Predicate<String> HAS_SIX_DIGIT_SUFFIX = Pattern.compile("\\d{6}").asMatchPredicate();
+    private static final Predicate<String> IS_ANOMALIES_SHARED_INDEX = Pattern.compile(
+        AnomalyDetectorsIndexFields.RESULTS_INDEX_PREFIX + AnomalyDetectorsIndexFields.RESULTS_INDEX_DEFAULT + "-\\d{6}"
+    ).asMatchPredicate();
+    public static final String ROLLOVER_ALIAS_SUFFIX = ".rollover_alias";
 
     static final Comparator<String> INDEX_NAME_COMPARATOR = (index1, index2) -> {
         String[] index1Parts = index1.split("-");
@@ -224,6 +235,21 @@ public static void createIndexAndAliasIfNecessary(
         loggingListener.onResponse(false);
     }
 
+    /**
+     * Creates a system index based on the provided descriptor if it does not already exist.
+     * <p>
+     * The check for existence is simple and will return the listener on the calling thread if successful.
+     * If the index needs to be created an async call will be made and this method will wait for the index to reach at least
+     * a yellow health status before notifying the listener, ensuring it is ready for use
+     * upon a successful response. A {@link ResourceAlreadyExistsException} during creation
+     * is handled gracefully and treated as a success.
+     *
+     * @param client            The client to use for the create index request.
+     * @param clusterState      The current cluster state, used for the initial existence check.
+     * @param descriptor        The descriptor containing the index name, settings, and mappings.
+     * @param masterNodeTimeout The timeout for waiting on the master node.
+     * @param finalListener     Async listener
+     */
     public static void createSystemIndexIfNecessary(
         Client client,
         ClusterState clusterState,
@@ -323,6 +349,16 @@ private static void createFirstConcreteIndex(
         );
     }
 
+    /**
+     * Creates or moves a write alias from one index to another.
+     *
+     * @param client            The client to use for the add alias request.
+     * @param alias             The alias to update.
+     * @param currentIndex      The index the alias is currently pointing to.
+     * @param newIndex          The new index the alias should point to.
+     * @param masterNodeTimeout The timeout for waiting on the master node.
+     * @param listener          Async listener
+     */
     public static void updateWriteAlias(
         Client client,
         String alias,
@@ -357,7 +393,7 @@ public static void updateWriteAlias(
     /**
      * Installs the index template specified by {@code templateConfig} if it is not in already
      * installed in {@code clusterState}.
-     *
+     * <p>
      * The check for presence is simple and will return the listener on
      * the calling thread if successful. If the template has to be installed
      * an async call will be made.
@@ -427,17 +463,38 @@ public static void installIndexTemplateIfRequired(
         executeAsyncWithOrigin(client, ML_ORIGIN, TransportPutComposableIndexTemplateAction.TYPE, templateRequest, innerListener);
     }
 
-    public static boolean hasIndexTemplate(ClusterState state, String templateName, long version) {
+    private static boolean hasIndexTemplate(ClusterState state, String templateName, long version) {
         var template = state.getMetadata().getProject().templatesV2().get(templateName);
         return template != null && Long.valueOf(version).equals(template.version());
     }
 
+    public static String ensureValidResultsIndexName(String indexName) {
+        // The results index name is either the original one provided or the original with a suffix appended.
+        return has6DigitSuffix(indexName) ? indexName : indexName + FIRST_INDEX_SIX_DIGIT_SUFFIX;
+    }
+
+    /**
+     * Checks if an index name ends with a 6-digit suffix (e.g., "-000001").
+     *
+     * @param indexName The name of the index to check.
+     * @return {@code true} if the index name has a 6-digit suffix, {@code false} otherwise.
+     */
     public static boolean has6DigitSuffix(String indexName) {
         String[] indexParts = indexName.split("-");
         String suffix = indexParts[indexParts.length - 1];
         return HAS_SIX_DIGIT_SUFFIX.test(suffix);
     }
 
+    /**
+     * Checks if an index name matches the pattern for the default ML anomalies indices (e.g., ".ml-anomalies-shared-000001").
+     *
+     * @param indexName The name of the index to check.
+     * @return {@code true} if the index is a shared anomalies index, {@code false} otherwise.
+     */
+    public static boolean isAnomaliesSharedIndex(String indexName) {
+        return IS_ANOMALIES_SHARED_INDEX.test(indexName);
+    }
+
     /**
      * Returns the latest index. Latest is the index with the highest
      * 6 digit suffix.
@@ -456,4 +513,194 @@ public static String latestIndex(String[] concreteIndices) {
     public static boolean indexIsReadWriteCompatibleInV9(IndexVersion version) {
         return version.onOrAfter(IndexVersions.V_8_0_0);
     }
+
+    /**
+     * 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
+     */
+    public 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 + "*"
+        );
+
+        // We used to assert here if no matching indices could be found. However, when called _before_ a job is created it may be the case
+        // that no .ml-anomalies-shared* indices yet exist
+        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);
+    }
+
+    /**
+     * Executes a rollover request. It handles {@link ResourceAlreadyExistsException} gracefully by treating it as a success
+     * and returning the name of the existing index.
+     *
+     * @param client            The client to use for the rollover request.
+     * @param rolloverRequest   The rollover request to execute.
+     * @param listener          A listener that will be notified with the name of the new (or pre-existing) index on success,
+     *                          or an exception on failure.
+     */
+    public static void rollover(Client client, RolloverRequest rolloverRequest, ActionListener<String> listener) {
+        client.admin()
+            .indices()
+            .rolloverIndex(rolloverRequest, ActionListener.wrap(response -> listener.onResponse(response.getNewIndex()), e -> {
+                if (e instanceof ResourceAlreadyExistsException alreadyExistsException) {
+                    // The destination index already exists possibly because it has been rolled over already.
+                    listener.onResponse(alreadyExistsException.getIndex().getName());
+                } else {
+                    listener.onFailure(e);
+                }
+            }));
+    }
+
+    public static Tuple<String, String> createRolloverAliasAndNewIndexName(String index) {
+        String indexName = Objects.requireNonNull(index);
+
+        // Create an alias specifically for rolling over.
+        // The ml-anomalies index has aliases for each job, any
+        // of which could be used but that means one alias is
+        // treated differently.
+        // ROLLOVER_ALIAS_SUFFIX puts a `.` in the alias name to avoid any conflicts
+        // as AD job Ids cannot start with `.`
+        String rolloverAlias = indexName + ROLLOVER_ALIAS_SUFFIX;
+
+        // If the index does not end in a digit then rollover does not know
+        // what to name the new index so it must be specified in the request.
+        // Otherwise leave null and rollover will calculate the new name
+        String newIndexName = MlIndexAndAlias.has6DigitSuffix(index) ? null : indexName + MlIndexAndAlias.FIRST_INDEX_SIX_DIGIT_SUFFIX;
+
+        return new Tuple<>(rolloverAlias, newIndexName);
+    }
+
+    public static IndicesAliasesRequestBuilder createIndicesAliasesRequestBuilder(Client client) {
+        return client.admin().indices().prepareAliases(TimeValue.THIRTY_SECONDS, TimeValue.THIRTY_SECONDS);
+    }
+
+    /**
+     * Creates a hidden alias for an index, typically used as a rollover target.
+     *
+     * @param client      The client to use for the alias request.
+     * @param indexName   The name of the index to which the alias will be added.
+     * @param aliasName   The name of the alias to create.
+     * @param listener    A listener that will be notified with the response.
+     */
+    public static void createAliasForRollover(
+        Client client,
+        String indexName,
+        String aliasName,
+        ActionListener<IndicesAliasesResponse> listener
+    ) {
+        logger.info("creating rollover [{}] alias for [{}]", aliasName, indexName);
+        createIndicesAliasesRequestBuilder(client).addAliasAction(
+            IndicesAliasesRequest.AliasActions.add().index(indexName).alias(aliasName).isHidden(true)
+        ).execute(listener);
+    }
+
+    /**
+     * Executes a prepared {@link IndicesAliasesRequestBuilder} and notifies the listener of the result.
+     *
+     * @param request  The prepared request builder containing alias actions.
+     * @param listener A listener that will be notified with {@code true} on success.
+     */
+    public static void updateAliases(IndicesAliasesRequestBuilder request, ActionListener<Boolean> listener) {
+        request.execute(listener.delegateFailure((l, response) -> l.onResponse(Boolean.TRUE)));
+    }
+
+    /**
+     * Adds alias actions to a request builder to move ML job aliases from an old index to a new one after a rollover.
+     * This includes moving the write alias and re-creating the filtered read aliases on the new index.
+     *
+     * @param aliasRequestBuilder The request builder to add actions to.
+     * @param oldIndex            The index from which aliases are being moved.
+     * @param newIndex            The new index to which aliases will be moved.
+     * @param clusterState        The current cluster state, used to inspect existing aliases on the old index.
+     * @return The modified {@link IndicesAliasesRequestBuilder}.
+     */
+    public static IndicesAliasesRequestBuilder addIndexAliasesRequests(
+        IndicesAliasesRequestBuilder aliasRequestBuilder,
+        String oldIndex,
+        String newIndex,
+        ClusterState clusterState
+    ) {
+        // Multiple jobs can share the same index each job
+        // has a read and write alias that needs updating
+        // after the rollover
+        var meta = clusterState.metadata().getProject().index(oldIndex);
+        assert meta != null;
+        if (meta == null) {
+            return aliasRequestBuilder;
+        }
+
+        for (var alias : meta.getAliases().values()) {
+            if (isAnomaliesWriteAlias(alias.alias())) {
+                aliasRequestBuilder.addAliasAction(
+                    IndicesAliasesRequest.AliasActions.add().index(newIndex).alias(alias.alias()).isHidden(true).writeIndex(true)
+                );
+                aliasRequestBuilder.addAliasAction(IndicesAliasesRequest.AliasActions.remove().index(oldIndex).alias(alias.alias()));
+            } else if (isAnomaliesReadAlias(alias.alias())) {
+                String jobId = AnomalyDetectorsIndex.jobIdFromAlias(alias.alias());
+                aliasRequestBuilder.addAliasAction(
+                    IndicesAliasesRequest.AliasActions.add()
+                        .index(newIndex)
+                        .alias(alias.alias())
+                        .isHidden(true)
+                        .filter(QueryBuilders.termQuery(Job.ID.getPreferredName(), jobId))
+                );
+            }
+        }
+
+        return aliasRequestBuilder;
+    }
+
+    /**
+     * Determines if an alias name is an ML anomalies write alias.
+     *
+     * @param aliasName The alias name to check.
+     * @return {@code true} if the name matches the write alias pattern, {@code false} otherwise.
+     */
+    public static boolean isAnomaliesWriteAlias(String aliasName) {
+        return aliasName.startsWith(AnomalyDetectorsIndexFields.RESULTS_INDEX_WRITE_PREFIX);
+    }
+
+    /**
+     * Determines if an alias name is an ML anomalies read alias.
+     *
+     * @param aliasName The alias name to check.
+     * @return {@code true} if the name matches the read alias pattern, {@code false} otherwise.
+     */
+    public static boolean isAnomaliesReadAlias(String aliasName) {
+        if (aliasName.startsWith(AnomalyDetectorsIndexFields.RESULTS_INDEX_PREFIX) == false) {
+            return false;
+        }
+
+        // See {@link AnomalyDetectorsIndex#jobResultsAliasedName}
+        String jobIdPart = aliasName.substring(AnomalyDetectorsIndexFields.RESULTS_INDEX_PREFIX.length());
+        // If this is a write alias it will start with a `.` character
+        // which is not a valid job id.
+        return MlStrings.isValidId(jobIdPart);
+    }
 }