Make SnapshotsInProgress project compatible

This PR adds project-id to both SnapshotsInProgress and Snapshot so that
they are aware of projects and ready to handle snapshots from multiple
projects.

Relates: ES-10224

Author: ywangd

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -199,6 +199,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion INDEXING_STATS_INCLUDES_RECENT_WRITE_LOAD = def(9_034_0_00);
     public static final TransportVersion ESQL_AGGREGATE_METRIC_DOUBLE_LITERAL = def(9_035_0_00);
     public static final TransportVersion INDEX_METADATA_INCLUDES_RECENT_WRITE_LOAD = def(9_036_0_00);
+    public static final TransportVersion PROJECT_ID_IN_SNAPSHOT = def(9_037_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/cluster/DiffableUtils.java

@@ -146,6 +146,26 @@ public static <K, T, M extends Map<K, T>> boolean hasKey(MapDiff<K, T, M> diff,
         return false;
     }
 
+    /**
+     * Create a new MapDiff by transforming the keys with the provided keyFunction
+     */
+    public static <K1, K2, T extends Diffable<T>, M1 extends Map<K1, T>> MapDiff<K2, T, Map<K2, T>> jdkMapDiffWithUpdatedKeys(
+        MapDiff<K1, T, M1> diff,
+        Function<K1, K2> keyFunction,
+        KeySerializer<K2> keySerializer
+    ) {
+        final List<K2> deletes = diff.getDeletes().stream().map(keyFunction).toList();
+        final List<Map.Entry<K2, Diff<T>>> diffs = diff.getDiffs()
+            .stream()
+            .map(entry -> Map.entry(keyFunction.apply(entry.getKey()), entry.getValue()))
+            .toList();
+        final List<Map.Entry<K2, T>> upserts = diff.getUpserts()
+            .stream()
+            .map(entry -> Map.entry(keyFunction.apply(entry.getKey()), entry.getValue()))
+            .toList();
+        return new MapDiff<>(keySerializer, DiffableValueSerializer.getWriteOnlyInstance(), deletes, diffs, upserts, JdkMapBuilder::new);
+    }
+
     /**
      * Creates a MapDiff that applies a single entry diff to a map
      */

server/src/main/java/org/elasticsearch/cluster/SnapshotsInProgress.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.TransportVersion;
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.cluster.ClusterState.Custom;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.SingleNodeShutdownMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.Strings;
@@ -32,6 +33,7 @@
 import org.elasticsearch.logging.Logger;
 import org.elasticsearch.repositories.IndexId;
 import org.elasticsearch.repositories.RepositoryOperation;
+import org.elasticsearch.repositories.RepositoryOperation.ProjectRepo;
 import org.elasticsearch.repositories.RepositoryShardId;
 import org.elasticsearch.repositories.ShardGeneration;
 import org.elasticsearch.repositories.ShardSnapshotResult;
@@ -56,6 +58,9 @@
 import java.util.Set;
 import java.util.stream.Stream;
 
+import static org.elasticsearch.repositories.RepositoryOperation.PROJECT_REPO_SERIALIZER;
+import static org.elasticsearch.repositories.RepositoryOperation.defaultProjectRepo;
+
 /**
  * Meta data about snapshots that are currently executing
  */
@@ -70,7 +75,7 @@ public class SnapshotsInProgress extends AbstractNamedDiffable<Custom> implement
     public static final String ABORTED_FAILURE_TEXT = "Snapshot was aborted by deletion";
 
     /** Maps repository name to list of snapshots in that repository */
-    private final Map<String, ByRepo> entries;
+    private final Map<ProjectRepo, ByRepo> entries;
 
     /**
      * IDs of nodes which are marked for removal, or which were previously marked for removal and still have running shard snapshots.
@@ -104,56 +109,69 @@ private static Set<String> readNodeIdsForRemoval(StreamInput in) throws IOExcept
             : Set.of();
     }
 
-    private static Map<String, ByRepo> collectByRepo(StreamInput in) throws IOException {
+    private static Map<ProjectRepo, ByRepo> collectByRepo(StreamInput in) throws IOException {
         final int count = in.readVInt();
         if (count == 0) {
             return Map.of();
         }
-        final Map<String, List<Entry>> entriesByRepo = new HashMap<>();
+        final Map<ProjectRepo, List<Entry>> entriesByRepo = new HashMap<>();
         for (int i = 0; i < count; i++) {
             final Entry entry = Entry.readFrom(in);
-            entriesByRepo.computeIfAbsent(entry.repository(), repo -> new ArrayList<>()).add(entry);
+            entriesByRepo.computeIfAbsent(new ProjectRepo(entry.projectId(), entry.repository()), repo -> new ArrayList<>()).add(entry);
         }
-        final Map<String, ByRepo> res = Maps.newMapWithExpectedSize(entriesByRepo.size());
-        for (Map.Entry<String, List<Entry>> entryForRepo : entriesByRepo.entrySet()) {
+        final Map<ProjectRepo, ByRepo> res = Maps.newMapWithExpectedSize(entriesByRepo.size());
+        for (Map.Entry<ProjectRepo, List<Entry>> entryForRepo : entriesByRepo.entrySet()) {
             res.put(entryForRepo.getKey(), new ByRepo(entryForRepo.getValue()));
         }
         return res;
     }
 
-    private SnapshotsInProgress(Map<String, ByRepo> entries, Set<String> nodesIdsForRemoval) {
+    private SnapshotsInProgress(Map<ProjectRepo, ByRepo> entries, Set<String> nodesIdsForRemoval) {
         this.entries = Map.copyOf(entries);
         this.nodesIdsForRemoval = nodesIdsForRemoval;
         assert assertConsistentEntries(this.entries);
     }
 
+    @Deprecated(forRemoval = true)
     public SnapshotsInProgress withUpdatedEntriesForRepo(String repository, List<Entry> updatedEntries) {
-        if (updatedEntries.equals(forRepo(repository))) {
+        return withUpdatedEntriesForRepo(defaultProjectRepo(repository), updatedEntries);
+    }
+
+    public SnapshotsInProgress withUpdatedEntriesForRepo(ProjectRepo projectRepo, List<Entry> updatedEntries) {
+        if (updatedEntries.equals(forRepo(projectRepo))) {
             return this;
         }
-        final Map<String, ByRepo> copy = new HashMap<>(this.entries);
+        final Map<ProjectRepo, ByRepo> copy = new HashMap<>(this.entries);
         if (updatedEntries.isEmpty()) {
-            copy.remove(repository);
+            copy.remove(projectRepo);
             if (copy.isEmpty()) {
                 return EMPTY;
             }
         } else {
-            copy.put(repository, new ByRepo(updatedEntries));
+            copy.put(projectRepo, new ByRepo(updatedEntries));
         }
         return new SnapshotsInProgress(copy, nodesIdsForRemoval);
     }
 
     public SnapshotsInProgress withAddedEntry(Entry entry) {
-        final List<Entry> forRepo = new ArrayList<>(forRepo(entry.repository()));
+        final List<Entry> forRepo = new ArrayList<>(forRepo(entry.projectRepo()));
         forRepo.add(entry);
-        return withUpdatedEntriesForRepo(entry.repository(), forRepo);
+        return withUpdatedEntriesForRepo(entry.projectRepo(), forRepo);
     }
 
     /**
      * Returns the list of snapshots in the specified repository.
      */
+    @Deprecated(forRemoval = true)
     public List<Entry> forRepo(String repository) {
-        return entries.getOrDefault(repository, ByRepo.EMPTY).entries;
+        return entries.getOrDefault(defaultProjectRepo(repository), ByRepo.EMPTY).entries;
+    }
+
+    /**
+     * Returns the list of snapshots in the specified repository.
+     */
+    public List<Entry> forRepo(ProjectRepo projectRepo) {
+        return entries.getOrDefault(projectRepo, ByRepo.EMPTY).entries;
     }
 
     public boolean isEmpty() {
@@ -178,7 +196,7 @@ public Stream<Entry> asStream() {
 
     @Nullable
     public Entry snapshot(final Snapshot snapshot) {
-        return findSnapshotInList(snapshot, forRepo(snapshot.getRepository()));
+        return findSnapshotInList(snapshot, forRepo(snapshot.getProjectRepo()));
     }
 
     /**
@@ -206,14 +224,32 @@ private static Entry findSnapshotInList(Snapshot snapshotToFind, List<Entry> for
      * in-progress shard snapshots that were not yet finalized when it began. All these other in-progress shard snapshot lists are scheduled
      * for deletion now.
      */
+    @Deprecated(forRemoval = true)
     public Map<RepositoryShardId, Set<ShardGeneration>> obsoleteGenerations(
         String repository,
         SnapshotsInProgress oldClusterStateSnapshots
+    ) {
+        return obsoleteGenerations(defaultProjectRepo(repository), oldClusterStateSnapshots);
+    }
+
+    /**
+     * Computes a map of repository shard id to set of shard generations, containing all shard generations that became obsolete and may be
+     * deleted from the repository as the cluster state moves from the given old value of {@link SnapshotsInProgress} to this instance.
+     * <p>
+     * An unique shard generation is created for every in-progress shard snapshot. The shard generation file contains information about all
+     * the files needed by pre-existing and any new shard snapshots that were in-progress. When a shard snapshot is finalized, its file list
+     * is promoted to the official shard snapshot list for the index shard. This final list will contain metadata about any other
+     * in-progress shard snapshots that were not yet finalized when it began. All these other in-progress shard snapshot lists are scheduled
+     * for deletion now.
+     */
+    public Map<RepositoryShardId, Set<ShardGeneration>> obsoleteGenerations(
+        ProjectRepo projectRepo,
+        SnapshotsInProgress oldClusterStateSnapshots
     ) {
         final Map<RepositoryShardId, Set<ShardGeneration>> obsoleteGenerations = new HashMap<>();
-        final List<Entry> latestSnapshots = forRepo(repository);
+        final List<Entry> latestSnapshots = forRepo(projectRepo);
 
-        for (Entry oldEntry : oldClusterStateSnapshots.forRepo(repository)) {
+        for (Entry oldEntry : oldClusterStateSnapshots.forRepo(projectRepo)) {
             final Entry matchingLatestEntry = findSnapshotInList(oldEntry.snapshot(), latestSnapshots);
             if (matchingLatestEntry == null || matchingLatestEntry == oldEntry) {
                 // The snapshot progress has not changed.
@@ -412,15 +448,15 @@ private static boolean hasFailures(Map<RepositoryShardId, ShardSnapshotStatus> c
         return false;
     }
 
-    private static boolean assertConsistentEntries(Map<String, ByRepo> entries) {
-        for (Map.Entry<String, ByRepo> repoEntries : entries.entrySet()) {
+    private static boolean assertConsistentEntries(Map<ProjectRepo, ByRepo> entries) {
+        for (Map.Entry<ProjectRepo, ByRepo> repoEntries : entries.entrySet()) {
             final Set<Tuple<String, Integer>> assignedShards = new HashSet<>();
             final Set<Tuple<String, Integer>> queuedShards = new HashSet<>();
             final List<Entry> entriesForRepository = repoEntries.getValue().entries;
-            final String repository = repoEntries.getKey();
+            final ProjectRepo repository = repoEntries.getKey();
             assert entriesForRepository.isEmpty() == false : "found empty list of snapshots for " + repository + " in " + entries;
             for (Entry entry : entriesForRepository) {
-                assert entry.repository().equals(repository) : "mismatched repository " + entry + " tracked under " + repository;
+                assert entry.projectRepo().equals(repository) : "mismatched repository " + entry + " tracked under " + repository;
                 for (Map.Entry<RepositoryShardId, ShardSnapshotStatus> shard : entry.shardSnapshotStatusByRepoShardId().entrySet()) {
                     final RepositoryShardId sid = shard.getKey();
                     final ShardSnapshotStatus shardSnapshotStatus = shard.getValue();
@@ -1241,11 +1277,20 @@ public Entry withStartedShards(Map<ShardId, ShardSnapshotStatus> shards) {
             return updated;
         }
 
+        @Override
+        public ProjectId projectId() {
+            return snapshot.getProjectId();
+        }
+
         @Override
         public String repository() {
             return snapshot.getRepository();
         }
 
+        public ProjectRepo projectRepo() {
+            return new ProjectRepo(projectId(), repository());
+        }
+
         public Snapshot snapshot() {
             return this.snapshot;
         }
@@ -1391,6 +1436,7 @@ public String toString() {
         @Override
         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
             builder.startObject();
+            builder.field("project_id", snapshot.getProjectId());
             builder.field("repository", snapshot.getRepository());
             builder.field("snapshot", snapshot.getSnapshotId().getName());
             builder.field("uuid", snapshot.getSnapshotId().getUUID());
@@ -1725,25 +1771,43 @@ private static final class SnapshotInProgressDiff implements NamedDiff<Custom> {
 
         private final SnapshotsInProgress after;
 
-        private final DiffableUtils.MapDiff<String, ByRepo, Map<String, ByRepo>> mapDiff;
+        private final DiffableUtils.MapDiff<ProjectRepo, ByRepo, Map<ProjectRepo, ByRepo>> mapDiff;
         private final Set<String> nodeIdsForRemoval;
 
         SnapshotInProgressDiff(SnapshotsInProgress before, SnapshotsInProgress after) {
-            this.mapDiff = DiffableUtils.diff(before.entries, after.entries, DiffableUtils.getStringKeySerializer());
+            this.mapDiff = DiffableUtils.diff(before.entries, after.entries, PROJECT_REPO_SERIALIZER);
             this.nodeIdsForRemoval = after.nodesIdsForRemoval;
             this.after = after;
         }
 
         SnapshotInProgressDiff(StreamInput in) throws IOException {
-            this.mapDiff = DiffableUtils.readJdkMapDiff(
-                in,
-                DiffableUtils.getStringKeySerializer(),
-                i -> new ByRepo(i.readCollectionAsImmutableList(Entry::readFrom)),
-                i -> new ByRepo.ByRepoDiff(
-                    DiffableUtils.readJdkMapDiff(i, DiffableUtils.getStringKeySerializer(), Entry::readFrom, EntryDiff::new),
-                    DiffableUtils.readJdkMapDiff(i, DiffableUtils.getStringKeySerializer(), ByRepo.INT_DIFF_VALUE_SERIALIZER)
-                )
-            );
+            if (in.getTransportVersion().before(TransportVersions.PROJECT_ID_IN_SNAPSHOT)) {
+                final var oldMapDiff = DiffableUtils.readJdkMapDiff(
+                    in,
+                    DiffableUtils.getStringKeySerializer(),
+                    i -> new ByRepo(i.readCollectionAsImmutableList(Entry::readFrom)),
+                    i -> new ByRepo.ByRepoDiff(
+                        DiffableUtils.readJdkMapDiff(i, DiffableUtils.getStringKeySerializer(), Entry::readFrom, EntryDiff::new),
+                        DiffableUtils.readJdkMapDiff(i, DiffableUtils.getStringKeySerializer(), ByRepo.INT_DIFF_VALUE_SERIALIZER)
+                    )
+                );
+                this.mapDiff = DiffableUtils.jdkMapDiffWithUpdatedKeys(
+                    oldMapDiff,
+                    RepositoryOperation::defaultProjectRepo,
+                    PROJECT_REPO_SERIALIZER
+                );
+            } else {
+                this.mapDiff = DiffableUtils.readJdkMapDiff(
+                    in,
+                    PROJECT_REPO_SERIALIZER,
+                    i -> new ByRepo(i.readCollectionAsImmutableList(Entry::readFrom)),
+                    i -> new ByRepo.ByRepoDiff(
+                        DiffableUtils.readJdkMapDiff(i, DiffableUtils.getStringKeySerializer(), Entry::readFrom, EntryDiff::new),
+                        DiffableUtils.readJdkMapDiff(i, DiffableUtils.getStringKeySerializer(), ByRepo.INT_DIFF_VALUE_SERIALIZER)
+                    )
+                );
+            }
+
             this.nodeIdsForRemoval = readNodeIdsForRemoval(in);
             this.after = null;
         }
@@ -1768,7 +1832,21 @@ public String getWriteableName() {
         public void writeTo(StreamOutput out) throws IOException {
             assert after != null : "should only write instances that were diffed from this node's state";
             if (out.getTransportVersion().onOrAfter(DIFFABLE_VERSION)) {
-                mapDiff.writeTo(out);
+                if (out.getTransportVersion().before(TransportVersions.PROJECT_ID_IN_SNAPSHOT)) {
+                    DiffableUtils.jdkMapDiffWithUpdatedKeys(mapDiff, projectRepo -> {
+                        if (ProjectId.DEFAULT.equals(projectRepo.projectId()) == false) {
+                            throw new IllegalArgumentException(
+                                "Cannot write instance with non-default project id "
+                                    + projectRepo.projectId()
+                                    + " to version before "
+                                    + TransportVersions.PROJECT_ID_IN_SNAPSHOT
+                            );
+                        }
+                        return projectRepo.repoName();
+                    }, DiffableUtils.getStringKeySerializer()).writeTo(out);
+                } else {
+                    mapDiff.writeTo(out);
+                }
             } else {
                 new SimpleDiffable.CompleteDiff<>(after).writeTo(out);
             }

server/src/main/java/org/elasticsearch/repositories/RepositoryOperation.java

@@ -8,18 +8,69 @@
  */
 package org.elasticsearch.repositories;
 
+import org.elasticsearch.cluster.DiffableUtils;
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.core.FixForMultiProject;
+
+import java.io.IOException;
+
 /**
  * Coordinates of an operation that modifies a repository, assuming that repository at a specific generation.
  */
 public interface RepositoryOperation {
 
+    /**
+     * Project for which repository belongs to.
+     */
+    @FixForMultiProject(description = "default implementation is temporary")
+    default ProjectId projectId() {
+        return ProjectId.DEFAULT;
+    }
+
     /**
      * Name of the repository affected.
      */
     String repository();
 
+    default ProjectRepo projectRepo() {
+        return new ProjectRepo(projectId(), repository());
+    }
+
     /**
      * The repository state id at the time the operation began.
      */
     long repositoryStateId();
+
+    record ProjectRepo(ProjectId projectId, String repoName) implements Writeable {
+
+        public ProjectRepo(StreamInput in) throws IOException {
+            this(ProjectId.readFrom(in), in.readString());
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            projectId.writeTo(out);
+            out.writeString(repoName);
+        }
+    }
+
+    DiffableUtils.KeySerializer<ProjectRepo> PROJECT_REPO_SERIALIZER = new DiffableUtils.KeySerializer<>() {
+        @Override
+        public void writeKey(ProjectRepo key, StreamOutput out) throws IOException {
+            key.writeTo(out);
+        }
+
+        @Override
+        public ProjectRepo readKey(StreamInput in) throws IOException {
+            return new ProjectRepo(in);
+        }
+    };
+
+    @Deprecated(forRemoval = true)
+    static ProjectRepo defaultProjectRepo(String repoName) {
+        return new ProjectRepo(ProjectId.DEFAULT, repoName);
+    }
 }