Fix Snapshots Capturing Incomplete Datastreams (#58630)

Only snapshot datastreams that are recorded in `SnapshotInfo` and clean those
that aren't from the snapshotted metadata.
Do not restore all datastreams by default when restoring global metadata, use the same
mechanics used for indices here.

Closes #58544

server/src/internalClusterTest/java/org/elasticsearch/snapshots/DataStreamsSnapshotsIT.java

@@ -42,6 +42,10 @@ import java.nio.file.Path;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.is;
 
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 
@@ -258,4 +262,22 @@ public class DataStreamsSnapshotsIT extends AbstractSnapshotIntegTestCase {
         GetDataStreamAction.Request getRequest = new GetDataStreamAction.Request("ds");
         expectThrows(ResourceNotFoundException.class, () -> client.admin().indices().getDataStreams(getRequest).actionGet());
     }
+
+    public void testDataStreamNotIncludedInLimitedSnapshot() throws ExecutionException, InterruptedException {
+        final String snapshotName = "test-snap";
+        CreateSnapshotResponse createSnapshotResponse = client.admin().cluster()
+                .prepareCreateSnapshot(REPO, snapshotName)
+                .setWaitForCompletion(true)
+                .setIndices("does-not-exist-*")
+                .setIncludeGlobalState(true)
+                .get();
+        assertThat(createSnapshotResponse.getSnapshotInfo().state(), is(SnapshotState.SUCCESS));
+
+        assertThat(client().admin().indices()
+                .deleteDataStream(new DeleteDataStreamAction.Request("*")).get().isAcknowledged(), is(true));
+
+        final RestoreSnapshotResponse restoreSnapshotResponse =
+                client().admin().cluster().prepareRestoreSnapshot(REPO, snapshotName).get();
+        assertThat(restoreSnapshotResponse.getRestoreInfo().indices(), empty());
+    }
 }

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

@@ -216,9 +216,12 @@ public class RestoreService implements ClusterStateApplier {
                     dataStreams = new HashMap<>();
                 } else {
                     globalMetadata = repository.getSnapshotGlobalMetadata(snapshotId);
-                    dataStreams = globalMetadata.dataStreams();
-                    if (request.includeGlobalState() == false) {
-                        dataStreams.keySet().retainAll(requestedDataStreams);
+                    final Map<String, DataStream> dataStreamsInSnapshot = globalMetadata.dataStreams();
+                    dataStreams = new HashMap<>(requestedDataStreams.size());
+                    for (String requestedDataStream : requestedDataStreams) {
+                        final DataStream dataStreamInSnapshot = dataStreamsInSnapshot.get(requestedDataStream);
+                        assert dataStreamInSnapshot != null : "DataStream [" + requestedDataStream + "] not found in snapshot";
+                        dataStreams.put(requestedDataStream, dataStreamInSnapshot);
                     }
                 }
                 requestIndices.removeAll(dataStreams.keySet());

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

@@ -230,13 +230,8 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
                 List<String> indices = Arrays.asList(indexNameExpressionResolver.concreteIndexNames(currentState,
                     request.indicesOptions(), true, request.indices()));
 
-                Map<String, DataStream> allDataStreams = currentState.metadata().dataStreams();
-                List<String> dataStreams;
-                if (request.includeGlobalState()) {
-                    dataStreams = new ArrayList<>(allDataStreams.keySet());
-                } else {
-                    dataStreams = indexNameExpressionResolver.dataStreamNames(currentState, request.indicesOptions(), request.indices());
-                }
+                final List<String> dataStreams =
+                        indexNameExpressionResolver.dataStreamNames(currentState, request.indicesOptions(), request.indices());
 
                 logger.trace("[{}][{}] creating snapshot for indices [{}]", repositoryName, snapshotName, indices);
 
@@ -354,9 +349,10 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
     }
 
     private static Metadata metadataForSnapshot(SnapshotsInProgress.Entry snapshot, Metadata metadata) {
+        final Metadata.Builder builder;
         if (snapshot.includeGlobalState() == false) {
             // Remove global state from the cluster state
-            Metadata.Builder builder = Metadata.builder();
+            builder = Metadata.builder();
             for (IndexId index : snapshot.indices()) {
                 final IndexMetadata indexMetadata = metadata.index(index.getName());
                 if (indexMetadata == null) {
@@ -365,21 +361,21 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
                     builder.put(indexMetadata, false);
                 }
             }
-
-            Map<String, DataStream> dataStreams = new HashMap<>();
-            for (String dataStreamName : snapshot.dataStreams()) {
-                DataStream dataStream = metadata.dataStreams().get(dataStreamName);
-                if (dataStream == null) {
-                    assert snapshot.partial() : "Data stream [" + dataStreamName +
+        } else {
+            builder = Metadata.builder(metadata);
+        }
+        // Only keep those data streams in the metadata that were actually requested by the initial snapshot create operation
+        Map<String, DataStream> dataStreams = new HashMap<>();
+        for (String dataStreamName : snapshot.dataStreams()) {
+            DataStream dataStream = metadata.dataStreams().get(dataStreamName);
+            if (dataStream == null) {
+                assert snapshot.partial() : "Data stream [" + dataStreamName +
                         "] was deleted during a snapshot but snapshot was not partial.";
-                } else {
-                    dataStreams.put(dataStreamName, dataStream);
-                }
+            } else {
+                dataStreams.put(dataStreamName, dataStream);
             }
-            builder.dataStreams(dataStreams);
-            metadata = builder.build();
-        }
-        return metadata;
+        };
+        return builder.dataStreams(dataStreams).build();
     }
 
     /**