Add custom metadata to snapshots (#41281)

Adds a metadata field to snapshots which can be used to store arbitrary
key-value information. This may be useful for attaching a description of
why a snapshot was taken, tagging snapshots to make categorization
easier, or identifying the source of automatically-created snapshots.

client/rest-high-level/src/test/java/org/elasticsearch/client/SnapshotIT.java

@@ -41,9 +41,12 @@ import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.repositories.fs.FsRepository;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.snapshots.RestoreInfo;
+import org.elasticsearch.snapshots.SnapshotInfo;
 
 import java.io.IOException;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 import static org.hamcrest.Matchers.contains;
@@ -139,6 +142,9 @@ public class SnapshotIT extends ESRestHighLevelClientTestCase {
         CreateSnapshotRequest request = new CreateSnapshotRequest(repository, snapshot);
         boolean waitForCompletion = randomBoolean();
         request.waitForCompletion(waitForCompletion);
+        if (randomBoolean()) {
+            request.userMetadata(randomUserMetadata());
+        }
         request.partial(randomBoolean());
         request.includeGlobalState(randomBoolean());
 
@@ -167,6 +173,8 @@ public class SnapshotIT extends ESRestHighLevelClientTestCase {
         CreateSnapshotResponse putSnapshotResponse1 = createTestSnapshot(createSnapshotRequest1);
         CreateSnapshotRequest createSnapshotRequest2 = new CreateSnapshotRequest(repository, snapshot2);
         createSnapshotRequest2.waitForCompletion(true);
+        Map<String, Object> originalMetadata = randomUserMetadata();
+        createSnapshotRequest2.userMetadata(originalMetadata);
         CreateSnapshotResponse putSnapshotResponse2 = createTestSnapshot(createSnapshotRequest2);
         // check that the request went ok without parsing JSON here. When using the high level client, check acknowledgement instead.
         assertEquals(RestStatus.OK, putSnapshotResponse1.status());
@@ -186,6 +194,12 @@ public class SnapshotIT extends ESRestHighLevelClientTestCase {
         assertEquals(2, response.getSnapshots().size());
         assertThat(response.getSnapshots().stream().map((s) -> s.snapshotId().getName()).collect(Collectors.toList()),
             contains("test_snapshot1", "test_snapshot2"));
+        response.getSnapshots().stream()
+            .filter(s -> s.snapshotId().getName().equals("test_snapshot2"))
+            .findFirst()
+            .map(SnapshotInfo::userMetadata)
+            .ifPresentOrElse(metadata -> assertEquals(originalMetadata, metadata),
+                () -> assertNull("retrieved metadata is null, expected non-null metadata", originalMetadata));
     }
 
     public void testSnapshotsStatus() throws IOException {
@@ -231,6 +245,9 @@ public class SnapshotIT extends ESRestHighLevelClientTestCase {
         CreateSnapshotRequest createSnapshotRequest = new CreateSnapshotRequest(testRepository, testSnapshot);
         createSnapshotRequest.indices(testIndex);
         createSnapshotRequest.waitForCompletion(true);
+        if (randomBoolean()) {
+            createSnapshotRequest.userMetadata(randomUserMetadata());
+        }
         CreateSnapshotResponse createSnapshotResponse = createTestSnapshot(createSnapshotRequest);
         assertEquals(RestStatus.OK, createSnapshotResponse.status());
 
@@ -261,6 +278,9 @@ public class SnapshotIT extends ESRestHighLevelClientTestCase {
 
         CreateSnapshotRequest createSnapshotRequest = new CreateSnapshotRequest(repository, snapshot);
         createSnapshotRequest.waitForCompletion(true);
+        if (randomBoolean()) {
+            createSnapshotRequest.userMetadata(randomUserMetadata());
+        }
         CreateSnapshotResponse createSnapshotResponse = createTestSnapshot(createSnapshotRequest);
         // check that the request went ok without parsing JSON here. When using the high level client, check acknowledgement instead.
         assertEquals(RestStatus.OK, createSnapshotResponse.status());
@@ -270,4 +290,28 @@ public class SnapshotIT extends ESRestHighLevelClientTestCase {
 
         assertTrue(response.isAcknowledged());
     }
+
+    private static Map<String, Object> randomUserMetadata() {
+        if (randomBoolean()) {
+            return null;
+        }
+
+        Map<String, Object> metadata = new HashMap<>();
+        long fields = randomLongBetween(0, 4);
+        for (int i = 0; i < fields; i++) {
+            if (randomBoolean()) {
+                metadata.put(randomValueOtherThanMany(metadata::containsKey, () -> randomAlphaOfLengthBetween(2,10)),
+                    randomAlphaOfLengthBetween(5, 5));
+            } else {
+                Map<String, Object> nested = new HashMap<>();
+                long nestedFields = randomLongBetween(0, 4);
+                for (int j = 0; j < nestedFields; j++) {
+                    nested.put(randomValueOtherThanMany(nested::containsKey, () -> randomAlphaOfLengthBetween(2,10)),
+                        randomAlphaOfLengthBetween(5, 5));
+                }
+                metadata.put(randomValueOtherThanMany(metadata::containsKey, () -> randomAlphaOfLengthBetween(2,10)), nested);
+            }
+        }
+        return metadata;
+    }
 }

docs/reference/modules/snapshots.asciidoc

@@ -349,7 +349,11 @@ PUT /_snapshot/my_backup/snapshot_2?wait_for_completion=true
 {
   "indices": "index_1,index_2",
   "ignore_unavailable": true,
-  "include_global_state": false
+  "include_global_state": false,
+  "_meta": {
+    "taken_by": "kimchy",
+    "taken_because": "backup before upgrading"
+  }
 }
 -----------------------------------
 // CONSOLE
@@ -363,6 +367,9 @@ By setting `include_global_state` to false it's possible to prevent the cluster
 the snapshot. By default, the entire snapshot will fail if one or more indices participating in the snapshot don't have
 all primary shards available. This behaviour can be changed by setting `partial` to `true`.
 
+The `_meta` field can be used to attach arbitrary metadata to the snapshot. This may be a record of who took the snapshot,
+why it was taken, or any other data that might be useful.
+
 Snapshot names can be automatically derived using <<date-math-index-names,date math expressions>>, similarly as when creating
 new indices. Note that special characters need to be URI encoded.
 

rest-api-spec/src/main/resources/rest-api-spec/test/snapshot.get/10_basic.yml

@@ -87,6 +87,7 @@ setup:
   - is_false: snapshots.0.failures
   - is_false: snapshots.0.shards
   - is_false: snapshots.0.version
+  - is_false: snapshots.0._meta
 
   - do:
       snapshot.delete:
@@ -149,3 +150,41 @@ setup:
       snapshot.delete:
         repository: test_repo_get_1
         snapshot: test_snapshot_without_include_global_state
+
+---
+"Get snapshot info with metadata":
+  - skip:
+      version: " - 7.9.99"
+      reason: "https://github.com/elastic/elasticsearch/pull/41281 not yet backported to 7.x"
+
+  - do:
+      indices.create:
+        index: test_index
+        body:
+          settings:
+            number_of_shards:   1
+            number_of_replicas: 0
+
+  - do:
+      snapshot.create:
+        repository: test_repo_get_1
+        snapshot: test_snapshot_with_metadata
+        wait_for_completion: true
+        body: |
+          { "metadata": {"taken_by": "test", "foo": {"bar": "baz"}} }
+
+  - do:
+      snapshot.get:
+        repository: test_repo_get_1
+        snapshot: test_snapshot_with_metadata
+
+  - is_true: snapshots
+  - match: { snapshots.0.snapshot: test_snapshot_with_metadata }
+  - match: { snapshots.0.state: SUCCESS }
+  - match: { snapshots.0.metadata.taken_by: test }
+  - match: { snapshots.0.metadata.foo.bar: baz }
+
+  - do:
+      snapshot.delete:
+        repository: test_repo_get_1
+        snapshot: test_snapshot_with_metadata

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/create/CreateSnapshotRequest.java

@@ -19,12 +19,14 @@
 
 package org.elasticsearch.action.admin.cluster.snapshots.create;
 
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchGenerationException;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.master.MasterNodeRequest;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.Settings;
@@ -46,6 +48,7 @@ import static org.elasticsearch.common.settings.Settings.Builder.EMPTY_SETTINGS;
 import static org.elasticsearch.common.settings.Settings.readSettingsFromStream;
 import static org.elasticsearch.common.settings.Settings.writeSettingsToStream;
 import static org.elasticsearch.common.xcontent.support.XContentMapValues.nodeBooleanValue;
+import static org.elasticsearch.snapshots.SnapshotInfo.METADATA_FIELD_INTRODUCED;
 
 /**
  * Create snapshot request
@@ -63,6 +66,7 @@ import static org.elasticsearch.common.xcontent.support.XContentMapValues.nodeBo
  */
 public class CreateSnapshotRequest extends MasterNodeRequest<CreateSnapshotRequest>
         implements IndicesRequest.Replaceable, ToXContentObject {
+    public static int MAXIMUM_METADATA_BYTES = 1024; // chosen arbitrarily
 
     private String snapshot;
 
@@ -80,6 +84,8 @@ public class CreateSnapshotRequest extends MasterNodeRequest<CreateSnapshotReque
 
     private boolean waitForCompletion;
 
+    private Map<String, Object> userMetadata;
+
     public CreateSnapshotRequest() {
     }
 
@@ -104,6 +110,9 @@ public class CreateSnapshotRequest extends MasterNodeRequest<CreateSnapshotReque
         includeGlobalState = in.readBoolean();
         waitForCompletion = in.readBoolean();
         partial = in.readBoolean();
+        if (in.getVersion().onOrAfter(METADATA_FIELD_INTRODUCED)) {
+            userMetadata = in.readMap();
+        }
     }
 
     @Override
@@ -117,6 +126,9 @@ public class CreateSnapshotRequest extends MasterNodeRequest<CreateSnapshotReque
         out.writeBoolean(includeGlobalState);
         out.writeBoolean(waitForCompletion);
         out.writeBoolean(partial);
+        if (out.getVersion().onOrAfter(METADATA_FIELD_INTRODUCED)) {
+            out.writeMap(userMetadata);
+        }
     }
 
     @Override
@@ -144,9 +156,28 @@ public class CreateSnapshotRequest extends MasterNodeRequest<CreateSnapshotReque
         if (settings == null) {
             validationException = addValidationError("settings is null", validationException);
         }
+        final int metadataSize = metadataSize(userMetadata);
+        if (metadataSize > MAXIMUM_METADATA_BYTES) {
+            validationException = addValidationError("metadata must be smaller than 1024 bytes, but was [" + metadataSize + "]",
+                validationException);
+        }
         return validationException;
     }
 
+    private static int metadataSize(Map<String, Object> userMetadata) {
+        if (userMetadata == null) {
+            return 0;
+        }
+        try (XContentBuilder builder = XContentFactory.jsonBuilder()) {
+            builder.value(userMetadata);
+            int size = BytesReference.bytes(builder).length();
+            return size;
+        } catch (IOException e) {
+            // This should not be possible as we are just rendering the xcontent in memory
+            throw new ElasticsearchException(e);
+        }
+    }
+
     /**
      * Sets the snapshot name
      *
@@ -378,6 +409,15 @@ public class CreateSnapshotRequest extends MasterNodeRequest<CreateSnapshotReque
         return includeGlobalState;
     }
 
+    public Map<String, Object> userMetadata() {
+        return userMetadata;
+    }
+
+    public CreateSnapshotRequest userMetadata(Map<String, Object> userMetadata) {
+        this.userMetadata = userMetadata;
+        return this;
+    }
+
     /**
      * Parses snapshot definition.
      *
@@ -405,6 +445,11 @@ public class CreateSnapshotRequest extends MasterNodeRequest<CreateSnapshotReque
                 settings((Map<String, Object>) entry.getValue());
             } else if (name.equals("include_global_state")) {
                 includeGlobalState = nodeBooleanValue(entry.getValue(), "include_global_state");
+            } else if (name.equals("metadata")) {
+                if (entry.getValue() != null && (entry.getValue() instanceof Map == false)) {
+                    throw new IllegalArgumentException("malformed metadata, should be an object");
+                }
+                userMetadata((Map<String, Object>) entry.getValue());
             }
         }
         indicesOptions(IndicesOptions.fromMap(source, indicesOptions));
@@ -433,6 +478,7 @@ public class CreateSnapshotRequest extends MasterNodeRequest<CreateSnapshotReque
         if (indicesOptions != null) {
             indicesOptions.toXContent(builder, params);
         }
+        builder.field("metadata", userMetadata);
         builder.endObject();
         return builder;
     }
@@ -460,12 +506,14 @@ public class CreateSnapshotRequest extends MasterNodeRequest<CreateSnapshotReque
             Arrays.equals(indices, that.indices) &&
             Objects.equals(indicesOptions, that.indicesOptions) &&
             Objects.equals(settings, that.settings) &&
-            Objects.equals(masterNodeTimeout, that.masterNodeTimeout);
+            Objects.equals(masterNodeTimeout, that.masterNodeTimeout) &&
+            Objects.equals(userMetadata, that.userMetadata);
     }
 
     @Override
     public int hashCode() {
-        int result = Objects.hash(snapshot, repository, indicesOptions, partial, settings, includeGlobalState, waitForCompletion);
+        int result = Objects.hash(snapshot, repository, indicesOptions, partial, settings, includeGlobalState,
+            waitForCompletion, userMetadata);
         result = 31 * result + Arrays.hashCode(indices);
         return result;
     }
@@ -482,6 +530,7 @@ public class CreateSnapshotRequest extends MasterNodeRequest<CreateSnapshotReque
             ", includeGlobalState=" + includeGlobalState +
             ", waitForCompletion=" + waitForCompletion +
             ", masterNodeTimeout=" + masterNodeTimeout +
+            ", metadata=" + userMetadata +
             '}';
     }
 }

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/get/GetSnapshotsResponse.java

@@ -21,6 +21,7 @@ package org.elasticsearch.action.admin.cluster.snapshots.get;
 
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.xcontent.ConstructingObjectParser;
@@ -117,4 +118,9 @@ public class GetSnapshotsResponse extends ActionResponse implements ToXContentOb
     public int hashCode() {
         return Objects.hash(snapshots);
     }
+
+    @Override
+    public String toString() {
+        return Strings.toString(this);
+    }
 }

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

@@ -44,6 +44,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
+import static org.elasticsearch.snapshots.SnapshotInfo.METADATA_FIELD_INTRODUCED;
+
 /**
  * Meta data about snapshots that are currently executing
  */
@@ -84,11 +86,12 @@ public class SnapshotsInProgress extends AbstractNamedDiffable<Custom> implement
         private final ImmutableOpenMap<String, List<ShardId>> waitingIndices;
         private final long startTime;
         private final long repositoryStateId;
+        @Nullable private final Map<String, Object> userMetadata;
         @Nullable private final String failure;
 
         public Entry(Snapshot snapshot, boolean includeGlobalState, boolean partial, State state, List<IndexId> indices,
                      long startTime, long repositoryStateId, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards,
-                     String failure) {
+                     String failure, Map<String, Object> userMetadata) {
             this.state = state;
             this.snapshot = snapshot;
             this.includeGlobalState = includeGlobalState;
@@ -104,21 +107,23 @@ public class SnapshotsInProgress extends AbstractNamedDiffable<Custom> implement
             }
             this.repositoryStateId = repositoryStateId;
             this.failure = failure;
+            this.userMetadata = userMetadata;
         }
 
         public Entry(Snapshot snapshot, boolean includeGlobalState, boolean partial, State state, List<IndexId> indices,
-            long startTime, long repositoryStateId, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards) {
-            this(snapshot, includeGlobalState, partial, state, indices, startTime, repositoryStateId, shards, null);
+                     long startTime, long repositoryStateId, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards,
+                     Map<String, Object> userMetadata) {
+            this(snapshot, includeGlobalState, partial, state, indices, startTime, repositoryStateId, shards, null, userMetadata);
         }
 
         public Entry(Entry entry, State state, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards) {
             this(entry.snapshot, entry.includeGlobalState, entry.partial, state, entry.indices, entry.startTime,
-                entry.repositoryStateId, shards, entry.failure);
+                entry.repositoryStateId, shards, entry.failure, entry.userMetadata);
         }
 
         public Entry(Entry entry, State state, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards, String failure) {
             this(entry.snapshot, entry.includeGlobalState, entry.partial, state, entry.indices, entry.startTime,
-                 entry.repositoryStateId, shards, failure);
+                 entry.repositoryStateId, shards, failure, entry.userMetadata);
         }
 
         public Entry(Entry entry, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards) {
@@ -149,6 +154,10 @@ public class SnapshotsInProgress extends AbstractNamedDiffable<Custom> implement
             return includeGlobalState;
         }
 
+        public Map<String, Object> userMetadata() {
+            return userMetadata;
+        }
+
         public boolean partial() {
             return partial;
         }
@@ -419,6 +428,10 @@ public class SnapshotsInProgress extends AbstractNamedDiffable<Custom> implement
             }
             long repositoryStateId = in.readLong();
             final String failure = in.readOptionalString();
+            Map<String, Object> userMetadata = null;
+            if (in.getVersion().onOrAfter(METADATA_FIELD_INTRODUCED)) {
+                userMetadata = in.readMap();
+            }
             entries[i] = new Entry(snapshot,
                                    includeGlobalState,
                                    partial,
@@ -427,7 +440,9 @@ public class SnapshotsInProgress extends AbstractNamedDiffable<Custom> implement
                                    startTime,
                                    repositoryStateId,
                                    builder.build(),
-                                   failure);
+                                   failure,
+                                   userMetadata
+                );
         }
         this.entries = Arrays.asList(entries);
     }
@@ -452,6 +467,9 @@ public class SnapshotsInProgress extends AbstractNamedDiffable<Custom> implement
             }
             out.writeLong(entry.repositoryStateId);
             out.writeOptionalString(entry.failure);
+            if (out.getVersion().onOrAfter(METADATA_FIELD_INTRODUCED)) {
+                out.writeMap(entry.userMetadata);
+            }
         }
     }
 

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

@@ -38,6 +38,7 @@ import org.elasticsearch.snapshots.SnapshotShardFailure;
 
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
 
 public class FilterRepository implements Repository {
 
@@ -79,9 +80,10 @@ public class FilterRepository implements Repository {
 
     @Override
     public SnapshotInfo finalizeSnapshot(SnapshotId snapshotId, List<IndexId> indices, long startTime, String failure, int totalShards,
-                                         List<SnapshotShardFailure> shardFailures, long repositoryStateId, boolean includeGlobalState) {
+                                         List<SnapshotShardFailure> shardFailures, long repositoryStateId, boolean includeGlobalState,
+                                         Map<String, Object> userMetadata) {
         return in.finalizeSnapshot(snapshotId, indices, startTime, failure, totalShards, shardFailures, repositoryStateId,
-            includeGlobalState);
+            includeGlobalState, userMetadata);
     }
 
     @Override

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

@@ -37,6 +37,7 @@ import org.elasticsearch.snapshots.SnapshotShardFailure;
 
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
 import java.util.function.Function;
 
 /**
@@ -134,7 +135,8 @@ public interface Repository extends LifecycleComponent {
      * @return snapshot description
      */
     SnapshotInfo finalizeSnapshot(SnapshotId snapshotId, List<IndexId> indices, long startTime, String failure, int totalShards,
-                                  List<SnapshotShardFailure> shardFailures, long repositoryStateId, boolean includeGlobalState);
+                                  List<SnapshotShardFailure> shardFailures, long repositoryStateId, boolean includeGlobalState,
+                                  Map<String, Object> userMetadata);
 
     /**
      * Deletes snapshot

server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java

@@ -494,11 +494,12 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp
                                          final int totalShards,
                                          final List<SnapshotShardFailure> shardFailures,
                                          final long repositoryStateId,
-                                         final boolean includeGlobalState) {
+                                         final boolean includeGlobalState,
+                                         final Map<String, Object> userMetadata) {
         SnapshotInfo blobStoreSnapshot = new SnapshotInfo(snapshotId,
             indices.stream().map(IndexId::getName).collect(Collectors.toList()),
             startTime, failure, System.currentTimeMillis(), totalShards, shardFailures,
-            includeGlobalState);
+            includeGlobalState, userMetadata);
         try {
             snapshotFormat.write(blobStoreSnapshot, blobContainer(), snapshotId.getUUID());
             final RepositoryData repositoryData = getRepositoryData();

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

@@ -42,6 +42,7 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 
 /**
@@ -51,6 +52,7 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
 
     public static final String CONTEXT_MODE_PARAM = "context_mode";
     public static final String CONTEXT_MODE_SNAPSHOT = "SNAPSHOT";
+    public static final Version METADATA_FIELD_INTRODUCED = Version.V_8_0_0; // TODO Set this to the earliest version this is backported to
     private static final DateFormatter DATE_TIME_FORMATTER = DateFormatter.forPattern("strictDateOptionalTime");
     private static final String SNAPSHOT = "snapshot";
     private static final String UUID = "uuid";
@@ -74,6 +76,7 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
     private static final String TOTAL_SHARDS = "total_shards";
     private static final String SUCCESSFUL_SHARDS = "successful_shards";
     private static final String INCLUDE_GLOBAL_STATE = "include_global_state";
+    private static final String USER_METADATA = "metadata";
 
     private static final Comparator<SnapshotInfo> COMPARATOR =
         Comparator.comparing(SnapshotInfo::startTime).thenComparing(SnapshotInfo::snapshotId);
@@ -88,6 +91,7 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
         private long endTime = 0L;
         private ShardStatsBuilder shardStatsBuilder = null;
         private Boolean includeGlobalState = null;
+        private Map<String, Object> userMetadata = null;
         private int version = -1;
         private List<SnapshotShardFailure> shardFailures = null;
 
@@ -127,6 +131,10 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
             this.includeGlobalState = includeGlobalState;
         }
 
+        private void setUserMetadata(Map<String, Object> userMetadata) {
+            this.userMetadata = userMetadata;
+        }
+
         private void setVersion(int version) {
             this.version = version;
         }
@@ -153,7 +161,7 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
             }
 
             return new SnapshotInfo(snapshotId, indices, snapshotState, reason, version, startTime, endTime,
-                    totalShards, successfulShards, shardFailures, includeGlobalState);
+                    totalShards, successfulShards, shardFailures, includeGlobalState, userMetadata);
         }
     }
 
@@ -194,6 +202,7 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
         SNAPSHOT_INFO_PARSER.declareLong(SnapshotInfoBuilder::setEndTime, new ParseField(END_TIME_IN_MILLIS));
         SNAPSHOT_INFO_PARSER.declareObject(SnapshotInfoBuilder::setShardStatsBuilder, SHARD_STATS_PARSER, new ParseField(SHARDS));
         SNAPSHOT_INFO_PARSER.declareBoolean(SnapshotInfoBuilder::setIncludeGlobalState, new ParseField(INCLUDE_GLOBAL_STATE));
+        SNAPSHOT_INFO_PARSER.declareObject(SnapshotInfoBuilder::setUserMetadata, (p, c) -> p.map() , new ParseField(USER_METADATA));
         SNAPSHOT_INFO_PARSER.declareInt(SnapshotInfoBuilder::setVersion, new ParseField(VERSION_ID));
         SNAPSHOT_INFO_PARSER.declareObjectArray(SnapshotInfoBuilder::setShardFailures, SnapshotShardFailure.SNAPSHOT_SHARD_FAILURE_PARSER,
             new ParseField(FAILURES));
@@ -223,6 +232,9 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
     @Nullable
     private final Boolean includeGlobalState;
 
+    @Nullable
+    private final Map<String, Object> userMetadata;
+
     @Nullable
     private final Version version;
 
@@ -230,28 +242,30 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
 
     public SnapshotInfo(SnapshotId snapshotId, List<String> indices, SnapshotState state) {
         this(snapshotId, indices, state, null, null, 0L, 0L, 0, 0,
-            Collections.emptyList(), null);
+            Collections.emptyList(), null, null);
     }
 
     public SnapshotInfo(SnapshotId snapshotId, List<String> indices, SnapshotState state, Version version) {
         this(snapshotId, indices, state, null, version, 0L, 0L, 0, 0,
-            Collections.emptyList(), null);
+            Collections.emptyList(), null, null);
     }
 
-    public SnapshotInfo(SnapshotId snapshotId, List<String> indices, long startTime, Boolean includeGlobalState) {
+    public SnapshotInfo(SnapshotId snapshotId, List<String> indices, long startTime, Boolean includeGlobalState,
+                        Map<String, Object> userMetadata) {
         this(snapshotId, indices, SnapshotState.IN_PROGRESS, null, Version.CURRENT, startTime, 0L,
-            0, 0, Collections.emptyList(), includeGlobalState);
+            0, 0, Collections.emptyList(), includeGlobalState, userMetadata);
     }
 
     public SnapshotInfo(SnapshotId snapshotId, List<String> indices, long startTime, String reason, long endTime,
-                        int totalShards, List<SnapshotShardFailure> shardFailures, Boolean includeGlobalState) {
+                        int totalShards, List<SnapshotShardFailure> shardFailures, Boolean includeGlobalState,
+                        Map<String, Object> userMetadata) {
         this(snapshotId, indices, snapshotState(reason, shardFailures), reason, Version.CURRENT,
-             startTime, endTime, totalShards, totalShards - shardFailures.size(), shardFailures, includeGlobalState);
+             startTime, endTime, totalShards, totalShards - shardFailures.size(), shardFailures, includeGlobalState, userMetadata);
     }
 
     private SnapshotInfo(SnapshotId snapshotId, List<String> indices, SnapshotState state, String reason, Version version,
                          long startTime, long endTime, int totalShards, int successfulShards, List<SnapshotShardFailure> shardFailures,
-                         Boolean includeGlobalState) {
+                         Boolean includeGlobalState, Map<String, Object> userMetadata) {
         this.snapshotId = Objects.requireNonNull(snapshotId);
         this.indices = Collections.unmodifiableList(Objects.requireNonNull(indices));
         this.state = state;
@@ -263,6 +277,7 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
         this.successfulShards = successfulShards;
         this.shardFailures = Objects.requireNonNull(shardFailures);
         this.includeGlobalState = includeGlobalState;
+        this.userMetadata = userMetadata;
     }
 
     /**
@@ -294,6 +309,11 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
         }
         version = in.readBoolean() ? Version.readVersion(in) : null;
         includeGlobalState = in.readOptionalBoolean();
+        if (in.getVersion().onOrAfter(METADATA_FIELD_INTRODUCED)) {
+            userMetadata = in.readMap();
+        } else {
+            userMetadata = null;
+        }
     }
 
     /**
@@ -304,7 +324,7 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
         return new SnapshotInfo(snapshotId, Collections.emptyList(), SnapshotState.INCOMPATIBLE,
                                 "the snapshot is incompatible with the current version of Elasticsearch and its exact version is unknown",
                                 null, 0L, 0L, 0, 0,
-                                Collections.emptyList(), null);
+                                Collections.emptyList(), null, null);
     }
 
     /**
@@ -428,6 +448,15 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
         return version;
     }
 
+    /**
+     * Returns the custom metadata that was attached to this snapshot at creation time.
+     * @return custom metadata
+     */
+    @Nullable
+    public Map<String, Object> userMetadata() {
+        return userMetadata;
+    }
+
     /**
      * Compares two snapshots by their start time; if the start times are the same, then
      * compares the two snapshots by their snapshot ids.
@@ -492,6 +521,9 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
         if (includeGlobalState != null) {
             builder.field(INCLUDE_GLOBAL_STATE, includeGlobalState);
         }
+        if (userMetadata != null) {
+            builder.field(USER_METADATA, userMetadata);
+        }
         if (verbose || state != null) {
             builder.field(STATE, state);
         }
@@ -543,6 +575,7 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
         if (includeGlobalState != null) {
             builder.field(INCLUDE_GLOBAL_STATE, includeGlobalState);
         }
+        builder.field(USER_METADATA, userMetadata);
         builder.field(START_TIME, startTime);
         builder.field(END_TIME, endTime);
         builder.field(TOTAL_SHARDS, totalShards);
@@ -573,6 +606,7 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
         int totalShards = 0;
         int successfulShards = 0;
         Boolean includeGlobalState = null;
+        Map<String, Object> userMetadata = null;
         List<SnapshotShardFailure> shardFailures = Collections.emptyList();
         if (parser.currentToken() == null) { // fresh parser? move to the first token
             parser.nextToken();
@@ -628,8 +662,12 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
                                 parser.skipChildren();
                             }
                         } else if (token == XContentParser.Token.START_OBJECT) {
-                            // It was probably created by newer version - ignoring
-                            parser.skipChildren();
+                            if (USER_METADATA.equals(currentFieldName)) {
+                                userMetadata = parser.map();
+                            } else {
+                                // It was probably created by newer version - ignoring
+                                parser.skipChildren();
+                            }
                         }
                     }
                 }
@@ -651,7 +689,8 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
                                 totalShards,
                                 successfulShards,
                                 shardFailures,
-                                includeGlobalState);
+                                includeGlobalState,
+                                userMetadata);
     }
 
     @Override
@@ -683,6 +722,9 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
             out.writeBoolean(false);
         }
         out.writeOptionalBoolean(includeGlobalState);
+        if (out.getVersion().onOrAfter(METADATA_FIELD_INTRODUCED)) {
+            out.writeMap(userMetadata);
+        }
     }
 
     private static SnapshotState snapshotState(final String reason, final List<SnapshotShardFailure> shardFailures) {
@@ -712,13 +754,14 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
             Objects.equals(indices, that.indices) &&
             Objects.equals(includeGlobalState, that.includeGlobalState) &&
             Objects.equals(version, that.version) &&
-            Objects.equals(shardFailures, that.shardFailures);
+            Objects.equals(shardFailures, that.shardFailures) &&
+            Objects.equals(userMetadata, that.userMetadata);
     }
 
     @Override
     public int hashCode() {
 
         return Objects.hash(snapshotId, state, reason, indices, startTime, endTime,
-                totalShards, successfulShards, includeGlobalState, version, shardFailures);
+                totalShards, successfulShards, includeGlobalState, version, shardFailures, userMetadata);
     }
 }

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

@@ -287,7 +287,8 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
                                                                 snapshotIndices,
                                                                 System.currentTimeMillis(),
                                                                 repositoryData.getGenId(),
-                                                                null);
+                                                                null,
+                                                                request.userMetadata());
                     initializingSnapshots.add(newSnapshot.snapshot());
                     snapshots = new SnapshotsInProgress(newSnapshot);
                 } else {
@@ -557,7 +558,8 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
                                                          0,
                                                          Collections.emptyList(),
                                                          snapshot.getRepositoryStateId(),
-                                                         snapshot.includeGlobalState());
+                                                         snapshot.includeGlobalState(),
+                                                         snapshot.userMetadata());
                 } catch (Exception inner) {
                     inner.addSuppressed(exception);
                     logger.warn(() -> new ParameterizedMessage("[{}] failed to close snapshot in repository",
@@ -572,7 +574,7 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
     private static SnapshotInfo inProgressSnapshot(SnapshotsInProgress.Entry entry) {
         return new SnapshotInfo(entry.snapshot().getSnapshotId(),
                                    entry.indices().stream().map(IndexId::getName).collect(Collectors.toList()),
-                                   entry.startTime(), entry.includeGlobalState());
+                                   entry.startTime(), entry.includeGlobalState(), entry.userMetadata());
     }
 
     /**
@@ -988,7 +990,8 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
                     entry.shards().size(),
                     unmodifiableList(shardFailures),
                     entry.getRepositoryStateId(),
-                    entry.includeGlobalState());
+                    entry.includeGlobalState(),
+                    entry.userMetadata());
                 removeSnapshotFromClusterState(snapshot, snapshotInfo, null);
                 logger.info("snapshot [{}] completed with state [{}]", snapshot, snapshotInfo.state());
             }

server/src/test/java/org/elasticsearch/action/admin/cluster/snapshots/create/CreateSnapshotRequestTests.java

@@ -19,10 +19,12 @@
 
 package org.elasticsearch.action.admin.cluster.snapshots.create;
 
+import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.action.support.IndicesOptions.Option;
 import org.elasticsearch.action.support.IndicesOptions.WildcardStates;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.ToXContent.MapParams;
 import org.elasticsearch.common.xcontent.XContentBuilder;
@@ -41,6 +43,10 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.elasticsearch.snapshots.SnapshotInfoTests.randomUserMetadata;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
+
 public class CreateSnapshotRequestTests extends ESTestCase {
 
     // tests creating XContent and parsing with source(Map) equivalency
@@ -80,6 +86,10 @@ public class CreateSnapshotRequestTests extends ESTestCase {
             original.includeGlobalState(randomBoolean());
         }
 
+        if (randomBoolean()) {
+            original.userMetadata(randomUserMetadata());
+        }
+
         if (randomBoolean()) {
             Collection<WildcardStates> wildcardStates = randomSubsetOf(Arrays.asList(WildcardStates.values()));
             Collection<Option> options = randomSubsetOf(Arrays.asList(Option.ALLOW_NO_INDICES, Option.IGNORE_UNAVAILABLE));
@@ -108,4 +118,57 @@ public class CreateSnapshotRequestTests extends ESTestCase {
 
         assertEquals(original, processed);
     }
+
+    public void testSizeCheck() {
+        {
+            Map<String, Object> simple = new HashMap<>();
+            simple.put(randomAlphaOfLength(5), randomAlphaOfLength(25));
+            assertNull(createSnapshotRequestWithMetadata(simple).validate());
+        }
+
+        {
+            Map<String, Object> complex = new HashMap<>();
+            Map<String, Object> nested = new HashMap<>();
+            nested.put(randomAlphaOfLength(5), randomAlphaOfLength(5));
+            nested.put(randomAlphaOfLength(6), randomAlphaOfLength(5));
+            complex.put(randomAlphaOfLength(7), nested);
+            assertNull(createSnapshotRequestWithMetadata(complex).validate());
+        }
+
+        {
+            Map<String, Object> barelyFine = new HashMap<>();
+            barelyFine.put(randomAlphaOfLength(512), randomAlphaOfLength(505));
+            assertNull(createSnapshotRequestWithMetadata(barelyFine).validate());
+        }
+
+        {
+            Map<String, Object> barelyTooBig = new HashMap<>();
+            barelyTooBig.put(randomAlphaOfLength(512), randomAlphaOfLength(506));
+            ActionRequestValidationException validationException = createSnapshotRequestWithMetadata(barelyTooBig).validate();
+            assertNotNull(validationException);
+            assertThat(validationException.validationErrors(), hasSize(1));
+            assertThat(validationException.validationErrors().get(0), equalTo("metadata must be smaller than 1024 bytes, but was [1025]"));
+        }
+
+        {
+            Map<String, Object> tooBigOnlyIfNestedFieldsAreIncluded = new HashMap<>();
+            HashMap<Object, Object> nested = new HashMap<>();
+            nested.put(randomAlphaOfLength(500), randomAlphaOfLength(500));
+            tooBigOnlyIfNestedFieldsAreIncluded.put(randomAlphaOfLength(10), randomAlphaOfLength(10));
+            tooBigOnlyIfNestedFieldsAreIncluded.put(randomAlphaOfLength(11), nested);
+
+            ActionRequestValidationException validationException = createSnapshotRequestWithMetadata(tooBigOnlyIfNestedFieldsAreIncluded)
+                .validate();
+            assertNotNull(validationException);
+            assertThat(validationException.validationErrors(), hasSize(1));
+            assertThat(validationException.validationErrors().get(0), equalTo("metadata must be smaller than 1024 bytes, but was [1049]"));
+        }
+    }
+
+    private CreateSnapshotRequest createSnapshotRequestWithMetadata(Map<String, Object> metadata) {
+        return new CreateSnapshotRequest(randomAlphaOfLength(5), randomAlphaOfLength(5))
+            .indices(randomAlphaOfLength(5))
+            .settings(Settings.EMPTY)
+            .userMetadata(metadata);
+    }
 }

server/src/test/java/org/elasticsearch/action/admin/cluster/snapshots/create/CreateSnapshotResponseTests.java

@@ -23,6 +23,7 @@ import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.snapshots.SnapshotInfo;
+import org.elasticsearch.snapshots.SnapshotInfoTests;
 import org.elasticsearch.snapshots.SnapshotShardFailure;
 import org.elasticsearch.test.AbstractXContentTestCase;
 
@@ -30,6 +31,7 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
+import java.util.function.Predicate;
 
 public class CreateSnapshotResponseTests extends AbstractXContentTestCase<CreateSnapshotResponse> {
 
@@ -64,6 +66,17 @@ public class CreateSnapshotResponseTests extends AbstractXContentTestCase<Create
         boolean globalState = randomBoolean();
 
         return new CreateSnapshotResponse(
-            new SnapshotInfo(snapshotId, indices, startTime, reason, endTime, totalShards, shardFailures, globalState));
+            new SnapshotInfo(snapshotId, indices, startTime, reason, endTime, totalShards, shardFailures,
+                globalState, SnapshotInfoTests.randomUserMetadata()));
+    }
+
+    @Override
+    protected Predicate<String> getRandomFieldsExcludeFilter() {
+        // Don't inject random fields into the custom snapshot metadata, because the metadata map is equality-checked after doing a
+        // round-trip through xContent serialization/deserialization. Even though the rest of the object ignores unknown fields,
+        // `metadata` doesn't ignore unknown fields (it just includes them in the parsed object, because the keys are arbitrary), so any
+        // new fields added to the the metadata before it gets deserialized that weren't in the serialized version will cause the equality
+        // check to fail.
+        return field -> field.startsWith("snapshot.metadata");
     }
 }

server/src/test/java/org/elasticsearch/action/admin/cluster/snapshots/get/GetSnapshotsResponseTests.java

@@ -24,6 +24,7 @@ import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.snapshots.SnapshotInfo;
+import org.elasticsearch.snapshots.SnapshotInfoTests;
 import org.elasticsearch.snapshots.SnapshotShardFailure;
 import org.elasticsearch.test.AbstractStreamableXContentTestCase;
 
@@ -32,6 +33,8 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.function.Predicate;
+import java.util.regex.Pattern;
 
 public class GetSnapshotsResponseTests extends AbstractStreamableXContentTestCase<GetSnapshotsResponse> {
 
@@ -54,9 +57,21 @@ public class GetSnapshotsResponseTests extends AbstractStreamableXContentTestCas
             ShardId shardId = new ShardId("index", UUIDs.base64UUID(), 2);
             List<SnapshotShardFailure> shardFailures = Collections.singletonList(new SnapshotShardFailure("node-id", shardId, "reason"));
             snapshots.add(new SnapshotInfo(snapshotId, Arrays.asList("indice1", "indice2"), System.currentTimeMillis(), reason,
-                System.currentTimeMillis(), randomIntBetween(2, 3), shardFailures, randomBoolean()));
-
+                System.currentTimeMillis(), randomIntBetween(2, 3), shardFailures, randomBoolean(),
+                SnapshotInfoTests.randomUserMetadata()));
         }
         return new GetSnapshotsResponse(snapshots);
     }
+
+    @Override
+    protected Predicate<String> getRandomFieldsExcludeFilter() {
+        // Don't inject random fields into the custom snapshot metadata, because the metadata map is equality-checked after doing a
+        // round-trip through xContent serialization/deserialization. Even though the rest of the object ignores unknown fields,
+        // `metadata` doesn't ignore unknown fields (it just includes them in the parsed object, because the keys are arbitrary), so any
+        // new fields added to the the metadata before it gets deserialized that weren't in the serialized version will cause the equality
+        // check to fail.
+
+        // The actual fields are nested in an array, so this regex matches fields with names of the form `snapshots.3.metadata`
+        return Pattern.compile("snapshots\\.\\d+\\.metadata.*").asMatchPredicate();
+    }
 }

server/src/test/java/org/elasticsearch/cluster/ClusterStateDiffIT.java

@@ -32,11 +32,6 @@ import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.metadata.IndexTemplateMetaData;
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.metadata.RepositoriesMetaData;
-import org.elasticsearch.common.bytes.BytesReference;
-import org.elasticsearch.common.io.stream.NamedWriteableAwareStreamInput;
-import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
-import org.elasticsearch.common.util.set.Sets;
-import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.IndexRoutingTable;
@@ -47,15 +42,21 @@ import org.elasticsearch.cluster.routing.ShardRoutingState;
 import org.elasticsearch.cluster.routing.TestShardRouting;
 import org.elasticsearch.cluster.routing.UnassignedInfo;
 import org.elasticsearch.common.UUIDs;
+import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.common.io.stream.NamedWriteableAwareStreamInput;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.gateway.GatewayService;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.snapshots.Snapshot;
+import org.elasticsearch.snapshots.SnapshotId;
+import org.elasticsearch.snapshots.SnapshotInfoTests;
 import org.elasticsearch.test.ESIntegTestCase;
 
 import java.util.Collections;
@@ -719,7 +720,8 @@ public class ClusterStateDiffIT extends ESIntegTestCase {
                                 Collections.emptyList(),
                                 Math.abs(randomLong()),
                                 (long) randomIntBetween(0, 1000),
-                                ImmutableOpenMap.of()));
+                                ImmutableOpenMap.of(),
+                                SnapshotInfoTests.randomUserMetadata()));
                     case 1:
                         return new RestoreInProgress.Builder().add(
                             new RestoreInProgress.Entry(

server/src/test/java/org/elasticsearch/cluster/SnapshotsInProgressTests.java

@@ -28,6 +28,7 @@ import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.repositories.IndexId;
 import org.elasticsearch.snapshots.Snapshot;
 import org.elasticsearch.snapshots.SnapshotId;
+import org.elasticsearch.snapshots.SnapshotInfoTests;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.Arrays;
@@ -65,7 +66,7 @@ public class SnapshotsInProgressTests extends ESTestCase {
         // test no waiting shards in an index
         shards.put(new ShardId(idx3Name, idx3UUID, 0), new ShardSnapshotStatus(randomAlphaOfLength(2), randomNonWaitingState(), ""));
         Entry entry = new Entry(snapshot, randomBoolean(), randomBoolean(), State.INIT,
-                                indices, System.currentTimeMillis(), randomLong(), shards.build());
+                                indices, System.currentTimeMillis(), randomLong(), shards.build(), SnapshotInfoTests.randomUserMetadata());
 
         ImmutableOpenMap<String, List<ShardId>> waitingIndices = entry.waitingIndices();
         assertEquals(2, waitingIndices.get(idx1Name).size());

server/src/test/java/org/elasticsearch/cluster/metadata/MetaDataDeleteIndexServiceTests.java

@@ -33,6 +33,7 @@ import org.elasticsearch.repositories.IndexId;
 import org.elasticsearch.snapshots.Snapshot;
 import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.snapshots.SnapshotInProgressException;
+import org.elasticsearch.snapshots.SnapshotInfoTests;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.VersionUtils;
 
@@ -60,7 +61,8 @@ public class MetaDataDeleteIndexServiceTests extends ESTestCase {
         Snapshot snapshot = new Snapshot("doesn't matter", new SnapshotId("snapshot name", "snapshot uuid"));
         SnapshotsInProgress snaps = new SnapshotsInProgress(new SnapshotsInProgress.Entry(snapshot, true, false,
                 SnapshotsInProgress.State.INIT, singletonList(new IndexId(index, "doesn't matter")),
-                System.currentTimeMillis(), (long) randomIntBetween(0, 1000), ImmutableOpenMap.of()));
+                System.currentTimeMillis(), (long) randomIntBetween(0, 1000), ImmutableOpenMap.of(),
+                SnapshotInfoTests.randomUserMetadata()));
         ClusterState state = ClusterState.builder(clusterState(index))
                 .putCustom(SnapshotsInProgress.TYPE, snaps)
                 .build();

server/src/test/java/org/elasticsearch/cluster/metadata/MetaDataIndexStateServiceTests.java

@@ -48,6 +48,7 @@ import org.elasticsearch.repositories.IndexId;
 import org.elasticsearch.snapshots.Snapshot;
 import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.snapshots.SnapshotInProgressException;
+import org.elasticsearch.snapshots.SnapshotInfoTests;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.Arrays;
@@ -432,7 +433,8 @@ public class MetaDataIndexStateServiceTests extends ESTestCase {
         final Snapshot snapshot = new Snapshot(randomAlphaOfLength(10), new SnapshotId(randomAlphaOfLength(5), randomAlphaOfLength(5)));
         final SnapshotsInProgress.Entry entry =
             new SnapshotsInProgress.Entry(snapshot, randomBoolean(), false, SnapshotsInProgress.State.INIT,
-                Collections.singletonList(new IndexId(index, index)), randomNonNegativeLong(), randomLong(), shardsBuilder.build());
+                Collections.singletonList(new IndexId(index, index)), randomNonNegativeLong(), randomLong(), shardsBuilder.build(),
+                SnapshotInfoTests.randomUserMetadata());
         return ClusterState.builder(newState).putCustom(SnapshotsInProgress.TYPE, new SnapshotsInProgress(entry)).build();
     }
 

server/src/test/java/org/elasticsearch/repositories/RepositoriesServiceTests.java

@@ -49,6 +49,7 @@ import org.elasticsearch.transport.TransportService;
 import java.io.IOException;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 import static org.mockito.Mockito.mock;
 
@@ -160,7 +161,7 @@ public class RepositoriesServiceTests extends ESTestCase {
         @Override
         public SnapshotInfo finalizeSnapshot(SnapshotId snapshotId, List<IndexId> indices, long startTime, String failure,
                                              int totalShards, List<SnapshotShardFailure> shardFailures, long repositoryStateId,
-                                             boolean includeGlobalState) {
+                                             boolean includeGlobalState, Map<String, Object> userMetadata) {
             return null;
         }
 

server/src/test/java/org/elasticsearch/snapshots/SharedClusterSnapshotRestoreIT.java

@@ -2714,7 +2714,8 @@ public class SharedClusterSnapshotRestoreIT extends AbstractSnapshotIntegTestCas
                                 Collections.singletonList(indexId),
                                 System.currentTimeMillis(),
                                 repositoryData.getGenId(),
-                                shards.build()))))
+                                shards.build(),
+                                SnapshotInfoTests.randomUserMetadata()))))
                         .build();
             }
 

server/src/test/java/org/elasticsearch/snapshots/SnapshotInfoTests.java

@@ -0,0 +1,149 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.snapshots;
+
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class SnapshotInfoTests extends AbstractWireSerializingTestCase<SnapshotInfo> {
+
+    @Override
+    protected SnapshotInfo createTestInstance() {
+        SnapshotId snapshotId = new SnapshotId(randomAlphaOfLength(5), randomAlphaOfLength(5));
+        List<String> indices = Arrays.asList(randomArray(1, 10, String[]::new, () -> randomAlphaOfLengthBetween(2, 20)));
+
+        String reason = randomBoolean() ? null : randomAlphaOfLengthBetween(5, 15);
+
+        long startTime = randomNonNegativeLong();
+        long endTime = randomNonNegativeLong();
+
+        int totalShards = randomIntBetween(0, 100);
+        int failedShards = randomIntBetween(0, totalShards);
+
+        List<SnapshotShardFailure> shardFailures = Arrays.asList(randomArray(failedShards, failedShards,
+            SnapshotShardFailure[]::new, () -> {
+                String indexName = randomAlphaOfLengthBetween(3, 50);
+                int id = randomInt();
+                ShardId shardId = ShardId.fromString("[" + indexName + "][" + id + "]");
+
+                return new SnapshotShardFailure(randomAlphaOfLengthBetween(5, 10), shardId, randomAlphaOfLengthBetween(5, 10));
+            }));
+
+        Boolean includeGlobalState = randomBoolean() ? null : randomBoolean();
+
+        Map<String, Object> userMetadata = randomUserMetadata();
+
+        return new SnapshotInfo(snapshotId, indices, startTime, reason, endTime, totalShards, shardFailures,
+            includeGlobalState, userMetadata);
+    }
+
+    @Override
+    protected Writeable.Reader<SnapshotInfo> instanceReader() {
+        return SnapshotInfo::new;
+    }
+
+    @Override
+    protected SnapshotInfo mutateInstance(SnapshotInfo instance) {
+        switch (randomIntBetween(0, 7)) {
+            case 0:
+                SnapshotId snapshotId = new SnapshotId(
+                    randomValueOtherThan(instance.snapshotId().getName(), () -> randomAlphaOfLength(5)),
+                    randomValueOtherThan(instance.snapshotId().getUUID(), () -> randomAlphaOfLength(5)));
+                return new SnapshotInfo(snapshotId, instance.indices(), instance.startTime(), instance.reason(),
+                    instance.endTime(), instance.totalShards(), instance.shardFailures(), instance.includeGlobalState(),
+                    instance.userMetadata());
+            case 1:
+                int indicesSize = randomValueOtherThan(instance.indices().size(), () -> randomIntBetween(1, 10));
+                List<String> indices = Arrays.asList(randomArray(indicesSize, indicesSize, String[]::new,
+                    () -> randomAlphaOfLengthBetween(2, 20)));
+                return new SnapshotInfo(instance.snapshotId(), indices, instance.startTime(), instance.reason(),
+                    instance.endTime(), instance.totalShards(), instance.shardFailures(), instance.includeGlobalState(),
+                    instance.userMetadata());
+            case 2:
+                return new SnapshotInfo(instance.snapshotId(), instance.indices(),
+                    randomValueOtherThan(instance.startTime(), ESTestCase::randomNonNegativeLong), instance.reason(),
+                    instance.endTime(), instance.totalShards(), instance.shardFailures(), instance.includeGlobalState(),
+                    instance.userMetadata());
+            case 3:
+                return new SnapshotInfo(instance.snapshotId(), instance.indices(), instance.startTime(),
+                    randomValueOtherThan(instance.reason(), () -> randomAlphaOfLengthBetween(5, 15)), instance.endTime(),
+                    instance.totalShards(), instance.shardFailures(), instance.includeGlobalState(), instance.userMetadata());
+            case 4:
+                return new SnapshotInfo(instance.snapshotId(), instance.indices(), instance.startTime(), instance.reason(),
+                    randomValueOtherThan(instance.endTime(), ESTestCase::randomNonNegativeLong), instance.totalShards(),
+                    instance.shardFailures(), instance.includeGlobalState(), instance.userMetadata());
+            case 5:
+                int totalShards = randomValueOtherThan(instance.totalShards(), () -> randomIntBetween(0, 100));
+                int failedShards = randomIntBetween(0, totalShards);
+
+                List<SnapshotShardFailure> shardFailures = Arrays.asList(randomArray(failedShards, failedShards,
+                    SnapshotShardFailure[]::new, () -> {
+                        String indexName = randomAlphaOfLengthBetween(3, 50);
+                        int id = randomInt();
+                        ShardId shardId = ShardId.fromString("[" + indexName + "][" + id + "]");
+
+                        return new SnapshotShardFailure(randomAlphaOfLengthBetween(5, 10), shardId, randomAlphaOfLengthBetween(5, 10));
+                    }));
+                return new SnapshotInfo(instance.snapshotId(), instance.indices(), instance.startTime(), instance.reason(),
+                    instance.endTime(), totalShards, shardFailures, instance.includeGlobalState(), instance.userMetadata());
+            case 6:
+                return new SnapshotInfo(instance.snapshotId(), instance.indices(), instance.startTime(), instance.reason(),
+                    instance.endTime(), instance.totalShards(), instance.shardFailures(),
+                    Boolean.FALSE.equals(instance.includeGlobalState()), instance.userMetadata());
+            case 7:
+                return new SnapshotInfo(instance.snapshotId(), instance.indices(), instance.startTime(), instance.reason(),
+                    instance.endTime(), instance.totalShards(), instance.shardFailures(), instance.includeGlobalState(),
+                    randomValueOtherThan(instance.userMetadata(), SnapshotInfoTests::randomUserMetadata));
+            default:
+                throw new IllegalArgumentException("invalid randomization case");
+        }
+    }
+
+    public static Map<String, Object> randomUserMetadata() {
+        if (randomBoolean()) {
+            return null;
+        }
+
+        Map<String, Object> metadata = new HashMap<>();
+        long fields = randomLongBetween(0, 25);
+        for (int i = 0; i < fields; i++) {
+            if (randomBoolean()) {
+                metadata.put(randomValueOtherThanMany(metadata::containsKey, () -> randomAlphaOfLengthBetween(2,10)),
+                    randomAlphaOfLengthBetween(5, 15));
+            } else {
+                Map<String, Object> nested = new HashMap<>();
+                long nestedFields = randomLongBetween(0, 25);
+                for (int j = 0; j < nestedFields; j++) {
+                    nested.put(randomValueOtherThanMany(nested::containsKey, () -> randomAlphaOfLengthBetween(2,10)),
+                        randomAlphaOfLengthBetween(5, 15));
+                }
+                metadata.put(randomValueOtherThanMany(metadata::containsKey, () -> randomAlphaOfLengthBetween(2,10)), nested);
+            }
+        }
+        return metadata;
+    }
+}

server/src/test/java/org/elasticsearch/snapshots/SnapshotsInProgressSerializationTests.java

@@ -71,7 +71,8 @@ public class SnapshotsInProgressSerializationTests extends AbstractDiffableWireS
                 shardState.failed() ? randomAlphaOfLength(10) : null));
         }
         ImmutableOpenMap<ShardId, SnapshotsInProgress.ShardSnapshotStatus> shards = builder.build();
-        return new Entry(snapshot, includeGlobalState, partial, state, indices, startTime, repositoryStateId, shards);
+        return new Entry(snapshot, includeGlobalState, partial, state, indices, startTime, repositoryStateId, shards,
+            SnapshotInfoTests.randomUserMetadata());
     }
 
     @Override

test/framework/src/main/java/org/elasticsearch/index/shard/RestoreOnlyRepository.java

@@ -100,7 +100,7 @@ public abstract class RestoreOnlyRepository extends AbstractLifecycleComponent i
     @Override
     public SnapshotInfo finalizeSnapshot(SnapshotId snapshotId, List<IndexId> indices, long startTime, String failure,
                                          int totalShards, List<SnapshotShardFailure> shardFailures, long repositoryStateId,
-                                         boolean includeGlobalState) {
+                                         boolean includeGlobalState, Map<String, Object> userMetadata) {
         return null;
     }
 

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/repository/CcrRepository.java

@@ -255,7 +255,8 @@ public class CcrRepository extends AbstractLifecycleComponent implements Reposit
 
     @Override
     public SnapshotInfo finalizeSnapshot(SnapshotId snapshotId, List<IndexId> indices, long startTime, String failure, int totalShards,
-                                         List<SnapshotShardFailure> shardFailures, long repositoryStateId, boolean includeGlobalState) {
+                                         List<SnapshotShardFailure> shardFailures, long repositoryStateId, boolean includeGlobalState,
+                                         Map<String, Object> userMetadata) {
         throw new UnsupportedOperationException("Unsupported for repository of type: " + TYPE);
     }