Use Cluster State to Track Repository Generation (#49729)

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/get/GetRepositoriesResponse.java

@@ -29,6 +29,7 @@
 import org.elasticsearch.common.xcontent.XContentParser;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.List;
 
 import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
@@ -66,7 +67,8 @@ public void writeTo(StreamOutput out) throws IOException {
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
-        repositories.toXContent(builder, params);
+        repositories.toXContent(builder,
+            new DelegatingMapParams(Collections.singletonMap(RepositoriesMetaData.HIDE_GENERATIONS_PARAM, "true"), params));
         builder.endObject();
         return builder;
     }

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

@@ -24,12 +24,15 @@
 import org.elasticsearch.cluster.AbstractNamedDiffable;
 import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.cluster.metadata.MetaData.Custom;
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.repositories.RepositoryData;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -45,6 +48,12 @@ public class RepositoriesMetaData extends AbstractNamedDiffable<Custom> implemen
 
     public static final String TYPE = "repositories";
 
+    /**
+     * Serialization parameter used to hide the {@link RepositoryMetaData#generation()} and {@link RepositoryMetaData#pendingGeneration()}
+     * in {@link org.elasticsearch.action.admin.cluster.repositories.get.GetRepositoriesResponse}.
+     */
+    public static final String HIDE_GENERATIONS_PARAM = "hide_generations";
+
     private final List<RepositoryMetaData> repositories;
 
     /**
@@ -56,6 +65,30 @@ public RepositoriesMetaData(List<RepositoryMetaData> repositories) {
         this.repositories = Collections.unmodifiableList(repositories);
     }
 
+    /**
+     * Creates a new instance that has the given repository moved to the given {@code safeGeneration} and {@code pendingGeneration}.
+     *
+     * @param repoName          repository name
+     * @param safeGeneration    new safe generation
+     * @param pendingGeneration new pending generation
+     * @return new instance with updated generations
+     */
+    public RepositoriesMetaData withUpdatedGeneration(String repoName, long safeGeneration, long pendingGeneration) {
+        int indexOfRepo = -1;
+        for (int i = 0; i < repositories.size(); i++) {
+            if (repositories.get(i).name().equals(repoName)) {
+                indexOfRepo = i;
+                break;
+            }
+        }
+        if (indexOfRepo < 0) {
+            throw new IllegalArgumentException("Unknown repository [" + repoName + "]");
+        }
+        final List<RepositoryMetaData> updatedRepos = new ArrayList<>(repositories);
+        updatedRepos.set(indexOfRepo, new RepositoryMetaData(repositories.get(indexOfRepo), safeGeneration, pendingGeneration));
+        return new RepositoriesMetaData(updatedRepos);
+    }
+
     /**
      * Returns list of currently registered repositories
      *
@@ -88,7 +121,29 @@ public boolean equals(Object o) {
         RepositoriesMetaData that = (RepositoriesMetaData) o;
 
         return repositories.equals(that.repositories);
+    }
 
+    /**
+     * Checks if this instance and the given instance share the same repositories by checking that this instances' repositories and the
+     * repositories in {@code other} are equal or only differ in their values of {@link RepositoryMetaData#generation()} and
+     * {@link RepositoryMetaData#pendingGeneration()}.
+     *
+     * @param other other repositories metadata
+     * @return {@code true} iff both instances contain the same repositories apart from differences in generations
+     */
+    public boolean equalsIgnoreGenerations(@Nullable RepositoriesMetaData other) {
+        if (other == null) {
+            return false;
+        }
+        if (other.repositories.size() != repositories.size()) {
+            return false;
+        }
+        for (int i = 0; i < repositories.size(); i++) {
+            if (repositories.get(i).equalsIgnoreGenerations(other.repositories.get(i)) == false) {
+                return false;
+            }
+        }
+        return true;
     }
 
     @Override
@@ -143,6 +198,8 @@ public static RepositoriesMetaData fromXContent(XContentParser parser) throws IO
                 }
                 String type = null;
                 Settings settings = Settings.EMPTY;
+                long generation = RepositoryData.UNKNOWN_REPO_GEN;
+                long pendingGeneration = RepositoryData.EMPTY_REPO_GEN;
                 while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
                     if (token == XContentParser.Token.FIELD_NAME) {
                         String currentFieldName = parser.currentName();
@@ -156,6 +213,16 @@ public static RepositoriesMetaData fromXContent(XContentParser parser) throws IO
                                 throw new ElasticsearchParseException("failed to parse repository [{}], incompatible params", name);
                             }
                             settings = Settings.fromXContent(parser);
+                        } else if ("generation".equals(currentFieldName)) {
+                            if (parser.nextToken() != XContentParser.Token.VALUE_NUMBER) {
+                                throw new ElasticsearchParseException("failed to parse repository [{}], unknown type", name);
+                            }
+                            generation = parser.longValue();
+                        } else if ("pending_generation".equals(currentFieldName)) {
+                            if (parser.nextToken() != XContentParser.Token.VALUE_NUMBER) {
+                                throw new ElasticsearchParseException("failed to parse repository [{}], unknown type", name);
+                            }
+                            pendingGeneration = parser.longValue();
                         } else {
                             throw new ElasticsearchParseException("failed to parse repository [{}], unknown field [{}]",
                                 name, currentFieldName);
@@ -167,7 +234,7 @@ public static RepositoriesMetaData fromXContent(XContentParser parser) throws IO
                 if (type == null) {
                     throw new ElasticsearchParseException("failed to parse repository [{}], missing repository type", name);
                 }
-                repository.add(new RepositoryMetaData(name, type, settings));
+                repository.add(new RepositoryMetaData(name, type, settings, generation, pendingGeneration));
             } else {
                 throw new ElasticsearchParseException("failed to parse repositories");
             }
@@ -205,6 +272,15 @@ public static void toXContent(RepositoryMetaData repository, XContentBuilder bui
         repository.settings().toXContent(builder, params);
         builder.endObject();
 
+        if (params.paramAsBoolean(HIDE_GENERATIONS_PARAM, false) == false) {
+            builder.field("generation", repository.generation());
+            builder.field("pending_generation", repository.pendingGeneration());
+        }
         builder.endObject();
     }
+
+    @Override
+    public String toString() {
+        return Strings.toString(this);
+    }
 }

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

@@ -18,20 +18,36 @@
  */
 package org.elasticsearch.cluster.metadata;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.repositories.RepositoryData;
 
 import java.io.IOException;
+import java.util.Objects;
 
 /**
  * Metadata about registered repository
  */
 public class RepositoryMetaData {
+
+    public static final Version REPO_GEN_IN_CS_VERSION = Version.V_7_6_0;
+
     private final String name;
     private final String type;
     private final Settings settings;
 
+    /**
+     * Safe repository generation.
+     */
+    private final long generation;
+
+    /**
+     * Pending repository generation.
+     */
+    private final long pendingGeneration;
+
     /**
      * Constructs new repository metadata
      *
@@ -40,9 +56,21 @@ public class RepositoryMetaData {
      * @param settings repository settings
      */
     public RepositoryMetaData(String name, String type, Settings settings) {
+        this(name, type, settings, RepositoryData.UNKNOWN_REPO_GEN, RepositoryData.EMPTY_REPO_GEN);
+    }
+
+    public RepositoryMetaData(RepositoryMetaData metaData, long generation, long pendingGeneration) {
+        this(metaData.name, metaData.type, metaData.settings, generation, pendingGeneration);
+    }
+
+    public RepositoryMetaData(String name, String type, Settings settings, long generation, long pendingGeneration) {
         this.name = name;
         this.type = type;
         this.settings = settings;
+        this.generation = generation;
+        this.pendingGeneration = pendingGeneration;
+        assert generation <= pendingGeneration :
+            "Pending generation [" + pendingGeneration + "] must be greater or equal to generation [" + generation + "]";
     }
 
     /**
@@ -72,11 +100,41 @@ public Settings settings() {
         return this.settings;
     }
 
+    /**
+     * Returns the safe repository generation. {@link RepositoryData} for this generation is assumed to exist in the repository.
+     * All operations on the repository must be based on the {@link RepositoryData} at this generation.
+     * See package level documentation for the blob store based repositories {@link org.elasticsearch.repositories.blobstore} for details
+     * on how this value is used during snapshots.
+     * @return safe repository generation
+     */
+    public long generation() {
+        return generation;
+    }
+
+    /**
+     * Returns the pending repository generation. {@link RepositoryData} for this generation and all generations down to the safe
+     * generation {@link #generation} may exist in the repository and should not be reused for writing new {@link RepositoryData} to the
+     * repository.
+     * See package level documentation for the blob store based repositories {@link org.elasticsearch.repositories.blobstore} for details
+     * on how this value is used during snapshots.
+     *
+     * @return highest pending repository generation
+     */
+    public long pendingGeneration() {
+        return pendingGeneration;
+    }
 
     public RepositoryMetaData(StreamInput in) throws IOException {
         name = in.readString();
         type = in.readString();
         settings = Settings.readSettingsFromStream(in);
+        if (in.getVersion().onOrAfter(REPO_GEN_IN_CS_VERSION)) {
+            generation = in.readLong();
+            pendingGeneration = in.readLong();
+        } else {
+            generation = RepositoryData.UNKNOWN_REPO_GEN;
+            pendingGeneration = RepositoryData.EMPTY_REPO_GEN;
+        }
     }
 
     /**
@@ -88,6 +146,20 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeString(name);
         out.writeString(type);
         Settings.writeSettingsToStream(settings, out);
+        if (out.getVersion().onOrAfter(REPO_GEN_IN_CS_VERSION)) {
+            out.writeLong(generation);
+            out.writeLong(pendingGeneration);
+        }
+    }
+
+    /**
+     * Checks if this instance is equal to the other instance in all fields other than {@link #generation} and {@link #pendingGeneration}.
+     *
+     * @param other other repository metadata
+     * @return {@code true} if both instances equal in all fields but the generation fields
+     */
+    public boolean equalsIgnoreGenerations(RepositoryMetaData other) {
+        return name.equals(other.name) && type.equals(other.type()) && settings.equals(other.settings());
     }
 
     @Override
@@ -99,15 +171,18 @@ public boolean equals(Object o) {
 
         if (!name.equals(that.name)) return false;
         if (!type.equals(that.type)) return false;
+        if (generation != that.generation) return false;
+        if (pendingGeneration != that.pendingGeneration) return false;
         return settings.equals(that.settings);
-
     }
 
     @Override
     public int hashCode() {
-        int result = name.hashCode();
-        result = 31 * result + type.hashCode();
-        result = 31 * result + settings.hashCode();
-        return result;
+        return Objects.hash(name, type, settings, generation, pendingGeneration);
+    }
+
+    @Override
+    public String toString() {
+        return "RepositoryMetaData{" + name + "}{" + type + "}{" + settings + "}{" + generation + "}{" + pendingGeneration + "}";
     }
 }

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

@@ -150,7 +150,7 @@ public ClusterState execute(ClusterState currentState) {
 
                         for (RepositoryMetaData repositoryMetaData : repositories.repositories()) {
                             if (repositoryMetaData.name().equals(newRepositoryMetaData.name())) {
-                                if (newRepositoryMetaData.equals(repositoryMetaData)) {
+                                if (newRepositoryMetaData.equalsIgnoreGenerations(repositoryMetaData)) {
                                     // Previous version is the same as this one no update is needed.
                                     return currentState;
                                 }
@@ -292,7 +292,10 @@ public void applyClusterState(ClusterChangedEvent event) {
             RepositoriesMetaData newMetaData = state.getMetaData().custom(RepositoriesMetaData.TYPE);
 
             // Check if repositories got changed
-            if ((oldMetaData == null && newMetaData == null) || (oldMetaData != null && oldMetaData.equals(newMetaData))) {
+            if ((oldMetaData == null && newMetaData == null) || (oldMetaData != null && oldMetaData.equalsIgnoreGenerations(newMetaData))) {
+                for (Repository repo : repositories.values()) {
+                    repo.updateState(state);
+                }
                 return;
             }
 

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

@@ -52,6 +52,12 @@ public final class RepositoryData {
      * The generation value indicating the repository has no index generational files.
      */
     public static final long EMPTY_REPO_GEN = -1L;
+
+    /**
+     * The generation value indicating that the repository generation is unknown.
+     */
+    public static final long UNKNOWN_REPO_GEN = -2L;
+
     /**
      * An instance initialized for an empty repository.
      */