Skip to content

Commit

Permalink
WIP changes
Browse files Browse the repository at this point in the history 
Signed-off-by: Gaurav Bafna <[email protected]>
  • Loading branch information
@gbbafna
gbbafna committed Sep 14, 2024
1 parent 66c7780 commit b2158e0
Show file tree
Hide file tree
Showing 2 changed files with 378 additions and 25 deletions.
140 changes: 131 additions & 9 deletions server/src/main/java/org/opensearch/cluster/SnapshotsInProgress.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,38 @@ public static Entry startedEntry(
);
}

public static Entry startedEntry(
Snapshot snapshot,
boolean includeGlobalState,
boolean partial,
List<IndexId> indices,
List<String> dataStreams,
long startTime,
long repositoryStateId,
final Map<ShardId, ShardSnapshotStatus> shards,
Map<String, Object> userMetadata,
Version version,
boolean remoteStoreIndexShallowCopy,
boolean remoteStoreIndexShallowCopyV2
) {
return new SnapshotsInProgress.Entry(
snapshot,
includeGlobalState,
partial,
completed(shards.values()) ? State.SUCCESS : State.STARTED,
indices,
dataStreams,
startTime,
repositoryStateId,
shards,
null,
userMetadata,
version,
remoteStoreIndexShallowCopy,
remoteStoreIndexShallowCopyV2
);
}

/**
* Creates the initial snapshot clone entry
*
Expand Down Expand Up @@ -168,11 +200,42 @@ public static Entry startClone(
version,
source,
Map.of(),
false // initialising to false, will be updated in startCloning method of SnapshotsService while updating entry with
false,
false// initialising to false, will be updated in startCloning method of SnapshotsService while updating entry with
// clone jobs
);
}

public static Entry startClone(
Snapshot snapshot,
SnapshotId source,
List<IndexId> indices,
long startTime,
long repositoryStateId,
Version version,
boolean remoteStoreIndexShallowCopyV2
) {
return new SnapshotsInProgress.Entry(
snapshot,
true,
false,
State.STARTED,
indices,
Collections.emptyList(),
startTime,
repositoryStateId,
Map.of(),
null,
Collections.emptyMap(),
version,
source,
Map.of(),
remoteStoreIndexShallowCopyV2,
remoteStoreIndexShallowCopyV2// initialising to false, will be updated in startCloning method of SnapshotsService while updating entry with
// clone jobs
);
}

/**
* Entry in the collection.
*
Expand All @@ -183,6 +246,8 @@ public static class Entry implements Writeable, ToXContent, RepositoryOperation
private final Snapshot snapshot;
private final boolean includeGlobalState;
private final boolean remoteStoreIndexShallowCopy;

private final boolean remoteStoreIndexShallowCopyV2;
private final boolean partial;
/**
* Map of {@link ShardId} to {@link ShardSnapshotStatus} tracking the state of each shard snapshot operation.
Expand Down Expand Up @@ -212,6 +277,42 @@ public static class Entry implements Writeable, ToXContent, RepositoryOperation
@Nullable
private final String failure;

public Entry(
Snapshot snapshot,
boolean includeGlobalState,
boolean partial,
State state,
List<IndexId> indices,
List<String> dataStreams,
long startTime,
long repositoryStateId,
final Map<ShardId, ShardSnapshotStatus> shards,
String failure,
Map<String, Object> userMetadata,
Version version,
boolean remoteStoreIndexShallowCopy,
boolean remoteStoreIndexShallowCopyV2
) {
this(
snapshot,
includeGlobalState,
partial,
state,
indices,
dataStreams,
startTime,
repositoryStateId,
shards,
failure,
userMetadata,
version,
null,
Map.of(),
remoteStoreIndexShallowCopy,
remoteStoreIndexShallowCopyV2
);
}

// visible for testing, use #startedEntry and copy constructors in production code
public Entry(
Snapshot snapshot,
Expand Down Expand Up @@ -243,11 +344,12 @@ public Entry(
version,
null,
Map.of(),
remoteStoreIndexShallowCopy
remoteStoreIndexShallowCopy,
false
);
}

private Entry(
public Entry(
Snapshot snapshot,
boolean includeGlobalState,
boolean partial,
Expand All @@ -262,7 +364,8 @@ private Entry(
Version version,
@Nullable SnapshotId source,
@Nullable final Map<RepositoryShardId, ShardSnapshotStatus> clones,
boolean remoteStoreIndexShallowCopy
boolean remoteStoreIndexShallowCopy,
boolean remoteStoreIndexShallowCopyV2
) {
this.state = state;
this.snapshot = snapshot;
Expand All @@ -284,7 +387,8 @@ private Entry(
this.clones = Collections.unmodifiableMap(clones);
}
this.remoteStoreIndexShallowCopy = remoteStoreIndexShallowCopy;
assert assertShardsConsistent(this.source, this.state, this.indices, this.shards, this.clones);
this.remoteStoreIndexShallowCopyV2 = remoteStoreIndexShallowCopyV2;
assert this.remoteStoreIndexShallowCopyV2 || assertShardsConsistent(this.source, this.state, this.indices, this.shards, this.clones);
}

private Entry(StreamInput in) throws IOException {
Expand All @@ -307,6 +411,11 @@ private Entry(StreamInput in) throws IOException {
} else {
remoteStoreIndexShallowCopy = false;
}
if (in.getVersion().onOrAfter(Version.CURRENT)) {
remoteStoreIndexShallowCopyV2 = in.readBoolean();
} else {
remoteStoreIndexShallowCopyV2 = false;
}
}

private static boolean assertShardsConsistent(
Expand Down Expand Up @@ -428,7 +537,8 @@ public Entry withRepoGen(long newRepoGen) {
version,
source,
clones,
remoteStoreIndexShallowCopy
remoteStoreIndexShallowCopy,
remoteStoreIndexShallowCopyV2
);
}

Expand All @@ -451,7 +561,8 @@ public Entry withClones(final Map<RepositoryShardId, ShardSnapshotStatus> update
version,
source,
updatedClones,
remoteStoreIndexShallowCopy
remoteStoreIndexShallowCopy,
remoteStoreIndexShallowCopyV2
);
}

Expand All @@ -471,7 +582,8 @@ public Entry withRemoteStoreIndexShallowCopy(final boolean remoteStoreIndexShall
version,
source,
clones,
remoteStoreIndexShallowCopy
remoteStoreIndexShallowCopy,
remoteStoreIndexShallowCopyV2
);
}

Expand Down Expand Up @@ -527,7 +639,8 @@ public Entry fail(final Map<ShardId, ShardSnapshotStatus> shards, State state, S
version,
source,
clones,
remoteStoreIndexShallowCopy
remoteStoreIndexShallowCopy,
remoteStoreIndexShallowCopyV2
);
}

Expand Down Expand Up @@ -614,6 +727,10 @@ public boolean remoteStoreIndexShallowCopy() {
return remoteStoreIndexShallowCopy;
}

public boolean remoteStoreIndexShallowCopyV2() {
return remoteStoreIndexShallowCopyV2;
}

public Map<String, Object> userMetadata() {
return userMetadata;
}
Expand Down Expand Up @@ -678,6 +795,7 @@ public boolean equals(Object o) {
if (Objects.equals(source, ((Entry) o).source) == false) return false;
if (clones.equals(((Entry) o).clones) == false) return false;
if (remoteStoreIndexShallowCopy != entry.remoteStoreIndexShallowCopy) return false;
if (remoteStoreIndexShallowCopyV2 != entry.remoteStoreIndexShallowCopyV2) return false;
return true;
}

Expand All @@ -695,6 +813,7 @@ public int hashCode() {
result = 31 * result + (source == null ? 0 : source.hashCode());
result = 31 * result + clones.hashCode();
result = 31 * result + (remoteStoreIndexShallowCopy ? 1 : 0);
result = 31 * result + (remoteStoreIndexShallowCopyV2 ? 1 : 0);
return result;
}

Expand Down Expand Up @@ -766,6 +885,9 @@ public void writeTo(StreamOutput out) throws IOException {
if (out.getVersion().onOrAfter(Version.V_2_9_0)) {
out.writeBoolean(remoteStoreIndexShallowCopy);
}
if (out.getVersion().onOrAfter(Version.CURRENT)) {
out.writeBoolean(remoteStoreIndexShallowCopyV2);
}
}

@Override
Expand Down
Loading

0 comments on commit b2158e0

Please sign in to comment.