Make DeleteSourceAndAddDownsampleToDS project-aware (#124808)

This is part of the work to make DLM project-aware.

Author: nielsbauman

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

@@ -53,6 +53,7 @@
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.ResolvedExpression;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.SelectorResolver;
 import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
@@ -63,6 +64,7 @@
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.core.FixForMultiProject;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Strings;
 import org.elasticsearch.core.TimeValue;
@@ -694,10 +696,13 @@ private void replaceBackingIndexWithDownsampleIndexOnce(DataStream dataStream, S
                     downsampleIndexName,
                     dataStream
                 );
+                @FixForMultiProject(description = "The correct project ID should be passed here")
+                final var projectId = ProjectId.DEFAULT;
                 swapSourceWithDownsampleIndexQueue.submitTask(
                     "data-stream-lifecycle-delete-source[" + backingIndexName + "]-add-to-datastream-[" + downsampleIndexName + "]",
                     new DeleteSourceAndAddDownsampleToDS(
                         settings,
+                        projectId,
                         dataStream.getName(),
                         backingIndexName,
                         downsampleIndexName,

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

@@ -17,8 +17,9 @@
 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.cluster.metadata.MetadataDeleteIndexService;
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexSettings;
@@ -39,18 +40,21 @@ public class DeleteSourceAndAddDownsampleToDS implements ClusterStateTaskListene
     private static final Logger LOGGER = LogManager.getLogger(DeleteSourceAndAddDownsampleToDS.class);
     private final Settings settings;
     private ActionListener<Void> listener;
+    private final ProjectId projectId;
     private final String dataStreamName;
     private final String sourceBackingIndex;
     private final String downsampleIndex;
 
     public DeleteSourceAndAddDownsampleToDS(
         Settings settings,
+        ProjectId projectId,
         String dataStreamName,
         String sourceBackingIndex,
         String downsampleIndex,
         ActionListener<Void> listener
     ) {
         this.settings = settings;
+        this.projectId = projectId;
         this.dataStreamName = dataStreamName;
         this.sourceBackingIndex = sourceBackingIndex;
         this.downsampleIndex = downsampleIndex;
@@ -64,7 +68,8 @@ ClusterState execute(ClusterState state) {
             downsampleIndex,
             dataStreamName
         );
-        IndexMetadata downsampleIndexMeta = state.metadata().getProject().index(downsampleIndex);
+        final var project = state.metadata().getProject(projectId);
+        IndexMetadata downsampleIndexMeta = project.index(downsampleIndex);
         if (downsampleIndexMeta == null) {
             // the downsample index doesn't exist anymore so nothing to replace here
             LOGGER.trace(
@@ -77,9 +82,9 @@ ClusterState execute(ClusterState state) {
             );
             return state;
         }
-        IndexAbstraction sourceIndexAbstraction = state.metadata().getProject().getIndicesLookup().get(sourceBackingIndex);
+        IndexAbstraction sourceIndexAbstraction = project.getIndicesLookup().get(sourceBackingIndex);
         if (sourceIndexAbstraction == null) {
-            DataStream dataStream = state.metadata().getProject().dataStreams().get(dataStreamName);
+            DataStream dataStream = project.dataStreams().get(dataStreamName);
             // 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
@@ -91,9 +96,9 @@ ClusterState execute(ClusterState state) {
                     downsampleIndex,
                     dataStreamName
                 );
-                Metadata.Builder newMetaData = Metadata.builder(state.metadata())
-                    .put(dataStream.addBackingIndex(state.metadata().getProject(), downsampleIndexMeta.getIndex()));
-                return ClusterState.builder(state).metadata(newMetaData).build();
+                ProjectMetadata.Builder newProject = ProjectMetadata.builder(project)
+                    .put(dataStream.addBackingIndex(project, downsampleIndexMeta.getIndex()));
+                return ClusterState.builder(state).putProjectMetadata(newProject).build();
             }
         } else {
             DataStream sourceParentDataStream = sourceIndexAbstraction.getParentDataStream();
@@ -107,13 +112,13 @@ ClusterState execute(ClusterState state) {
                 throw new IllegalStateException(errorMessage);
             }
 
-            IndexMetadata sourceIndexMeta = state.metadata().getProject().index(sourceBackingIndex);
+            IndexMetadata sourceIndexMeta = project.index(sourceBackingIndex);
             assert sourceIndexMeta != null
                 : "the source index abstraction exists in the indices lookup, so the index metadata must "
                     + "exist in the same cluster state metadata";
             // the source index exists so let's start by deleting it
-            state = MetadataDeleteIndexService.deleteIndices(state, Set.of(sourceIndexMeta.getIndex()), settings);
-            DataStream dataStream = state.metadata().getProject().dataStreams().get(dataStreamName);
+            state = MetadataDeleteIndexService.deleteIndices(state.projectState(projectId), Set.of(sourceIndexMeta.getIndex()), settings);
+            DataStream dataStream = state.metadata().getProject(projectId).dataStreams().get(dataStreamName);
             if (sourceParentDataStream != null) {
                 assert sourceParentDataStream.getName().equals(dataStreamName)
                     : "the backing index must be part of the provided data "
@@ -142,13 +147,14 @@ ClusterState execute(ClusterState state) {
      * This method is private as it fits into the flow of this cluster state task - i.e. the source index has already been removed from
      * the provided state.
      */
-    private static ClusterState addDownsampleIndexToDataStream(
+    private ClusterState addDownsampleIndexToDataStream(
         ClusterState state,
         DataStream dataStream,
         IndexMetadata sourceIndexMeta,
         IndexMetadata downsampleIndexMeta
     ) {
-        Metadata.Builder newMetaData = Metadata.builder(state.getMetadata());
+        final var project = state.metadata().getProject(projectId);
+        ProjectMetadata.Builder newProject = ProjectMetadata.builder(project);
         TimeValue generationLifecycleDate = dataStream.getGenerationLifecycleDate(sourceIndexMeta);
         // the generation lifecycle date is null only for the write index
         // we fail already if attempting to delete/downsample the write index, so the following assertion just re-inforces that
@@ -159,10 +165,10 @@ private static ClusterState addDownsampleIndexToDataStream(
             generationLifecycleDate.millis()
         );
 
-        newMetaData.put(updatedDownsampleMetadata, true);
+        newProject.put(updatedDownsampleMetadata, true);
         // we deleted the source already so let's add the downsample index to the data stream
-        newMetaData.put(dataStream.addBackingIndex(state.metadata().getProject(), downsampleIndexMeta.getIndex()));
-        return ClusterState.builder(state).metadata(newMetaData).build();
+        newProject.put(dataStream.addBackingIndex(project, downsampleIndexMeta.getIndex()));
+        return ClusterState.builder(state).putProjectMetadata(newProject).build();
     }
 
     /**

modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/DataStreamLifecycleFixtures.java

@@ -18,6 +18,7 @@
 import org.elasticsearch.cluster.metadata.DataStreamLifecycle;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.metadata.Template;
 import org.elasticsearch.common.compress.CompressedXContent;
 import org.elasticsearch.common.settings.Settings;
@@ -54,6 +55,22 @@ public static DataStream createDataStream(
         Settings.Builder backingIndicesSettings,
         @Nullable DataStreamLifecycle lifecycle,
         Long now
+    ) {
+        var projectBuilder = builder.getProject(Metadata.DEFAULT_PROJECT_ID);
+        if (projectBuilder == null) {
+            projectBuilder = ProjectMetadata.builder(Metadata.DEFAULT_PROJECT_ID);
+            builder.put(projectBuilder);
+        }
+        return createDataStream(projectBuilder, dataStreamName, backingIndicesCount, 0, backingIndicesSettings, lifecycle, now);
+    }
+
+    public static DataStream createDataStream(
+        ProjectMetadata.Builder builder,
+        String dataStreamName,
+        int backingIndicesCount,
+        Settings.Builder backingIndicesSettings,
+        @Nullable DataStreamLifecycle lifecycle,
+        Long now
     ) {
         return createDataStream(builder, dataStreamName, backingIndicesCount, 0, backingIndicesSettings, lifecycle, now);
     }
@@ -66,6 +83,31 @@ public static DataStream createDataStream(
         Settings.Builder backingIndicesSettings,
         @Nullable DataStreamLifecycle lifecycle,
         Long now
+    ) {
+        var projectBuilder = builder.getProject(Metadata.DEFAULT_PROJECT_ID);
+        if (projectBuilder == null) {
+            projectBuilder = ProjectMetadata.builder(Metadata.DEFAULT_PROJECT_ID);
+            builder.put(projectBuilder);
+        }
+        return createDataStream(
+            projectBuilder,
+            dataStreamName,
+            backingIndicesCount,
+            failureIndicesCount,
+            backingIndicesSettings,
+            lifecycle,
+            now
+        );
+    }
+
+    public static DataStream createDataStream(
+        ProjectMetadata.Builder builder,
+        String dataStreamName,
+        int backingIndicesCount,
+        int failureIndicesCount,
+        Settings.Builder backingIndicesSettings,
+        @Nullable DataStreamLifecycle lifecycle,
+        Long now
     ) {
         final List<Index> backingIndices = new ArrayList<>();
         final List<Index> failureIndices = new ArrayList<>();

modules/data-streams/src/test/java/org/elasticsearch/datastreams/lifecycle/downsampling/DeleteSourceAndAddDownsampleIndexExecutorTests.java

@@ -25,6 +25,7 @@
 public class DeleteSourceAndAddDownsampleIndexExecutorTests extends ESTestCase {
 
     public void testExecutorNotifiesListenerAndReroutesAllocationService() {
+        final var projectId = randomUniqueProjectId();
         String dataStreamName = randomAlphaOfLengthBetween(10, 100);
         String sourceIndex = randomAlphaOfLengthBetween(10, 100);
         String downsampleIndex = randomAlphaOfLengthBetween(10, 100);
@@ -34,18 +35,25 @@ public void testExecutorNotifiesListenerAndReroutesAllocationService() {
 
         AtomicBoolean taskListenerCalled = new AtomicBoolean(false);
         executor.taskSucceeded(
-            new DeleteSourceAndAddDownsampleToDS(Settings.EMPTY, dataStreamName, sourceIndex, downsampleIndex, new ActionListener<>() {
-                @Override
-                public void onResponse(Void unused) {
-                    taskListenerCalled.set(true);
-                }
+            new DeleteSourceAndAddDownsampleToDS(
+                Settings.EMPTY,
+                projectId,
+                dataStreamName,
+                sourceIndex,
+                downsampleIndex,
+                new ActionListener<>() {
+                    @Override
+                    public void onResponse(Void unused) {
+                        taskListenerCalled.set(true);
+                    }
 
-                @Override
-                public void onFailure(Exception e) {
-                    logger.error(e.getMessage(), e);
-                    fail("unexpected exception: " + e.getMessage());
+                    @Override
+                    public void onFailure(Exception e) {
+                        logger.error(e.getMessage(), e);
+                        fail("unexpected exception: " + e.getMessage());
+                    }
                 }
-            }),
+            ),
             null
         );
         assertThat(taskListenerCalled.get(), is(true));