Bug Fix: System Data Streams Should Be Restorable

docs/changelog/124651.yaml

@@ -0,0 +1,5 @@
+pr: 124651
+summary: "System Data Streams Should Be Restorable"
+area: Infra/Core
+type: bug
+issues: []

server/build.gradle

@@ -80,6 +80,7 @@ dependencies {
   }
   internalClusterTestImplementation(project(':modules:reindex'))
   internalClusterTestImplementation(project(':modules:mapper-extras'))
+  internalClusterTestImplementation(project(':modules:data-streams'))
 }
 
 spotless {

server/src/main/java/org/elasticsearch/cluster/metadata/Metadata.java

@@ -1865,6 +1865,19 @@ public Optional<ProjectMetadata> lookupProject(Index index) {
         return getProjectLookup().project(index);
     }
 
+    /**
+     * Attempt to find a project for the supplied {@link DataStream}.
+     */
+    @FixForMultiProject
+    public Optional<ProjectMetadata> lookupProject(DataStream dataStream) {
+        // Assumes single project for now.
+        if (getProject().dataStreams().containsKey(dataStream.getName())) {
+            return Optional.of(getProject());
+        } else {
+            return Optional.empty();
+        }
+    }
+
     /**
      * Attempt to find a project for the supplied {@link Index}.
      * @throws org.elasticsearch.index.IndexNotFoundException if the index does not exist in any project
@@ -1875,6 +1888,16 @@ public ProjectMetadata projectFor(Index index) {
         );
     }
 
+    /**
+     * Attempt to find a project for the supplied {@link DataStream}.
+     * @throws java.lang.IllegalArgumentException if the data stream does not exist in any project
+     */
+    public ProjectMetadata projectFor(DataStream dataStream) {
+        return lookupProject(dataStream).orElseThrow(
+            () -> new IllegalArgumentException("dataStream [" + dataStream.getName() + "] does not exist in any project")
+        );
+    }
+
     /**
      * Attempt to find the IndexMetadata for the supplied {@link Index}.
      * @throws org.elasticsearch.index.IndexNotFoundException if the index does not exist in any project

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataDeleteDataStreamService.java

@@ -0,0 +1,83 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", 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", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.cluster.metadata;
+
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.ProjectState;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.FixForMultiProject;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.logging.LogManager;
+import org.elasticsearch.logging.Logger;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public class MetadataDeleteDataStreamService {
+
+    private static final Logger LOGGER = LogManager.getLogger(MetadataDeleteDataStreamService.class);
+
+    /**
+     * Removes the given data streams from the Cluster State.
+     *
+     * @param clusterState The cluster state
+     * @param dataStreams  The data streams to remove
+     * @param settings     The settings
+     * @return The updated Cluster State
+     */
+    @FixForMultiProject // Once callers have Project State we can update/remove this method.
+    public static ClusterState deleteDataStreams(ClusterState clusterState, Set<DataStream> dataStreams, Settings settings) {
+        final Map<ProjectId, Set<DataStream>> byProject = new HashMap<>();
+        for (DataStream dataStream : dataStreams) {
+            final ProjectMetadata project = clusterState.metadata().projectFor(dataStream);
+            byProject.computeIfAbsent(project.id(), ignore -> new HashSet<>()).add(dataStream);
+        }
+
+        for (final Map.Entry<ProjectId, Set<DataStream>> entry : byProject.entrySet()) {
+            clusterState = deleteDataStream(clusterState.projectState(entry.getKey()), entry.getValue(), settings);
+        }
+        return clusterState;
+    }
+
+    /**
+     * Removes the given data streams from the Project State.
+     *
+     * @param projectState The project state
+     * @param dataStreams  The data streams to remove
+     * @param settings     The settings
+     * @return The updated Project State
+     */
+    public static ClusterState deleteDataStream(ProjectState projectState, Set<DataStream> dataStreams, Settings settings) {
+        if (dataStreams.isEmpty()) {
+            return projectState.cluster();
+        }
+
+        Set<Index> backingIndicesToRemove = new HashSet<>();
+        for (DataStream dataStream : dataStreams) {
+            assert dataStream != null;
+            backingIndicesToRemove.addAll(dataStream.getIndices());
+            backingIndicesToRemove.addAll(dataStream.getFailureIndices());
+        }
+
+        // first delete the data streams and then the indices:
+        // (this to avoid data stream validation from failing when deleting an index that is part of a data stream
+        // without updating the data stream)
+        // TODO: change order when "delete index api" also updates the data stream the "index to be removed" is a member of
+        ClusterState newState = projectState.updatedState(builder -> {
+            dataStreams.stream().map(DataStream::getName).forEach(ds -> {
+                LOGGER.info("removing data stream [{}]", ds);
+                builder.removeDataStream(ds);
+            });
+        });
+        return MetadataDeleteIndexService.deleteIndices(newState.projectState(projectState.projectId()), backingIndicesToRemove, settings);
+    }
+}

server/src/main/java/org/elasticsearch/snapshots/RestoreService.java

@@ -35,6 +35,7 @@
 import org.elasticsearch.cluster.metadata.MappingMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.MetadataCreateIndexService;
+import org.elasticsearch.cluster.metadata.MetadataDeleteDataStreamService;
 import org.elasticsearch.cluster.metadata.MetadataDeleteIndexService;
 import org.elasticsearch.cluster.metadata.MetadataIndexStateService;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
@@ -765,6 +766,29 @@ private Set<Index> resolveSystemIndicesToDelete(ClusterState currentState, Set<S
             .collect(Collectors.toUnmodifiableSet());
     }
 
+    /**
+     * Resolves a set of datastream names that currently exist in the cluster that are part of a feature state which is
+     * about to be restored, and should therefore be removed prior to restoring those feature states from the snapshot.
+     *
+     * @param currentState           The current cluster state
+     * @param featureStatesToRestore A set of feature state names that are about to be restored
+     * @return A set of datastream names that should be removed based on the feature states being restored
+     */
+    private Set<DataStream> resolveSystemDataStreamsToDelete(ClusterState currentState, Set<String> featureStatesToRestore) {
+        if (featureStatesToRestore == null) {
+            return Collections.emptySet();
+        }
+
+        return featureStatesToRestore.stream()
+            .map(systemIndices::getFeature)
+            .filter(Objects::nonNull) // Features that aren't present on this node will be warned about in `getFeatureStatesToRestore`
+            .flatMap(feature -> feature.getDataStreamDescriptors().stream())
+            .map(SystemDataStreamDescriptor::getDataStreamName)
+            .filter(datastreamName -> currentState.metadata().getProject().dataStreams().containsKey(datastreamName))
+            .map(dataStreamName -> currentState.metadata().getProject().dataStreams().get(dataStreamName))
+            .collect(Collectors.toUnmodifiableSet());
+    }
+
     // visible for testing
     static DataStream updateDataStream(DataStream dataStream, Metadata.Builder metadata, RestoreSnapshotRequest request) {
         String dataStreamName = dataStream.getName();
@@ -1336,6 +1360,13 @@ public ClusterState execute(ClusterState currentState) {
                 settings
             );
 
+            // Clear out all existing system data streams
+            currentState = MetadataDeleteDataStreamService.deleteDataStreams(
+                currentState,
+                resolveSystemDataStreamsToDelete(currentState, featureStatesToRestore),
+                settings
+            );
+
             // List of searchable snapshots indices to restore
             final Set<Index> searchableSnapshotsIndices = new HashSet<>();