3535import org .elasticsearch .common .xcontent .XContentBuilder ;
3636import org .elasticsearch .index .shard .ShardId ;
3737import org .elasticsearch .repositories .IndexId ;
38+ import org .elasticsearch .repositories .RepositoryShardId ;
3839import org .elasticsearch .repositories .RepositoryOperation ;
3940import org .elasticsearch .snapshots .Snapshot ;
41+ import org .elasticsearch .snapshots .SnapshotId ;
42+ import org .elasticsearch .snapshots .SnapshotsService ;
4043
4144import java .io .IOException ;
4245import java .util .Collections ;
@@ -96,18 +99,52 @@ public static Entry startedEntry(Snapshot snapshot, boolean includeGlobalState,
9699 indices , dataStreams , startTime , repositoryStateId , shards , null , userMetadata , version );
97100 }
98101
102+ /**
103+ * Creates the initial snapshot clone entry
104+ *
105+ * @param snapshot snapshot to clone into
106+ * @param source snapshot to clone from
107+ * @param indices indices to clone
108+ * @param startTime start time
109+ * @param repositoryStateId repository state id that this clone is based on
110+ * @param version repository metadata version to write
111+ * @return snapshot clone entry
112+ */
113+ public static Entry startClone (Snapshot snapshot , SnapshotId source , List <IndexId > indices , long startTime ,
114+ long repositoryStateId , Version version ) {
115+ return new SnapshotsInProgress .Entry (snapshot , true , false , State .STARTED , indices , Collections .emptyList (),
116+ startTime , repositoryStateId , ImmutableOpenMap .of (), null , Collections .emptyMap (), version , source ,
117+ ImmutableOpenMap .of ());
118+ }
119+
99120 public static class Entry implements Writeable , ToXContent , RepositoryOperation {
100121 private final State state ;
101122 private final Snapshot snapshot ;
102123 private final boolean includeGlobalState ;
103124 private final boolean partial ;
125+ /**
126+ * Map of {@link ShardId} to {@link ShardSnapshotStatus} tracking the state of each shard snapshot operation.
127+ */
104128 private final ImmutableOpenMap <ShardId , ShardSnapshotStatus > shards ;
105129 private final List <IndexId > indices ;
106130 private final List <String > dataStreams ;
107131 private final long startTime ;
108132 private final long repositoryStateId ;
109133 // see #useShardGenerations
110134 private final Version version ;
135+
136+ /**
137+ * Source snapshot if this is a clone operation or {@code null} if this is a snapshot.
138+ */
139+ @ Nullable
140+ private final SnapshotId source ;
141+
142+ /**
143+ * Map of {@link RepositoryShardId} to {@link ShardSnapshotStatus} tracking the state of each shard clone operation in this entry
144+ * the same way {@link #shards} tracks the status of each shard snapshot operation in non-clone entries.
145+ */
146+ private final ImmutableOpenMap <RepositoryShardId , ShardSnapshotStatus > clones ;
147+
111148 @ Nullable private final Map <String , Object > userMetadata ;
112149 @ Nullable private final String failure ;
113150
@@ -116,6 +153,15 @@ public Entry(Snapshot snapshot, boolean includeGlobalState, boolean partial, Sta
116153 List <String > dataStreams , long startTime , long repositoryStateId ,
117154 ImmutableOpenMap <ShardId , ShardSnapshotStatus > shards , String failure , Map <String , Object > userMetadata ,
118155 Version version ) {
156+ this (snapshot , includeGlobalState , partial , state , indices , dataStreams , startTime , repositoryStateId , shards , failure ,
157+ userMetadata , version , null , ImmutableOpenMap .of ());
158+ }
159+
160+ private Entry (Snapshot snapshot , boolean includeGlobalState , boolean partial , State state , List <IndexId > indices ,
161+ List <String > dataStreams , long startTime , long repositoryStateId ,
162+ ImmutableOpenMap <ShardId , ShardSnapshotStatus > shards , String failure , Map <String , Object > userMetadata ,
163+ Version version , @ Nullable SnapshotId source ,
164+ @ Nullable ImmutableOpenMap <RepositoryShardId , ShardSnapshotStatus > clones ) {
119165 this .state = state ;
120166 this .snapshot = snapshot ;
121167 this .includeGlobalState = includeGlobalState ;
@@ -124,11 +170,18 @@ public Entry(Snapshot snapshot, boolean includeGlobalState, boolean partial, Sta
124170 this .dataStreams = dataStreams ;
125171 this .startTime = startTime ;
126172 this .shards = shards ;
127- assert assertShardsConsistent (state , indices , shards );
128173 this .repositoryStateId = repositoryStateId ;
129174 this .failure = failure ;
130175 this .userMetadata = userMetadata ;
131176 this .version = version ;
177+ this .source = source ;
178+ if (source == null ) {
179+ assert clones == null || clones .isEmpty () : "Provided [" + clones + "] but no source" ;
180+ this .clones = ImmutableOpenMap .of ();
181+ } else {
182+ this .clones = clones ;
183+ }
184+ assert assertShardsConsistent (this .source , this .state , this .indices , this .shards , this .clones );
132185 }
133186
134187 private Entry (StreamInput in ) throws IOException {
@@ -144,29 +197,59 @@ private Entry(StreamInput in) throws IOException {
144197 userMetadata = in .readMap ();
145198 version = Version .readVersion (in );
146199 dataStreams = in .readStringList ();
200+ if (in .getVersion ().onOrAfter (SnapshotsService .CLONE_SNAPSHOT_VERSION )) {
201+ source = in .readOptionalWriteable (SnapshotId ::new );
202+ clones = in .readImmutableMap (RepositoryShardId ::new , ShardSnapshotStatus ::readFrom );
203+ } else {
204+ source = null ;
205+ clones = ImmutableOpenMap .of ();
206+ }
147207 }
148208
149- private static boolean assertShardsConsistent (State state , List <IndexId > indices ,
150- ImmutableOpenMap <ShardId , ShardSnapshotStatus > shards ) {
209+ private static boolean assertShardsConsistent (SnapshotId source , State state , List <IndexId > indices ,
210+ ImmutableOpenMap <ShardId , ShardSnapshotStatus > shards ,
211+ ImmutableOpenMap <RepositoryShardId , ShardSnapshotStatus > clones ) {
151212 if ((state == State .INIT || state == State .ABORTED ) && shards .isEmpty ()) {
152213 return true ;
153214 }
154215 final Set <String > indexNames = indices .stream ().map (IndexId ::getName ).collect (Collectors .toSet ());
155216 final Set <String > indexNamesInShards = new HashSet <>();
156- shards .keysIt ().forEachRemaining (s -> indexNamesInShards .add (s .getIndexName ()));
157- assert indexNames .equals (indexNamesInShards )
217+ shards .iterator ().forEachRemaining (s -> {
218+ indexNamesInShards .add (s .key .getIndexName ());
219+ assert source == null || s .value .nodeId == null :
220+ "Shard snapshot must not be assigned to data node when copying from snapshot [" + source + "]" ;
221+ });
222+ assert source == null || indexNames .isEmpty () == false : "No empty snapshot clones allowed" ;
223+ assert source != null || indexNames .equals (indexNamesInShards )
158224 : "Indices in shards " + indexNamesInShards + " differ from expected indices " + indexNames + " for state [" + state + "]" ;
159- final boolean shardsCompleted = completed (shards .values ());
160- assert (state .completed () && shardsCompleted ) || (state .completed () == false && shardsCompleted == false )
161- : "Completed state must imply all shards completed but saw state [" + state + "] and shards " + shards ;
225+ final boolean shardsCompleted = completed (shards .values ()) && completed (clones .values ());
226+ // Check state consistency for normal snapshots and started clone operations
227+ if (source == null || clones .isEmpty () == false ) {
228+ assert (state .completed () && shardsCompleted ) || (state .completed () == false && shardsCompleted == false )
229+ : "Completed state must imply all shards completed but saw state [" + state + "] and shards " + shards ;
230+ }
231+ if (source != null && state .completed ()) {
232+ assert hasFailures (clones ) == false || state == State .FAILED
233+ : "Failed shard clones in [" + clones + "] but state was [" + state + "]" ;
234+ }
162235 return true ;
163236 }
164237
165238 public Entry withRepoGen (long newRepoGen ) {
166239 assert newRepoGen > repositoryStateId : "Updated repository generation [" + newRepoGen
167240 + "] must be higher than current generation [" + repositoryStateId + "]" ;
168241 return new Entry (snapshot , includeGlobalState , partial , state , indices , dataStreams , startTime , newRepoGen , shards , failure ,
169- userMetadata , version );
242+ userMetadata , version , source , clones );
243+ }
244+
245+ public Entry withClones (ImmutableOpenMap <RepositoryShardId , ShardSnapshotStatus > updatedClones ) {
246+ if (updatedClones .equals (clones )) {
247+ return this ;
248+ }
249+ return new Entry (snapshot , includeGlobalState , partial ,
250+ completed (updatedClones .values ()) ? (hasFailures (updatedClones ) ? State .FAILED : State .SUCCESS ) :
251+ state , indices , dataStreams , startTime , repositoryStateId , shards , failure , userMetadata , version , source ,
252+ updatedClones );
170253 }
171254
172255 /**
@@ -203,7 +286,7 @@ public Entry abort() {
203286
204287 public Entry fail (ImmutableOpenMap <ShardId , ShardSnapshotStatus > shards , State state , String failure ) {
205288 return new Entry (snapshot , includeGlobalState , partial , state , indices , dataStreams , startTime , repositoryStateId , shards ,
206- failure , userMetadata , version );
289+ failure , userMetadata , version , source , clones );
207290 }
208291
209292 /**
@@ -291,6 +374,19 @@ public Version version() {
291374 return version ;
292375 }
293376
377+ @ Nullable
378+ public SnapshotId source () {
379+ return source ;
380+ }
381+
382+ public boolean isClone () {
383+ return source != null ;
384+ }
385+
386+ public ImmutableOpenMap <RepositoryShardId , ShardSnapshotStatus > clones () {
387+ return clones ;
388+ }
389+
294390 @ Override
295391 public boolean equals (Object o ) {
296392 if (this == o ) return true ;
@@ -307,6 +403,8 @@ public boolean equals(Object o) {
307403 if (state != entry .state ) return false ;
308404 if (repositoryStateId != entry .repositoryStateId ) return false ;
309405 if (version .equals (entry .version ) == false ) return false ;
406+ if (Objects .equals (source , ((Entry ) o ).source ) == false ) return false ;
407+ if (clones .equals (((Entry ) o ).clones ) == false ) return false ;
310408
311409 return true ;
312410 }
@@ -322,6 +420,8 @@ public int hashCode() {
322420 result = 31 * result + Long .hashCode (startTime );
323421 result = 31 * result + Long .hashCode (repositoryStateId );
324422 result = 31 * result + version .hashCode ();
423+ result = 31 * result + (source == null ? 0 : source .hashCode ());
424+ result = 31 * result + clones .hashCode ();
325425 return result ;
326426 }
327427
@@ -383,6 +483,10 @@ public void writeTo(StreamOutput out) throws IOException {
383483 out .writeMap (userMetadata );
384484 Version .writeVersion (version , out );
385485 out .writeStringCollection (dataStreams );
486+ if (out .getVersion ().onOrAfter (SnapshotsService .CLONE_SNAPSHOT_VERSION )) {
487+ out .writeOptionalWriteable (source );
488+ out .writeMap (clones );
489+ }
386490 }
387491
388492 @ Override
@@ -406,6 +510,15 @@ public static boolean completed(ObjectContainer<ShardSnapshotStatus> shards) {
406510 return true ;
407511 }
408512
513+ private static boolean hasFailures (ImmutableOpenMap <RepositoryShardId , ShardSnapshotStatus > clones ) {
514+ for (ObjectCursor <ShardSnapshotStatus > value : clones .values ()) {
515+ if (value .value .state ().failed ()) {
516+ return true ;
517+ }
518+ }
519+ return false ;
520+ }
521+
409522 public static class ShardSnapshotStatus implements Writeable {
410523
411524 /**
0 commit comments