From feae4bfefaff41cb994408a700bb9495b65c024b Mon Sep 17 00:00:00 2001 From: Sooraj Sinha <81695996+soosinha@users.noreply.github.com> Date: Thu, 7 Sep 2023 20:01:04 +0530 Subject: [PATCH] Add logic to fetch previousClusterUUID (#9746) Signed-off-by: Sooraj Sinha (cherry picked from commit bdf6f1d09702aebd8b1e94a2cb722c14a6825297) --- .../remote/ClusterMetadataManifest.java | 46 +++++- .../remote/RemoteClusterStateService.java | 146 ++++++++++++++++-- .../coordination/CoordinationStateTests.java | 6 +- .../remote/ClusterMetadataManifestTests.java | 22 ++- .../RemoteClusterStateServiceTests.java | 110 +++++++++++++ 5 files changed, 305 insertions(+), 25 deletions(-) diff --git a/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java b/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java index cac77f9996438..0ebbdc81661ad 100644 --- a/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java +++ b/server/src/main/java/org/opensearch/gateway/remote/ClusterMetadataManifest.java @@ -41,6 +41,7 @@ public class ClusterMetadataManifest implements Writeable, ToXContentFragment { private static final ParseField NODE_ID_FIELD = new ParseField("node_id"); private static final ParseField COMMITTED_FIELD = new ParseField("committed"); private static final ParseField INDICES_FIELD = new ParseField("indices"); + private static final ParseField PREVIOUS_CLUSTER_UUID = new ParseField("previous_cluster_uuid"); private static long term(Object[] fields) { return (long) fields[0]; @@ -74,6 +75,10 @@ private static List indices(Object[] fields) { return (List) fields[7]; } + private static String previousClusterUUID(Object[] fields) { + return (String) fields[8]; + } + private static final ConstructingObjectParser PARSER = new ConstructingObjectParser<>( "cluster_metadata_manifest", fields -> new ClusterMetadataManifest( @@ -84,7 +89,8 @@ private static List indices(Object[] fields) { opensearchVersion(fields), nodeId(fields), committed(fields), - indices(fields) + indices(fields), + previousClusterUUID(fields) ) ); @@ -101,6 +107,7 @@ private static List indices(Object[] fields) { (p, c) -> UploadedIndexMetadata.fromXContent(p), INDICES_FIELD ); + PARSER.declareString(ConstructingObjectParser.constructorArg(), PREVIOUS_CLUSTER_UUID); } private final List indices; @@ -111,6 +118,7 @@ private static List indices(Object[] fields) { private final Version opensearchVersion; private final String nodeId; private final boolean committed; + private final String previousClusterUUID; public List getIndices() { return indices; @@ -144,6 +152,10 @@ public boolean isCommitted() { return committed; } + public String getPreviousClusterUUID() { + return previousClusterUUID; + } + public ClusterMetadataManifest( long clusterTerm, long version, @@ -152,7 +164,8 @@ public ClusterMetadataManifest( Version opensearchVersion, String nodeId, boolean committed, - List indices + List indices, + String previousClusterUUID ) { this.clusterTerm = clusterTerm; this.stateVersion = version; @@ -162,6 +175,7 @@ public ClusterMetadataManifest( this.nodeId = nodeId; this.committed = committed; this.indices = Collections.unmodifiableList(indices); + this.previousClusterUUID = previousClusterUUID; } public ClusterMetadataManifest(StreamInput in) throws IOException { @@ -173,6 +187,7 @@ public ClusterMetadataManifest(StreamInput in) throws IOException { this.nodeId = in.readString(); this.committed = in.readBoolean(); this.indices = Collections.unmodifiableList(in.readList(UploadedIndexMetadata::new)); + this.previousClusterUUID = in.readString(); } public static Builder builder() { @@ -199,6 +214,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws } } builder.endArray(); + builder.field(PREVIOUS_CLUSTER_UUID.getPreferredName(), getPreviousClusterUUID()); return builder; } @@ -212,6 +228,7 @@ public void writeTo(StreamOutput out) throws IOException { out.writeString(nodeId); out.writeBoolean(committed); out.writeCollection(indices); + out.writeString(previousClusterUUID); } @Override @@ -230,12 +247,23 @@ public boolean equals(Object o) { && Objects.equals(stateUUID, that.stateUUID) && Objects.equals(opensearchVersion, that.opensearchVersion) && Objects.equals(nodeId, that.nodeId) - && Objects.equals(committed, that.committed); + && Objects.equals(committed, that.committed) + && Objects.equals(previousClusterUUID, that.previousClusterUUID); } @Override public int hashCode() { - return Objects.hash(indices, clusterTerm, stateVersion, clusterUUID, stateUUID, opensearchVersion, nodeId, committed); + return Objects.hash( + indices, + clusterTerm, + stateVersion, + clusterUUID, + stateUUID, + opensearchVersion, + nodeId, + committed, + previousClusterUUID + ); } @Override @@ -261,6 +289,7 @@ public static class Builder { private String stateUUID; private Version opensearchVersion; private String nodeId; + private String previousClusterUUID; private boolean committed; public Builder indices(List indices) { @@ -307,6 +336,11 @@ public List getIndices() { return indices; } + public Builder previousClusterUUID(String previousClusterUUID) { + this.previousClusterUUID = previousClusterUUID; + return this; + } + public Builder() { indices = new ArrayList<>(); } @@ -320,6 +354,7 @@ public Builder(ClusterMetadataManifest manifest) { this.nodeId = manifest.nodeId; this.committed = manifest.committed; this.indices = new ArrayList<>(manifest.indices); + this.previousClusterUUID = manifest.previousClusterUUID; } public ClusterMetadataManifest build() { @@ -331,7 +366,8 @@ public ClusterMetadataManifest build() { opensearchVersion, nodeId, committed, - indices + indices, + previousClusterUUID ); } diff --git a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java index d2f6fd8ebe228..990fc20a7e95d 100644 --- a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java +++ b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java @@ -46,6 +46,7 @@ import java.util.Locale; import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.function.Function; @@ -93,6 +94,11 @@ public class RemoteClusterStateService implements Closeable { private static final Logger logger = LogManager.getLogger(RemoteClusterStateService.class); public static final String DELIMITER = "__"; + private static final String CLUSTER_STATE_PATH_TOKEN = "cluster-state"; + private static final String INDEX_PATH_TOKEN = "index"; + private static final String MANIFEST_PATH_TOKEN = "manifest"; + private static final String MANIFEST_FILE_PREFIX = "manifest"; + private static final String INDEX_METADATA_FILE_PREFIX = "metadata"; private final String nodeId; private final Supplier repositoriesService; @@ -159,8 +165,8 @@ public ClusterMetadataManifest writeFullMetadata(ClusterState clusterState) thro /** * This method uploads the diff between the previous cluster state and the current cluster state. The previous manifest file is needed to create the new - * manifest. The new manifest file is created by using the unchanged metadata from the previous manifest and the new metadata changes from the current cluster - * state. + * manifest. The new manifest file is created by using the unchanged metadata from the previous manifest and the new metadata changes from the current + * cluster state. * * @return The uploaded ClusterMetadataManifest file */ @@ -238,10 +244,10 @@ public ClusterMetadataManifest writeIncrementalMetadata( /** * Uploads provided IndexMetadata's to remote store in parallel. The call is blocking so the method waits for upload to finish and then return. + * * @param clusterState current ClusterState * @param toUpload list of IndexMetadata to upload * @return {@code List} list of IndexMetadata uploaded to remote - * @throws IOException */ private List writeIndexMetadataParallel(ClusterState clusterState, List toUpload) throws IOException { @@ -312,10 +318,10 @@ private List writeIndexMetadataParallel(ClusterState clus /** * Allows async Upload of IndexMetadata to remote + * * @param clusterState current ClusterState * @param indexMetadata {@link IndexMetadata} to upload * @param latchedActionListener listener to respond back on after upload finishes - * @throws IOException */ private void writeIndexMetadataAsync( ClusterState clusterState, @@ -360,11 +366,6 @@ public ClusterMetadataManifest markLastStateAsCommitted(ClusterState clusterStat return uploadManifest(clusterState, previousManifest.getIndices(), true); } - public ClusterState getLatestClusterState(String clusterUUID) { - // todo - return null; - } - @Override public void close() throws IOException { if (blobStoreRepository != null) { @@ -402,7 +403,9 @@ private ClusterMetadataManifest uploadManifest( Version.CURRENT, nodeId, committed, - uploadedIndexMetadata + uploadedIndexMetadata, + // todo Change this to proper cluster UUID + ClusterState.UNKNOWN_UUID ); writeMetadataManifest(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID(), manifest, manifestFileName); return manifest; @@ -421,9 +424,9 @@ private BlobContainer indexMetadataContainer(String clusterName, String clusterU .blobContainer( blobStoreRepository.basePath() .add(encodeString(clusterName)) - .add("cluster-state") + .add(CLUSTER_STATE_PATH_TOKEN) .add(clusterUUID) - .add("index") + .add(INDEX_PATH_TOKEN) .add(indexUUID) ); } @@ -432,7 +435,20 @@ private BlobContainer manifestContainer(String clusterName, String clusterUUID) // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/manifest return blobStoreRepository.blobStore() .blobContainer( - blobStoreRepository.basePath().add(encodeString(clusterName)).add("cluster-state").add(clusterUUID).add("manifest") + blobStoreRepository.basePath() + .add(encodeString(clusterName)) + .add(CLUSTER_STATE_PATH_TOKEN) + .add(clusterUUID) + .add(MANIFEST_PATH_TOKEN) + ); + } + + private BlobContainer clusterUUIDContainer(String clusterName) { + return blobStoreRepository.blobStore() + .blobContainer( + blobStoreRepository.basePath() + .add(Base64.getUrlEncoder().withoutPadding().encodeToString(clusterName.getBytes(StandardCharsets.UTF_8))) + .add(CLUSTER_STATE_PATH_TOKEN) ); } @@ -444,7 +460,7 @@ private static String getManifestFileName(long term, long version) { // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/manifest/manifest_2147483642_2147483637_456536447 return String.join( DELIMITER, - "manifest", + MANIFEST_FILE_PREFIX, RemoteStoreUtils.invertLong(term), RemoteStoreUtils.invertLong(version), RemoteStoreUtils.invertLong(System.currentTimeMillis()) @@ -452,11 +468,17 @@ private static String getManifestFileName(long term, long version) { } private static String indexMetadataFileName(IndexMetadata indexMetadata) { - return String.join(DELIMITER, "metadata", String.valueOf(indexMetadata.getVersion()), String.valueOf(System.currentTimeMillis())); + return String.join( + DELIMITER, + INDEX_METADATA_FILE_PREFIX, + String.valueOf(indexMetadata.getVersion()), + String.valueOf(System.currentTimeMillis()) + ); } /** * Fetch latest index metadata from remote cluster state + * * @param clusterUUID uuid of cluster state to refer to in remote * @param clusterName name of the cluster * @return {@code Map} latest IndexUUID to IndexMetadata map @@ -476,6 +498,7 @@ public Map getLatestIndexMetadata(String clusterName, Str /** * Fetch index metadata from remote cluster state + * * @param clusterUUID uuid of cluster state to refer to in remote * @param clusterName name of the cluster * @param uploadedIndexMetadata {@link UploadedIndexMetadata} contains details about remote location of index metadata @@ -499,6 +522,7 @@ private IndexMetadata getIndexMetadata(String clusterName, String clusterUUID, U /** * Fetch latest ClusterMetadataManifest from remote state store + * * @param clusterUUID uuid of cluster state to refer to in remote * @param clusterName name of the cluster * @return ClusterMetadataManifest @@ -508,8 +532,97 @@ public ClusterMetadataManifest getLatestClusterMetadataManifest(String clusterNa return fetchRemoteClusterMetadataManifest(clusterName, clusterUUID, latestManifestFileName); } + /** + * Fetch the previous cluster UUIDs from remote state store and return the most recent valid cluster UUID + * + * @param clusterName The cluster name for which previous cluster UUID is to be fetched + * @return Last valid cluster UUID + */ + public String getLatestClusterUUID(String clusterName) { + try { + Set clusterUUIDs = getAllClusterUUIDs(clusterName); + Map latestManifests = getLatestManifestForAllClusterUUIDs(clusterName, clusterUUIDs); + List validChain = createClusterChain(latestManifests); + if (validChain.isEmpty()) { + return ClusterState.UNKNOWN_UUID; + } + return validChain.get(0); + } catch (IOException e) { + throw new IllegalStateException( + String.format(Locale.ROOT, "Error while fetching previous UUIDs from remote store for cluster name: %s", clusterName) + ); + } + } + + private Set getAllClusterUUIDs(String clusterName) throws IOException { + Map clusterUUIDMetadata = clusterUUIDContainer(clusterName).children(); + if (clusterUUIDMetadata == null) { + return Collections.emptySet(); + } + return Collections.unmodifiableSet(clusterUUIDMetadata.keySet()); + } + + private Map getLatestManifestForAllClusterUUIDs(String clusterName, Set clusterUUIDs) { + Map manifestsByClusterUUID = new HashMap<>(); + for (String clusterUUID : clusterUUIDs) { + try { + ClusterMetadataManifest manifest = getLatestClusterMetadataManifest(clusterName, clusterUUID); + manifestsByClusterUUID.put(clusterUUID, manifest); + } catch (Exception e) { + throw new IllegalStateException( + String.format(Locale.ROOT, "Exception in fetching manifest for clusterUUID: %s", clusterUUID) + ); + } + } + return manifestsByClusterUUID; + } + + /** + * This method creates a valid cluster UUID chain. + * + * @param manifestsByClusterUUID Map of latest ClusterMetadataManifest for every cluster UUID + * @return List of cluster UUIDs. The first element is the most recent cluster UUID in the chain + */ + private List createClusterChain(final Map manifestsByClusterUUID) { + final Map clusterUUIDGraph = manifestsByClusterUUID.values() + .stream() + .collect(Collectors.toMap(ClusterMetadataManifest::getClusterUUID, ClusterMetadataManifest::getPreviousClusterUUID)); + final List validClusterUUIDs = manifestsByClusterUUID.values() + .stream() + .filter(m -> !isInvalidClusterUUID(m) && !clusterUUIDGraph.containsValue(m.getClusterUUID())) + .map(ClusterMetadataManifest::getClusterUUID) + .collect(Collectors.toList()); + if (validClusterUUIDs.isEmpty()) { + logger.info("There is no valid previous cluster UUID"); + return Collections.emptyList(); + } + if (validClusterUUIDs.size() > 1) { + throw new IllegalStateException( + String.format( + Locale.ROOT, + "The system has ended into multiple valid cluster states in the remote store. " + + "Please check their latest manifest to decide which one you want to keep. Valid Cluster UUIDs: - %s", + validClusterUUIDs + ) + ); + } + final List validChain = new ArrayList<>(); + String currentUUID = validClusterUUIDs.get(0); + while (!ClusterState.UNKNOWN_UUID.equals(currentUUID)) { + validChain.add(currentUUID); + // Getting the previous cluster UUID of a cluster UUID from the clusterUUID Graph + currentUUID = clusterUUIDGraph.get(currentUUID); + } + return validChain; + } + + private boolean isInvalidClusterUUID(ClusterMetadataManifest manifest) { + return !manifest.isCommitted() && manifest.getIndices().isEmpty(); + } + /** * Fetch latest ClusterMetadataManifest file from remote state store + * * @param clusterUUID uuid of cluster state to refer to in remote * @param clusterName name of the cluster * @return latest ClusterMetadataManifest filename @@ -522,7 +635,7 @@ private String getLatestManifestFileName(String clusterName, String clusterUUID) * when sorted in LEXICOGRAPHIC order the latest uploaded manifest file comes on top. */ List manifestFilesMetadata = manifestContainer(clusterName, clusterUUID).listBlobsByPrefixInSortedOrder( - "manifest" + DELIMITER, + MANIFEST_FILE_PREFIX + DELIMITER, 1, BlobContainer.BlobNameSortOrder.LEXICOGRAPHIC ); @@ -538,6 +651,7 @@ private String getLatestManifestFileName(String clusterName, String clusterUUID) /** * Fetch ClusterMetadataManifest from remote state store + * * @param clusterUUID uuid of cluster state to refer to in remote * @param clusterName name of the cluster * @return ClusterMetadataManifest diff --git a/server/src/test/java/org/opensearch/cluster/coordination/CoordinationStateTests.java b/server/src/test/java/org/opensearch/cluster/coordination/CoordinationStateTests.java index ef56d70d6153c..aa4472c4fcec5 100644 --- a/server/src/test/java/org/opensearch/cluster/coordination/CoordinationStateTests.java +++ b/server/src/test/java/org/opensearch/cluster/coordination/CoordinationStateTests.java @@ -935,7 +935,8 @@ public void testHandlePrePublishAndCommitWhenRemoteStateEnabled() throws IOExcep Version.CURRENT, randomAlphaOfLength(10), false, - Collections.emptyList() + Collections.emptyList(), + randomAlphaOfLength(10) ) ); @@ -976,7 +977,8 @@ public void testHandlePrePublishAndCommitWhenRemoteStateEnabled() throws IOExcep Version.CURRENT, randomAlphaOfLength(10), false, - Collections.emptyList() + Collections.emptyList(), + randomAlphaOfLength(10) ); Mockito.when(remoteClusterStateService.writeFullMetadata(clusterState2)).thenReturn(manifest2); coordinationState.handlePrePublish(clusterState2); diff --git a/server/src/test/java/org/opensearch/gateway/remote/ClusterMetadataManifestTests.java b/server/src/test/java/org/opensearch/gateway/remote/ClusterMetadataManifestTests.java index 449af91c34531..9f8dde5ba9d45 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/ClusterMetadataManifestTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/ClusterMetadataManifestTests.java @@ -36,7 +36,8 @@ public void testClusterMetadataManifestXContent() throws IOException { Version.CURRENT, "test-node-id", false, - Collections.singletonList(uploadedIndexMetadata) + Collections.singletonList(uploadedIndexMetadata), + "prev-cluster-uuid" ); final XContentBuilder builder = JsonXContent.contentBuilder(); builder.startObject(); @@ -58,7 +59,8 @@ public void testClusterMetadataManifestSerializationEqualsHashCode() { Version.CURRENT, "B10RX1f5RJenMQvYccCgSQ", true, - randomUploadedIndexMetadataList() + randomUploadedIndexMetadataList(), + "yfObdx8KSMKKrXf8UyHhM" ); { // Mutate Cluster Term EqualsHashCodeTestUtils.checkEqualsAndHashCode( @@ -165,6 +167,22 @@ public void testClusterMetadataManifestSerializationEqualsHashCode() { } ); } + { // Mutate Previous cluster UUID + EqualsHashCodeTestUtils.checkEqualsAndHashCode( + initialManifest, + orig -> OpenSearchTestCase.copyWriteable( + orig, + new NamedWriteableRegistry(Collections.emptyList()), + ClusterMetadataManifest::new + ), + manifest -> { + ClusterMetadataManifest.Builder builder = ClusterMetadataManifest.builder(manifest); + builder.previousClusterUUID("vZX62DCQEOzGXlxXCrEu"); + return builder.build(); + } + ); + + } } private List randomUploadedIndexMetadataList() { diff --git a/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java b/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java index da1f97f7042f1..156138340db10 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java @@ -54,13 +54,16 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.Set; import java.util.function.Supplier; import org.mockito.ArgumentCaptor; +import org.mockito.ArgumentMatchers; import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_CLUSTER_STATE_REPOSITORY_NAME_ATTRIBUTE_KEY; import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_REPOSITORY_SETTINGS_ATTRIBUTE_KEY_PREFIX; import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_REPOSITORY_TYPE_ATTRIBUTE_KEY_FORMAT; +import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.notNullValue; import static org.hamcrest.Matchers.nullValue; @@ -163,6 +166,7 @@ public void testWriteFullMetadataSuccess() throws IOException { .stateVersion(1L) .stateUUID("state-uuid") .clusterUUID("cluster-uuid") + .previousClusterUUID("prev-cluster-uuid") .build(); assertThat(manifest.getIndices().size(), is(1)); @@ -290,6 +294,7 @@ public void testWriteIncrementalMetadataSuccess() throws IOException { .stateVersion(1L) .stateUUID("state-uuid") .clusterUUID("cluster-uuid") + .previousClusterUUID("prev-cluster-uuid") .build(); assertThat(manifest.getIndices().size(), is(1)); @@ -383,6 +388,7 @@ public void testReadLatestMetadataManifestSuccessButNoIndexMetadata() throws IOE .clusterUUID("cluster-uuid") .nodeId("nodeA") .opensearchVersion(VersionUtils.randomOpenSearchVersion(random())) + .previousClusterUUID("prev-cluster-uuid") .build(); BlobContainer blobContainer = mockBlobStoreObjects(); @@ -408,6 +414,7 @@ public void testReadLatestMetadataManifestSuccessButIndexMetadataFetchIOExceptio .clusterUUID("cluster-uuid") .nodeId("nodeA") .opensearchVersion(VersionUtils.randomOpenSearchVersion(random())) + .previousClusterUUID("prev-cluster-uuid") .build(); BlobContainer blobContainer = mockBlobStoreObjects(); @@ -438,6 +445,7 @@ public void testReadLatestMetadataManifestSuccess() throws IOException { .clusterUUID("cluster-uuid") .nodeId("nodeA") .opensearchVersion(VersionUtils.randomOpenSearchVersion(random())) + .previousClusterUUID("prev-cluster-uuid") .build(); mockBlobContainer(mockBlobStoreObjects(), expectedManifest, new HashMap<>()); @@ -482,6 +490,7 @@ public void testReadLatestIndexMetadataSuccess() throws IOException { .clusterUUID("cluster-uuid") .nodeId("nodeA") .opensearchVersion(VersionUtils.randomOpenSearchVersion(random())) + .previousClusterUUID("prev-cluster-uuid") .build(); mockBlobContainer(mockBlobStoreObjects(), expectedManifest, Map.of(index.getUUID(), indexMetadata)); @@ -514,6 +523,7 @@ public void testMarkLastStateAsCommittedSuccess() throws IOException { .stateUUID("state-uuid") .clusterUUID("cluster-uuid") .nodeId("nodeA") + .previousClusterUUID("prev-cluster-uuid") .build(); assertThat(manifest.getIndices().size(), is(1)); @@ -526,6 +536,98 @@ public void testMarkLastStateAsCommittedSuccess() throws IOException { assertThat(manifest.getStateUUID(), is(expectedManifest.getStateUUID())); } + public void testGetValidPreviousClusterUUID() throws IOException { + Map clusterUUIDsPointers = Map.of( + "cluster-uuid1", + ClusterState.UNKNOWN_UUID, + "cluster-uuid2", + "cluster-uuid1", + "cluster-uuid3", + "cluster-uuid2" + ); + mockObjectsForGettingPreviousClusterUUID(clusterUUIDsPointers); + + remoteClusterStateService.ensureRepositorySet(); + String previousClusterUUID = remoteClusterStateService.getLatestClusterUUID("test-cluster"); + assertThat(previousClusterUUID, equalTo("cluster-uuid3")); + } + + public void testGetValidPreviousClusterUUIDForInvalidChain() throws IOException { + Map clusterUUIDsPointers = Map.of( + "cluster-uuid1", + ClusterState.UNKNOWN_UUID, + "cluster-uuid2", + ClusterState.UNKNOWN_UUID, + "cluster-uuid3", + "cluster-uuid2" + ); + mockObjectsForGettingPreviousClusterUUID(clusterUUIDsPointers); + + remoteClusterStateService.ensureRepositorySet(); + assertThrows(IllegalStateException.class, () -> remoteClusterStateService.getLatestClusterUUID("test-cluster")); + } + + private void mockObjectsForGettingPreviousClusterUUID(Map clusterUUIDsPointers) throws IOException { + final BlobPath blobPath = mock(BlobPath.class); + when((blobStoreRepository.basePath())).thenReturn(blobPath); + when(blobPath.add(anyString())).thenReturn(blobPath); + when(blobPath.buildAsString()).thenReturn("/blob/path/"); + BlobContainer blobContainer1 = mock(BlobContainer.class); + BlobContainer blobContainer2 = mock(BlobContainer.class); + BlobContainer blobContainer3 = mock(BlobContainer.class); + BlobContainer uuidBlobContainer = mock(BlobContainer.class); + when(blobContainer1.path()).thenReturn(blobPath); + when(blobContainer2.path()).thenReturn(blobPath); + when(blobContainer3.path()).thenReturn(blobPath); + + mockBlobContainerForClusterUUIDs(uuidBlobContainer, clusterUUIDsPointers.keySet()); + final ClusterMetadataManifest clusterManifest1 = generateClusterMetadataManifest( + "cluster-uuid1", + clusterUUIDsPointers.get("cluster-uuid1"), + randomAlphaOfLength(10) + ); + mockBlobContainer(blobContainer1, clusterManifest1, Map.of()); + + final ClusterMetadataManifest clusterManifest2 = generateClusterMetadataManifest( + "cluster-uuid2", + clusterUUIDsPointers.get("cluster-uuid2"), + randomAlphaOfLength(10) + ); + mockBlobContainer(blobContainer2, clusterManifest2, Map.of()); + + final ClusterMetadataManifest clusterManifest3 = generateClusterMetadataManifest( + "cluster-uuid3", + clusterUUIDsPointers.get("cluster-uuid3"), + randomAlphaOfLength(10) + ); + mockBlobContainer(blobContainer3, clusterManifest3, Map.of()); + + when(blobStore.blobContainer(ArgumentMatchers.any())).thenReturn( + uuidBlobContainer, + blobContainer1, + blobContainer1, + blobContainer2, + blobContainer2, + blobContainer3, + blobContainer3 + ); + when(blobStoreRepository.getCompressor()).thenReturn(new DeflateCompressor()); + } + + private ClusterMetadataManifest generateClusterMetadataManifest(String clusterUUID, String previousClusterUUID, String stateUUID) { + return ClusterMetadataManifest.builder() + .indices(List.of()) + .clusterTerm(1L) + .stateVersion(1L) + .stateUUID(stateUUID) + .clusterUUID(clusterUUID) + .nodeId("nodeA") + .opensearchVersion(VersionUtils.randomOpenSearchVersion(random())) + .previousClusterUUID(previousClusterUUID) + .committed(true) + .build(); + } + private BlobContainer mockBlobStoreObjects() { return mockBlobStoreObjects(BlobContainer.class); } @@ -542,6 +644,14 @@ private BlobContainer mockBlobStoreObjects(Class blobCo return blobContainer; } + private void mockBlobContainerForClusterUUIDs(BlobContainer blobContainer, Set clusterUUIDs) throws IOException { + Map blobContainerMap = new HashMap<>(); + for (String clusterUUID : clusterUUIDs) { + blobContainerMap.put(clusterUUID, mockBlobStoreObjects()); + } + when(blobContainer.children()).thenReturn(blobContainerMap); + } + private void mockBlobContainer( BlobContainer blobContainer, ClusterMetadataManifest clusterMetadataManifest,