Initial data stream lifecycle support for downsampling (#98609)

This adds data stream lifecycle service implementation support
for downsampling.
Time series backing indices for a data stream with a lifecycle
that configures downsampling will be marked as read-only,
downsampled, removed from the data stream, replaced with the
corresponding downsample index, and deleted.

Multiple rounds can be configured for a data stream, and the
latest matching round will be the first one to be executed.
If one downsampling operation is in progress, we wait until it's
finished and then we start the next downsampling operation.
Note that in this scenario a data stream could have the following
backing indices:
```
[.ds-metrics-2023.08.22-000002, downsample-10s-.ds-metrics-2023.08.22-000001]
```

If this data stream has multiple rounds of downsampling configured,
the first generation index will subsequently be downsampled again
(and again).

Author: andreidan

modules/data-streams/src/main/java/org/elasticsearch/datastreams/DataStreamsPlugin.java

@@ -104,6 +104,8 @@ public class DataStreamsPlugin extends Plugin implements ActionPlugin {
         Setting.Property.IndexScope,
         Setting.Property.Dynamic
     );
+    public static final String LIFECYCLE_CUSTOM_INDEX_METADATA_KEY = "data_stream_lifecycle";
+
     // The dependency of index.look_ahead_time is a cluster setting and currently there is no clean validation approach for this:
     private final SetOnce<UpdateTimeSeriesRangeService> updateTimeSeriesRangeService = new SetOnce<>();
     private final SetOnce<DataStreamLifecycleErrorStore> errorStoreInitialisationService = new SetOnce<>();

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleService.java

@@ -88,6 +88,11 @@ public Request(String[] names, @Nullable TimeValue dataRetention) {
             this(names, dataRetention, null);
         }
 
+        public Request(String[] names, DataStreamLifecycle lifecycle) {
+            this.names = names;
+            this.lifecycle = lifecycle;
+        }
+
         public Request(String[] names, @Nullable TimeValue dataRetention, @Nullable Boolean enabled) {
             this.names = names;
             this.lifecycle = DataStreamLifecycle.newBuilder().dataRetention(dataRetention).enabled(enabled == null || enabled).build();

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/action/PutDataStreamLifecycleAction.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.datastreams.lifecycle.downsampling;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.SimpleBatchedExecutor;
+import org.elasticsearch.core.Strings;
+import org.elasticsearch.core.Tuple;
+import org.elasticsearch.index.IndexNotFoundException;
+import org.elasticsearch.snapshots.SnapshotInProgressException;
+
+/**
+ * Cluster service task (batched) executor that executes the replacement of data stream backing index with its
+ * downsampled index.
+ * After the task is executed the executor issues a delete API call for the source index however, it doesn't
+ * hold up the task listener (nb we notify the listener before we call the delete API so we don't introduce
+ * weird partial failure scenarios - if the delete API fails the
+ * {@link org.elasticsearch.datastreams.lifecycle.DataStreamLifecycleService} will retry on the next run so the source index will get
+ * deleted)
+ */
+public class ReplaceBackingWithDownsampleIndexExecutor extends SimpleBatchedExecutor<ReplaceSourceWithDownsampleIndexTask, Void> {
+    private static final Logger LOGGER = LogManager.getLogger(ReplaceSourceWithDownsampleIndexTask.class);
+    private final Client client;
+
+    public ReplaceBackingWithDownsampleIndexExecutor(Client client) {
+        this.client = client;
+    }
+
+    @Override
+    public Tuple<ClusterState, Void> executeTask(ReplaceSourceWithDownsampleIndexTask task, ClusterState clusterState) throws Exception {
+        return Tuple.tuple(task.execute(clusterState), null);
+    }
+
+    @Override
+    public void taskSucceeded(ReplaceSourceWithDownsampleIndexTask task, Void unused) {
+        LOGGER.trace(
+            "Updated cluster state and replaced index [{}] with index [{}] in data stream [{}]",
+            task.getSourceBackingIndex(),
+            task.getDownsampleIndex(),
+            task.getDataStreamName()
+        );
+        task.getListener().onResponse(null);
+
+        // chain an optimistic delete of the source index call here (if it fails it'll be retried by the data stream lifecycle loop)
+        client.admin().indices().delete(new DeleteIndexRequest(task.getSourceBackingIndex()), new ActionListener<>() {
+            @Override
+            public void onResponse(AcknowledgedResponse acknowledgedResponse) {
+                if (acknowledgedResponse.isAcknowledged()) {
+                    LOGGER.info(
+                        "Data stream lifecycle successfully deleted index [{}] due to being replaced by the downsampled index [{}] in"
+                            + " data stream [{}]",
+                        task.getSourceBackingIndex(),
+                        task.getDownsampleIndex(),
+                        task.getDataStreamName()
+                    );
+                } else {
+                    LOGGER.trace(
+                        "The delete request for index [{}] was not acknowledged. Data stream lifecycle service will retry on the"
+                            + " next run if the index still exists",
+                        task.getSourceBackingIndex()
+                    );
+                }
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                if (e instanceof IndexNotFoundException) {
+                    // index was already deleted, treat this as a success
+                    return;
+                }
+
+                if (e instanceof SnapshotInProgressException) {
+                    LOGGER.info(
+                        "Data stream lifecycle is unable to delete index [{}] because it's part of an ongoing snapshot. Retrying on "
+                            + "the next data stream lifecycle run",
+                        task.getSourceBackingIndex()
+                    );
+                } else {
+                    LOGGER.error(
+                        () -> Strings.format(
+                            "Data stream lifecycle encountered an error trying to delete index [%s]. It will retry on its next run.",
+                            task.getSourceBackingIndex()
+                        ),
+                        e
+                    );
+                }
+            }
+        });
+    }
+}

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/downsampling/ReplaceBackingWithDownsampleIndexExecutor.java

@@ -0,0 +1,232 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.datastreams.lifecycle.downsampling;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ClusterStateTaskListener;
+import org.elasticsearch.cluster.metadata.DataStream;
+import org.elasticsearch.cluster.metadata.IndexAbstraction;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.index.IndexSettings;
+
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.elasticsearch.datastreams.DataStreamsPlugin.LIFECYCLE_CUSTOM_INDEX_METADATA_KEY;
+
+/**
+ * Cluster state task that replaces a source index in a data stream with its downsample index.
+ * In the process it will configure the origination date for the downsample index (so it can
+ * have a correct generation time).
+ */
+public class ReplaceSourceWithDownsampleIndexTask implements ClusterStateTaskListener {
+    private static final Logger LOGGER = LogManager.getLogger(ReplaceSourceWithDownsampleIndexTask.class);
+    public static final String REPLACEMENT_SOURCE_INDEX = "replacement_source_index";
+    private ActionListener<Void> listener;
+    private final String dataStreamName;
+    private final String sourceBackingIndex;
+    private final String downsampleIndex;
+
+    public ReplaceSourceWithDownsampleIndexTask(
+        String dataStreamName,
+        String sourceBackingIndex,
+        String downsampleIndex,
+        ActionListener<Void> listener
+    ) {
+        this.dataStreamName = dataStreamName;
+        this.sourceBackingIndex = sourceBackingIndex;
+        this.downsampleIndex = downsampleIndex;
+        this.listener = listener;
+    }
+
+    ClusterState execute(ClusterState state) {
+        LOGGER.trace(
+            "Updating cluster state to replace index [{}] with [{}] in data stream [{}]",
+            sourceBackingIndex,
+            downsampleIndex,
+            dataStreamName
+        );
+        IndexAbstraction sourceIndexAbstraction = state.metadata().getIndicesLookup().get(sourceBackingIndex);
+        IndexMetadata downsampleIndexMeta = state.metadata().index(downsampleIndex);
+        if (downsampleIndexMeta == null) {
+            // the downsample index doesn't exist anymore so nothing to replace here
+            LOGGER.trace(
+                "Received request replace index [{}] with [{}] in data stream [{}] but the replacement index [{}] doesn't exist."
+                    + "Nothing to do here.",
+                sourceBackingIndex,
+                downsampleIndex,
+                dataStreamName,
+                downsampleIndex
+            );
+            return state;
+        }
+        IndexMetadata sourceIndexMeta = state.metadata().index(sourceBackingIndex);
+        DataStream dataStream = state.metadata().dataStreams().get(dataStreamName);
+        if (sourceIndexAbstraction == null) {
+            // index was deleted in the meantime, so let's check if we can make sure the downsample index ends up in the
+            // data stream (if not already there)
+            if (dataStream != null
+                && dataStream.getIndices().stream().filter(index -> index.getName().equals(downsampleIndex)).findAny().isEmpty()) {
+                // add downsample index to data stream
+                LOGGER.trace(
+                    "unable find source index [{}] but adding index [{}] to data stream [{}]",
+                    sourceBackingIndex,
+                    downsampleIndex,
+                    dataStreamName
+                );
+                Metadata.Builder newMetaData = Metadata.builder(state.metadata())
+                    .put(dataStream.addBackingIndex(state.metadata(), downsampleIndexMeta.getIndex()));
+                return ClusterState.builder(state).metadata(newMetaData).build();
+            }
+        } else {
+            // the source index exists
+            DataStream sourceParentDataStream = sourceIndexAbstraction.getParentDataStream();
+            if (sourceParentDataStream != null) {
+                assert sourceParentDataStream.getName().equals(dataStreamName)
+                    : "the backing index must be part of the provided data "
+                        + "stream ["
+                        + dataStreamName
+                        + "] but it is instead part of data stream ["
+                        + sourceParentDataStream.getName()
+                        + "]";
+                if (sourceParentDataStream.getWriteIndex().getName().equals(sourceBackingIndex)) {
+                    String errorMessage = String.format(
+                        Locale.ROOT,
+                        "index [%s] is the write index for data stream [%s] and cannot be replaced",
+                        sourceBackingIndex,
+                        sourceParentDataStream.getName()
+                    );
+                    throw new IllegalStateException(errorMessage);
+                }
+                if (sourceIndexMeta != null) {
+                    // both indices exist, let's copy the origination date from the source index to the downsample index
+                    Metadata.Builder newMetaData = Metadata.builder(state.getMetadata());
+                    TimeValue generationLifecycleDate = dataStream.getGenerationLifecycleDate(sourceIndexMeta);
+                    assert generationLifecycleDate != null : "write index must never be downsampled, or replaced";
+                    IndexMetadata updatedDownsampleMetadata = copyDataStreamLifecycleState(
+                        sourceIndexMeta,
+                        downsampleIndexMeta,
+                        generationLifecycleDate.millis()
+                    );
+
+                    newMetaData.put(updatedDownsampleMetadata, true);
+                    // replace source with downsample
+                    newMetaData.put(dataStream.replaceBackingIndex(sourceIndexMeta.getIndex(), downsampleIndexMeta.getIndex()));
+                    return ClusterState.builder(state).metadata(newMetaData).build();
+                }
+            } else {
+                // the source index is not part of a data stream, so let's check if we can make sure the downsample index ends up in the
+                // data stream
+                if (dataStream != null
+                    && dataStream.getIndices().stream().filter(index -> index.getName().equals(downsampleIndex)).findAny().isEmpty()) {
+                    Metadata.Builder newMetaData = Metadata.builder(state.getMetadata());
+                    TimeValue generationLifecycleDate = dataStream.getGenerationLifecycleDate(sourceIndexMeta);
+                    assert generationLifecycleDate != null : "write index must never be downsampled, or replaced";
+
+                    IndexMetadata updatedDownsampleMetadata = copyDataStreamLifecycleState(
+                        sourceIndexMeta,
+                        downsampleIndexMeta,
+                        generationLifecycleDate.millis()
+                    );
+                    newMetaData.put(updatedDownsampleMetadata, true);
+                    // add downsample index to data stream
+                    newMetaData.put(dataStream.addBackingIndex(state.metadata(), downsampleIndexMeta.getIndex()));
+                    return ClusterState.builder(state).metadata(newMetaData).build();
+                }
+            }
+        }
+
+        return state;
+    }
+
+    /**
+     * Copies the data stream lifecycle state information from the source index to the destination.
+     * This ensures the destination index will have a generation time by setting the {@link IndexSettings#LIFECYCLE_ORIGINATION_DATE} and
+     * that the source index is confingured in the
+     * {@link org.elasticsearch.datastreams.DataStreamsPlugin#LIFECYCLE_CUSTOM_INDEX_METADATA_KEY} custom.
+     */
+    private static IndexMetadata copyDataStreamLifecycleState(
+        IndexMetadata source,
+        IndexMetadata dest,
+        long sourceIndexGenerationTimeMillis
+    ) {
+        IndexMetadata.Builder downsampleIndexBuilder = IndexMetadata.builder(dest);
+        Map<String, String> lifecycleCustomMetadata = source.getCustomData(LIFECYCLE_CUSTOM_INDEX_METADATA_KEY);
+        Map<String, String> newCustomMetadata = new HashMap<>();
+        if (lifecycleCustomMetadata != null) {
+            newCustomMetadata.putAll(lifecycleCustomMetadata);
+        }
+        newCustomMetadata.put(REPLACEMENT_SOURCE_INDEX, source.getIndex().getName());
+        downsampleIndexBuilder.putCustom(LIFECYCLE_CUSTOM_INDEX_METADATA_KEY, newCustomMetadata);
+
+        if (IndexSettings.LIFECYCLE_ORIGINATION_DATE_SETTING.exists(dest.getSettings()) == false) {
+            downsampleIndexBuilder.settings(
+                Settings.builder()
+                    .put(dest.getSettings())
+                    .put(IndexSettings.LIFECYCLE_ORIGINATION_DATE, sourceIndexGenerationTimeMillis)
+                    .build()
+            ).settingsVersion(dest.getSettingsVersion() + 1L);
+        }
+        return downsampleIndexBuilder.build();
+    }
+
+    @Override
+    public void onFailure(Exception e) {
+        if (listener != null) {
+            listener.onFailure(e);
+        }
+    }
+
+    public String getDataStreamName() {
+        return dataStreamName;
+    }
+
+    public String getSourceBackingIndex() {
+        return sourceBackingIndex;
+    }
+
+    public String getDownsampleIndex() {
+        return downsampleIndex;
+    }
+
+    public ActionListener<Void> getListener() {
+        return listener;
+    }
+
+    public void setListener(ActionListener<Void> listener) {
+        this.listener = listener;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ReplaceSourceWithDownsampleIndexTask that = (ReplaceSourceWithDownsampleIndexTask) o;
+        return Objects.equals(dataStreamName, that.dataStreamName)
+            && Objects.equals(sourceBackingIndex, that.sourceBackingIndex)
+            && Objects.equals(downsampleIndex, that.downsampleIndex);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(dataStreamName, sourceBackingIndex, downsampleIndex);
+    }
+}

modules/data-streams/src/main/java/org/elasticsearch/datastreams/lifecycle/downsampling/ReplaceSourceWithDownsampleIndexTask.java

@@ -46,7 +46,7 @@
  */
 public class DataStreamLifecycleFixtures {
 
-    static DataStream createDataStream(
+    public static DataStream createDataStream(
         Metadata.Builder builder,
         String dataStreamName,
         int backingIndicesCount,