From d7052d57e6624d17e662296b9eaaf632215a75ab Mon Sep 17 00:00:00 2001 From: Sooraj Sinha Date: Wed, 6 Sep 2023 21:59:23 +0530 Subject: [PATCH] Restore cluster metadata during bootstrap Signed-off-by: Sooraj Sinha --- .../opensearch/gateway/GatewayMetaState.java | 42 ++-- .../remote/RemoteClusterStateService.java | 64 ++++-- .../main/java/org/opensearch/node/Node.java | 5 +- .../GatewayMetaStatePersistedStateTests.java | 185 ++++++++++++++++-- .../RemoteClusterStateServiceTests.java | 136 ++++++++++--- .../gateway/MockGatewayMetaState.java | 45 ++--- 6 files changed, 380 insertions(+), 97 deletions(-) diff --git a/server/src/main/java/org/opensearch/gateway/GatewayMetaState.java b/server/src/main/java/org/opensearch/gateway/GatewayMetaState.java index 4df18e4014c3d..eb959a7b198f7 100644 --- a/server/src/main/java/org/opensearch/gateway/GatewayMetaState.java +++ b/server/src/main/java/org/opensearch/gateway/GatewayMetaState.java @@ -63,6 +63,8 @@ import org.opensearch.env.NodeMetadata; import org.opensearch.gateway.remote.ClusterMetadataManifest; import org.opensearch.gateway.remote.RemoteClusterStateService; +import org.opensearch.index.recovery.RemoteStoreRestoreService; +import org.opensearch.index.recovery.RemoteStoreRestoreService.RemoteRestoreResult; import org.opensearch.node.Node; import org.opensearch.plugins.MetadataUpgrader; import org.opensearch.repositories.RepositoryMissingException; @@ -126,7 +128,8 @@ public void start( MetadataUpgrader metadataUpgrader, PersistedClusterStateService persistedClusterStateService, RemoteClusterStateService remoteClusterStateService, - PersistedStateRegistry persistedStateRegistry + PersistedStateRegistry persistedStateRegistry, + RemoteStoreRestoreService remoteStoreRestoreService ) { assert this.persistedStateRegistry == null : "Persisted state registry should only be set once"; this.persistedStateRegistry = persistedStateRegistry; @@ -154,7 +157,7 @@ public void start( PersistedState remotePersistedState = null; boolean success = false; try { - final ClusterState clusterState = prepareInitialClusterState( + ClusterState clusterState = prepareInitialClusterState( transportService, clusterService, ClusterState.builder(ClusterName.CLUSTER_NAME_SETTING.get(settings)) @@ -164,10 +167,28 @@ public void start( ); if (DiscoveryNode.isClusterManagerNode(settings)) { - persistedState = new LucenePersistedState(persistedClusterStateService, currentTerm, clusterState); if (REMOTE_CLUSTER_STATE_ENABLED_SETTING.get(settings) == true) { + // If the cluster UUID loaded from local is unknown (_na_) then fetch the best state from remote + // If there is no valid state on remote, continue with initial empty state + // If there is a valid state, then restore index metadata using this state + if (ClusterState.UNKNOWN_UUID.equals(clusterState.metadata().clusterUUID())) { + String previousClusterUUID = remoteClusterStateService.getLatestClusterUUID( + clusterState.getClusterName().value() + ); + if (!ClusterState.UNKNOWN_UUID.equals(previousClusterUUID)) { + // Load state from remote + final RemoteRestoreResult remoteRestoreResult = remoteStoreRestoreService.restore( + clusterState, + previousClusterUUID, + false, + new String[] {} + ); + clusterState = remoteRestoreResult.getClusterState(); + } + } remotePersistedState = new RemotePersistedState(remoteClusterStateService); } + persistedState = new LucenePersistedState(persistedClusterStateService, currentTerm, clusterState); } else { persistedState = new AsyncLucenePersistedState( settings, @@ -651,12 +672,6 @@ public void setCurrentTerm(long currentTerm) { @Override public void setLastAcceptedState(ClusterState clusterState) { try { - if (lastAcceptedState == null || lastAcceptedState.blocks().hasGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK)) { - // On the initial bootstrap, repository will not be available. So we do not persist the cluster state and bail out. - logger.info("Cluster is not yet ready to publish state to remote store"); - lastAcceptedState = clusterState; - return; - } final ClusterMetadataManifest manifest; if (shouldWriteFullClusterState(clusterState)) { manifest = remoteClusterStateService.writeFullMetadata(clusterState); @@ -706,13 +721,8 @@ private boolean shouldWriteFullClusterState(ClusterState clusterState) { @Override public void markLastAcceptedStateAsCommitted() { try { - if (lastAcceptedState == null - || lastAcceptedManifest == null - || lastAcceptedState.blocks().hasGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK)) { - // On the initial bootstrap, repository will not be available. So we do not persist the cluster state and bail out. - logger.trace("Cluster is not yet ready to publish state to remote store"); - return; - } + assert lastAcceptedState != null : "Last accepted state is not present"; + assert lastAcceptedManifest != null : "Last accepted manifest is not present"; final ClusterMetadataManifest committedManifest = remoteClusterStateService.markLastStateAsCommitted( lastAcceptedState, lastAcceptedManifest 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 2c38bec4802d7..98747b35b244b 100644 --- a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java +++ b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java @@ -44,6 +44,8 @@ import java.util.Locale; import java.util.Map; import java.util.Objects; +import java.util.Optional; +import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.function.Function; @@ -132,19 +134,25 @@ public RemoteClusterStateService( */ @Nullable public ClusterMetadataManifest writeFullMetadata(ClusterState clusterState) throws IOException { + // should fetch the previous cluster UUID before writing full cluster state. + // Whenever a new election happens, a new master will be elected and that master might have stale previous UUID final long startTimeNanos = relativeTimeNanosSupplier.getAsLong(); if (clusterState.nodes().isLocalNodeElectedClusterManager() == false) { logger.error("Local node is not elected cluster manager. Exiting"); return null; } - ensureRepositorySet(); + + final String previousClusterUUID = fetchPreviousClusterUUID( + clusterState.getClusterName().value(), + clusterState.metadata().clusterUUID() + ); // any validations before/after upload ? final List allUploadedIndexMetadata = writeIndexMetadataParallel( clusterState, new ArrayList<>(clusterState.metadata().indices().values()) ); - final ClusterMetadataManifest manifest = uploadManifest(clusterState, allUploadedIndexMetadata, false); + final ClusterMetadataManifest manifest = uploadManifest(clusterState, allUploadedIndexMetadata, previousClusterUUID, false); final long durationMillis = TimeValue.nsecToMSec(relativeTimeNanosSupplier.getAsLong() - startTimeNanos); if (durationMillis >= slowWriteLoggingThreshold.getMillis()) { logger.warn( @@ -221,7 +229,12 @@ public ClusterMetadataManifest writeIncrementalMetadata( for (String removedIndexName : previousStateIndexMetadataVersionByName.keySet()) { allUploadedIndexMetadata.remove(removedIndexName); } - final ClusterMetadataManifest manifest = uploadManifest(clusterState, new ArrayList<>(allUploadedIndexMetadata.values()), false); + final ClusterMetadataManifest manifest = uploadManifest( + clusterState, + new ArrayList<>(allUploadedIndexMetadata.values()), + previousManifest.getPreviousClusterUUID(), + false + ); final long durationMillis = TimeValue.nsecToMSec(relativeTimeNanosSupplier.getAsLong() - startTimeNanos); if (durationMillis >= slowWriteLoggingThreshold.getMillis()) { logger.warn( @@ -364,7 +377,7 @@ public ClusterMetadataManifest markLastStateAsCommitted(ClusterState clusterStat } assert clusterState != null : "Last accepted cluster state is not set"; assert previousManifest != null : "Last cluster metadata manifest is not set"; - return uploadManifest(clusterState, previousManifest.getIndices(), true); + return uploadManifest(clusterState, previousManifest.getIndices(), previousManifest.getPreviousClusterUUID(), true); } public ClusterState getLatestClusterState(String clusterUUID) { @@ -379,11 +392,7 @@ public void close() throws IOException { } } - // Visible for testing - void ensureRepositorySet() { - if (blobStoreRepository != null) { - return; - } + public void start() { assert REMOTE_CLUSTER_STATE_ENABLED_SETTING.get(settings) == true : "Remote cluster state is not enabled"; final String remoteStoreRepo = REMOTE_CLUSTER_STATE_REPOSITORY_SETTING.get(settings); assert remoteStoreRepo != null : "Remote Cluster State repository is not configured"; @@ -395,6 +404,7 @@ void ensureRepositorySet() { private ClusterMetadataManifest uploadManifest( ClusterState clusterState, List uploadedIndexMetadata, + String previousClusterUUID, boolean committed ) throws IOException { synchronized (this) { @@ -407,7 +417,8 @@ private ClusterMetadataManifest uploadManifest( Version.CURRENT, nodeId, committed, - uploadedIndexMetadata + uploadedIndexMetadata, + previousClusterUUID ); writeMetadataManifest(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID(), manifest, manifestFileName); return manifest; @@ -420,6 +431,11 @@ private void writeMetadataManifest(String clusterName, String clusterUUID, Clust CLUSTER_METADATA_MANIFEST_FORMAT.write(uploadManifest, metadataManifestContainer, fileName, blobStoreRepository.getCompressor()); } + private String fetchPreviousClusterUUID(String clusterName, String clusterUUID) { + final Optional latestManifest = getLatestClusterMetadataManifest(clusterName, clusterUUID); + return latestManifest.isPresent() ? latestManifest.get().getPreviousClusterUUID() : ClusterState.UNKNOWN_UUID; + } + private BlobContainer indexMetadataContainer(String clusterName, String clusterUUID, String indexUUID) { // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/index/ftqsCnn9TgOX return blobStoreRepository.blobStore() @@ -467,12 +483,15 @@ private static String indexMetadataFileName(IndexMetadata indexMetadata) { * @return {@code Map} latest IndexUUID to IndexMetadata map */ public Map getLatestIndexMetadata(String clusterName, String clusterUUID) throws IOException { - ensureRepositorySet(); + start(); Map remoteIndexMetadata = new HashMap<>(); - ClusterMetadataManifest clusterMetadataManifest = getLatestClusterMetadataManifest(clusterName, clusterUUID); - assert Objects.equals(clusterUUID, clusterMetadataManifest.getClusterUUID()) + Optional clusterMetadataManifest = getLatestClusterMetadataManifest(clusterName, clusterUUID); + if (!clusterMetadataManifest.isPresent()) { + throw new IllegalStateException("Latest index metadata is not present for the provided clusterUUID"); + } + assert Objects.equals(clusterUUID, clusterMetadataManifest.get().getClusterUUID()) : "Corrupt ClusterMetadataManifest found. Cluster UUID mismatch."; - for (UploadedIndexMetadata uploadedIndexMetadata : clusterMetadataManifest.getIndices()) { + for (UploadedIndexMetadata uploadedIndexMetadata : clusterMetadataManifest.get().getIndices()) { IndexMetadata indexMetadata = getIndexMetadata(clusterName, clusterUUID, uploadedIndexMetadata); remoteIndexMetadata.put(uploadedIndexMetadata.getIndexUUID(), indexMetadata); } @@ -508,9 +527,12 @@ private IndexMetadata getIndexMetadata(String clusterName, String clusterUUID, U * @param clusterName name of the cluster * @return ClusterMetadataManifest */ - public ClusterMetadataManifest getLatestClusterMetadataManifest(String clusterName, String clusterUUID) { - String latestManifestFileName = getLatestManifestFileName(clusterName, clusterUUID); - return fetchRemoteClusterMetadataManifest(clusterName, clusterUUID, latestManifestFileName); + public Optional getLatestClusterMetadataManifest(String clusterName, String clusterUUID) { + Optional latestManifestFileName = getLatestManifestFileName(clusterName, clusterUUID); + if (latestManifestFileName.isPresent()) { + return Optional.of(fetchRemoteClusterMetadataManifest(clusterName, clusterUUID, latestManifestFileName.get())); + } + return Optional.empty(); } /** @@ -519,7 +541,7 @@ public ClusterMetadataManifest getLatestClusterMetadataManifest(String clusterNa * @param clusterName name of the cluster * @return latest ClusterMetadataManifest filename */ - private String getLatestManifestFileName(String clusterName, String clusterUUID) throws IllegalStateException { + private Optional getLatestManifestFileName(String clusterName, String clusterUUID) throws IllegalStateException { try { /** * {@link BlobContainer#listBlobsByPrefixInSortedOrder} will get the latest manifest file @@ -532,13 +554,13 @@ private String getLatestManifestFileName(String clusterName, String clusterUUID) BlobContainer.BlobNameSortOrder.LEXICOGRAPHIC ); if (manifestFilesMetadata != null && !manifestFilesMetadata.isEmpty()) { - return manifestFilesMetadata.get(0).name(); + return Optional.of(manifestFilesMetadata.get(0).name()); } } catch (IOException e) { throw new IllegalStateException("Error while fetching latest manifest file for remote cluster state", e); } - - throw new IllegalStateException(String.format(Locale.ROOT, "Remote Cluster State not found - %s", clusterUUID)); + logger.info("No manifest file present in remote store for cluster name: {}, cluster UUID: {}", clusterName, clusterUUID); + return Optional.empty(); } /** diff --git a/server/src/main/java/org/opensearch/node/Node.java b/server/src/main/java/org/opensearch/node/Node.java index e2365adb440cc..cf65317cdb2b7 100644 --- a/server/src/main/java/org/opensearch/node/Node.java +++ b/server/src/main/java/org/opensearch/node/Node.java @@ -1334,6 +1334,8 @@ public Node start() throws NodeValidationException { injector.getInstance(PeerRecoverySourceService.class).start(); injector.getInstance(SegmentReplicationSourceService.class).start(); + final RemoteClusterStateService remoteClusterStateService = injector.getInstance(RemoteClusterStateService.class); + remoteClusterStateService.start(); // Load (and maybe upgrade) the metadata stored on disk final GatewayMetaState gatewayMetaState = injector.getInstance(GatewayMetaState.class); gatewayMetaState.start( @@ -1345,7 +1347,8 @@ public Node start() throws NodeValidationException { injector.getInstance(MetadataUpgrader.class), injector.getInstance(PersistedClusterStateService.class), injector.getInstance(RemoteClusterStateService.class), - injector.getInstance(PersistedStateRegistry.class) + injector.getInstance(PersistedStateRegistry.class), + injector.getInstance(RemoteStoreRestoreService.class) ); if (Assertions.ENABLED) { try { diff --git a/server/src/test/java/org/opensearch/gateway/GatewayMetaStatePersistedStateTests.java b/server/src/test/java/org/opensearch/gateway/GatewayMetaStatePersistedStateTests.java index c293768151fa4..a3ebb5ef7ca5b 100644 --- a/server/src/test/java/org/opensearch/gateway/GatewayMetaStatePersistedStateTests.java +++ b/server/src/test/java/org/opensearch/gateway/GatewayMetaStatePersistedStateTests.java @@ -42,6 +42,7 @@ import org.opensearch.cluster.coordination.CoordinationMetadata; import org.opensearch.cluster.coordination.CoordinationMetadata.VotingConfigExclusion; import org.opensearch.cluster.coordination.CoordinationState; +import org.opensearch.cluster.coordination.CoordinationState.PersistedState; import org.opensearch.cluster.coordination.PersistedStateRegistry; import org.opensearch.cluster.coordination.PersistedStateRegistry.PersistedStateType; import org.opensearch.cluster.metadata.IndexMetadata; @@ -51,6 +52,7 @@ import org.opensearch.cluster.node.DiscoveryNodeRole; import org.opensearch.cluster.node.DiscoveryNodes; import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.collect.Tuple; import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.settings.Settings; import org.opensearch.common.util.BigArrays; @@ -63,8 +65,11 @@ import org.opensearch.env.NodeEnvironment; import org.opensearch.env.TestEnvironment; import org.opensearch.gateway.GatewayMetaState.RemotePersistedState; +import org.opensearch.gateway.PersistedClusterStateService.Writer; import org.opensearch.gateway.remote.ClusterMetadataManifest; import org.opensearch.gateway.remote.RemoteClusterStateService; +import org.opensearch.index.recovery.RemoteStoreRestoreService; +import org.opensearch.index.recovery.RemoteStoreRestoreService.RemoteRestoreResult; import org.opensearch.node.Node; import org.opensearch.repositories.RepositoriesService; import org.opensearch.test.OpenSearchTestCase; @@ -85,13 +90,18 @@ import org.mockito.Mockito; +import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_INDEX_UUID; import static org.opensearch.test.NodeRoles.nonClusterManagerNode; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.not; import static org.hamcrest.Matchers.nullValue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; public class GatewayMetaStatePersistedStateTests extends OpenSearchTestCase { @@ -473,7 +483,8 @@ public void testDataOnlyNodePersistence() throws Exception { null, persistedClusterStateService, remoteClusterStateServiceSupplier.get(), - new PersistedStateRegistry() + new PersistedStateRegistry(), + null ); final CoordinationState.PersistedState persistedState = gateway.getPersistedState(); assertThat(persistedState, instanceOf(GatewayMetaState.AsyncLucenePersistedState.class)); @@ -706,7 +717,7 @@ public void testRemotePersistedState() throws IOException { ); remotePersistedState.setLastAcceptedState(clusterState); - Mockito.verify(remoteClusterStateService, times(0)).writeFullMetadata(Mockito.any()); + Mockito.verify(remoteClusterStateService).writeFullMetadata(clusterState); assertThat(remotePersistedState.getLastAcceptedState(), equalTo(clusterState)); assertThat(remotePersistedState.getCurrentTerm(), equalTo(clusterTerm)); @@ -717,7 +728,7 @@ public void testRemotePersistedState() throws IOException { ); remotePersistedState.setLastAcceptedState(secondClusterState); - Mockito.verify(remoteClusterStateService, times(1)).writeFullMetadata(Mockito.any()); + Mockito.verify(remoteClusterStateService, times(1)).writeFullMetadata(secondClusterState); assertThat(remotePersistedState.getLastAcceptedState(), equalTo(secondClusterState)); assertThat(remotePersistedState.getCurrentTerm(), equalTo(clusterTerm)); @@ -742,20 +753,19 @@ public void testRemotePersistedStateExceptionOnFullStateUpload() throws IOExcept Metadata.builder().coordinationMetadata(CoordinationMetadata.builder().term(clusterTerm).build()).build() ); - remotePersistedState.setLastAcceptedState(clusterState); - - final ClusterState secondClusterState = createClusterState( - randomNonNegativeLong(), - Metadata.builder().coordinationMetadata(CoordinationMetadata.builder().term(clusterTerm).build()).build() - ); - - assertThrows(OpenSearchException.class, () -> remotePersistedState.setLastAcceptedState(secondClusterState)); + assertThrows(OpenSearchException.class, () -> remotePersistedState.setLastAcceptedState(clusterState)); } public void testGatewayForRemoteState() throws IOException { MockGatewayMetaState gateway = null; try { - gateway = new MockGatewayMetaState(localNode, bigArrays); + RemoteClusterStateService remoteClusterStateService = mock(RemoteClusterStateService.class); + when(remoteClusterStateService.getLatestClusterUUID("test-cluster")).thenReturn("test-cluster-uuid"); + RemoteStoreRestoreService remoteStoreRestoreService = mock(RemoteStoreRestoreService.class); + when(remoteStoreRestoreService.restore(any(), any(), anyBoolean(), any())).thenReturn( + RemoteRestoreResult.build("test-cluster-uuid", null, ClusterState.EMPTY_STATE) + ); + gateway = new MockGatewayMetaState(localNode, bigArrays, remoteClusterStateService, remoteStoreRestoreService); final PersistedStateRegistry persistedStateRegistry = persistedStateRegistry(); final Settings settingWithRemoteStateEnabled = Settings.builder() .put(settings) @@ -777,6 +787,157 @@ public void testGatewayForRemoteState() throws IOException { } } + public void testGatewayForRemoteStateForInitialBootstrap() throws IOException { + MockGatewayMetaState gateway = null; + try { + final RemoteClusterStateService remoteClusterStateService = mock(RemoteClusterStateService.class); + when(remoteClusterStateService.getLatestClusterUUID(clusterName.value())).thenReturn(ClusterState.UNKNOWN_UUID); + + final RemoteStoreRestoreService remoteStoreRestoreService = mock(RemoteStoreRestoreService.class); + when(remoteStoreRestoreService.restore(any(), any(), anyBoolean(), any())).thenReturn( + RemoteRestoreResult.build("test-cluster-uuid", null, ClusterState.EMPTY_STATE) + ); + final PersistedStateRegistry persistedStateRegistry = persistedStateRegistry(); + gateway = newGatewayForRemoteState( + remoteClusterStateService, + remoteStoreRestoreService, + persistedStateRegistry, + ClusterState.EMPTY_STATE + ); + final CoordinationState.PersistedState lucenePersistedState = gateway.getPersistedState(); + PersistedState remotePersistedState = persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE); + verify(remoteClusterStateService).getLatestClusterUUID(Mockito.any()); // change this + verifyNoInteractions(remoteStoreRestoreService); + assertThat(remotePersistedState.getLastAcceptedState(), nullValue()); + assertThat(lucenePersistedState.getLastAcceptedState().metadata(), equalTo(ClusterState.EMPTY_STATE.metadata())); + } finally { + IOUtils.close(gateway); + } + } + + public void testGatewayForRemoteStateForNodeReplacement() throws IOException { + MockGatewayMetaState gateway = null; + try { + final RemoteClusterStateService remoteClusterStateService = mock(RemoteClusterStateService.class); + when(remoteClusterStateService.getLatestClusterUUID("test-cluster")).thenReturn("test-cluster-uuid"); + final ClusterState previousState = createClusterState( + randomNonNegativeLong(), + Metadata.builder() + .coordinationMetadata(CoordinationMetadata.builder().term(randomLong()).build()) + .put( + IndexMetadata.builder("test-index1") + .settings(settings(Version.CURRENT).put(SETTING_INDEX_UUID, randomAlphaOfLength(10))) + .numberOfShards(5) + .numberOfReplicas(1) + .build(), + false + ) + .clusterUUID(randomAlphaOfLength(10)) + .build() + ); + when(remoteClusterStateService.getLatestClusterUUID(clusterName.value())).thenReturn(previousState.metadata().clusterUUID()); + + final RemoteStoreRestoreService remoteStoreRestoreService = mock(RemoteStoreRestoreService.class); + when(remoteStoreRestoreService.restore(any(), any(), anyBoolean(), any())).thenReturn( + RemoteRestoreResult.build("test-cluster-uuid", null, previousState) + ); + final PersistedStateRegistry persistedStateRegistry = persistedStateRegistry(); + gateway = newGatewayForRemoteState( + remoteClusterStateService, + remoteStoreRestoreService, + persistedStateRegistry, + ClusterState.EMPTY_STATE + ); + final CoordinationState.PersistedState lucenePersistedState = gateway.getPersistedState(); + PersistedState remotePersistedState = persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE); + verify(remoteClusterStateService).getLatestClusterUUID(Mockito.any()); // change this + verify(remoteStoreRestoreService).restore(any(), any(), anyBoolean(), any()); + assertThat(remotePersistedState.getLastAcceptedState(), nullValue()); + assertThat(lucenePersistedState.getLastAcceptedState().metadata(), equalTo(previousState.metadata())); + } finally { + IOUtils.close(gateway); + } + } + + public void testGatewayForRemoteStateForNodeReboot() throws IOException { + MockGatewayMetaState gateway = null; + try { + final RemoteClusterStateService remoteClusterStateService = mock(RemoteClusterStateService.class); + final RemoteStoreRestoreService remoteStoreRestoreService = mock(RemoteStoreRestoreService.class); + final PersistedStateRegistry persistedStateRegistry = persistedStateRegistry(); + final IndexMetadata indexMetadata = IndexMetadata.builder("test-index1") + .settings(settings(Version.CURRENT).put(SETTING_INDEX_UUID, randomAlphaOfLength(10))) + .numberOfShards(5) + .numberOfReplicas(1) + .build(); + final ClusterState clusterState = createClusterState( + randomNonNegativeLong(), + Metadata.builder() + .coordinationMetadata(CoordinationMetadata.builder().term(randomLong()).build()) + .put(indexMetadata, false) + .clusterUUID(randomAlphaOfLength(10)) + .build() + ); + gateway = newGatewayForRemoteState(remoteClusterStateService, remoteStoreRestoreService, persistedStateRegistry, clusterState); + final CoordinationState.PersistedState lucenePersistedState = gateway.getPersistedState(); + PersistedState remotePersistedState = persistedStateRegistry.getPersistedState(PersistedStateType.REMOTE); + verifyNoInteractions(remoteClusterStateService); + verifyNoInteractions(remoteStoreRestoreService); + assertThat(remotePersistedState.getLastAcceptedState(), nullValue()); + logger.info("lucene state metadata: {}", lucenePersistedState.getLastAcceptedState().toString()); + logger.info("initial metadata: {}", clusterState.toString()); + assertThat(lucenePersistedState.getLastAcceptedState().metadata().indices().size(), equalTo(1)); + assertThat(lucenePersistedState.getLastAcceptedState().metadata().indices().get("test-index1"), equalTo(indexMetadata)); + } finally { + IOUtils.close(gateway); + } + } + + private MockGatewayMetaState newGatewayForRemoteState( + RemoteClusterStateService remoteClusterStateService, + RemoteStoreRestoreService remoteStoreRestoreService, + PersistedStateRegistry persistedStateRegistry, + ClusterState currentState + ) throws IOException { + MockGatewayMetaState gateway = new MockGatewayMetaState(localNode, bigArrays); + final Settings settingWithRemoteStateEnabled = Settings.builder() + .put(settings) + .put(RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING.getKey(), true) + .build(); + final TransportService transportService = mock(TransportService.class); + ClusterService clusterService = mock(ClusterService.class); + when(clusterService.getClusterSettings()).thenReturn( + new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS) + ); + final PersistedClusterStateService persistedClusterStateService = new PersistedClusterStateService( + nodeEnvironment, + xContentRegistry(), + getBigArrays(), + new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), + () -> 0L + ); + if (!ClusterState.EMPTY_STATE.equals(currentState)) { + Writer writer = persistedClusterStateService.createWriter(); + writer.writeFullStateAndCommit(currentState.term(), currentState); + writer.close(); + } + final MetaStateService metaStateService = mock(MetaStateService.class); + when(metaStateService.loadFullState()).thenReturn(new Tuple<>(Manifest.empty(), currentState.metadata())); + gateway.start( + settingWithRemoteStateEnabled, + transportService, + clusterService, + metaStateService, + null, + null, + persistedClusterStateService, + remoteClusterStateService, + persistedStateRegistry, + remoteStoreRestoreService + ); + return gateway; + } + private static BigArrays getBigArrays() { return usually() ? BigArrays.NON_RECYCLING_INSTANCE 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 1f4c32b59f183..379d017b93b5a 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java @@ -52,6 +52,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.function.Supplier; import org.mockito.ArgumentCaptor; @@ -117,22 +118,21 @@ public void testFailInitializationWhenRemoteStateDisabled() throws IOException { ); } - public void testFailWriteFullMetadataWhenRepositoryNotSet() { - final ClusterState clusterState = generateClusterStateWithOneIndex().nodes(nodesWithLocalNodeClusterManager()).build(); + public void testFailInitializeWhenRepositoryNotSet() { doThrow(new RepositoryMissingException("repository missing")).when(repositoriesService).repository("remote_store_repository"); - assertThrows(RepositoryMissingException.class, () -> remoteClusterStateService.writeFullMetadata(clusterState)); + assertThrows(RepositoryMissingException.class, () -> remoteClusterStateService.start()); } public void testFailWriteFullMetadataWhenNotBlobRepository() { final FilterRepository filterRepository = mock(FilterRepository.class); when(repositoriesService.repository("remote_store_repository")).thenReturn(filterRepository); - final ClusterState clusterState = generateClusterStateWithOneIndex().nodes(nodesWithLocalNodeClusterManager()).build(); - assertThrows(AssertionError.class, () -> remoteClusterStateService.writeFullMetadata(clusterState)); + assertThrows(AssertionError.class, () -> remoteClusterStateService.start()); } public void testWriteFullMetadataSuccess() throws IOException { final ClusterState clusterState = generateClusterStateWithOneIndex().nodes(nodesWithLocalNodeClusterManager()).build(); mockBlobStoreObjects(); + remoteClusterStateService.start(); final ClusterMetadataManifest manifest = remoteClusterStateService.writeFullMetadata(clusterState); final UploadedIndexMetadata uploadedIndexMetadata = new UploadedIndexMetadata("test-index", "index-uuid", "metadata-filename"); List indices = List.of(uploadedIndexMetadata); @@ -167,6 +167,7 @@ public void testWriteFullMetadataInParallelSuccess() throws IOException { return null; }).when(container).asyncBlobUpload(writeContextArgumentCaptor.capture(), actionListenerArgumentCaptor.capture()); + remoteClusterStateService.start(); final ClusterMetadataManifest manifest = remoteClusterStateService.writeFullMetadata(clusterState); final UploadedIndexMetadata uploadedIndexMetadata = new UploadedIndexMetadata("test-index", "index-uuid", "metadata-filename"); @@ -221,6 +222,7 @@ public void testWriteFullMetadataInParallelFailure() throws IOException { return null; }).when(container).asyncBlobUpload(any(WriteContext.class), actionListenerArgumentCaptor.capture()); + remoteClusterStateService.start(); assertThrows( RemoteClusterStateService.IndexMetadataTransferException.class, () -> remoteClusterStateService.writeFullMetadata(clusterState) @@ -229,6 +231,7 @@ public void testWriteFullMetadataInParallelFailure() throws IOException { public void testFailWriteIncrementalMetadataNonClusterManagerNode() throws IOException { final ClusterState clusterState = generateClusterStateWithOneIndex().build(); + remoteClusterStateService.start(); final ClusterMetadataManifest manifest = remoteClusterStateService.writeIncrementalMetadata(clusterState, clusterState, null); Assert.assertThat(manifest, nullValue()); } @@ -255,7 +258,7 @@ public void testWriteIncrementalMetadataSuccess() throws IOException { final ClusterMetadataManifest previousManifest = ClusterMetadataManifest.builder().indices(Collections.emptyList()).build(); - remoteClusterStateService.ensureRepositorySet(); + remoteClusterStateService.start(); final ClusterMetadataManifest manifest = remoteClusterStateService.writeIncrementalMetadata( previousClusterState, clusterState, @@ -294,7 +297,7 @@ public void testReadLatestMetadataManifestFailedIOException() throws IOException ) ).thenThrow(IOException.class); - remoteClusterStateService.ensureRepositorySet(); + remoteClusterStateService.start(); Exception e = assertThrows( IllegalStateException.class, () -> remoteClusterStateService.getLatestClusterMetadataManifest( @@ -317,15 +320,12 @@ public void testReadLatestMetadataManifestFailedNoManifestFileInRemote() throws ) ).thenReturn(List.of()); - remoteClusterStateService.ensureRepositorySet(); - Exception e = assertThrows( - IllegalStateException.class, - () -> remoteClusterStateService.getLatestClusterMetadataManifest( - clusterState.getClusterName().value(), - clusterState.metadata().clusterUUID() - ) + remoteClusterStateService.start(); + Optional manifest = remoteClusterStateService.getLatestClusterMetadataManifest( + clusterState.getClusterName().value(), + clusterState.metadata().clusterUUID() ); - assertEquals(e.getMessage(), "Remote Cluster State not found - " + clusterState.metadata().clusterUUID()); + assertEquals(manifest, Optional.empty()); } public void testReadLatestMetadataManifestFailedManifestFileRemoveAfterFetchInRemote() throws IOException { @@ -342,7 +342,7 @@ public void testReadLatestMetadataManifestFailedManifestFileRemoveAfterFetchInRe ).thenReturn(Arrays.asList(blobMetadata)); when(blobContainer.readBlob("manifestFileName")).thenThrow(FileNotFoundException.class); - remoteClusterStateService.ensureRepositorySet(); + remoteClusterStateService.start(); Exception e = assertThrows( IllegalStateException.class, () -> remoteClusterStateService.getLatestClusterMetadataManifest( @@ -368,7 +368,7 @@ public void testReadLatestMetadataManifestSuccessButNoIndexMetadata() throws IOE BlobContainer blobContainer = mockBlobStoreObjects(); mockBlobContainer(blobContainer, expectedManifest, Map.of()); - remoteClusterStateService.ensureRepositorySet(); + remoteClusterStateService.start(); assertEquals( remoteClusterStateService.getLatestIndexMetadata(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID()) .size(), @@ -394,7 +394,7 @@ public void testReadLatestMetadataManifestSuccessButIndexMetadataFetchIOExceptio mockBlobContainer(blobContainer, expectedManifest, Map.of()); when(blobContainer.readBlob(uploadedIndexMetadata.getUploadedFilename() + ".dat")).thenThrow(FileNotFoundException.class); - remoteClusterStateService.ensureRepositorySet(); + remoteClusterStateService.start(); Exception e = assertThrows( IllegalStateException.class, () -> remoteClusterStateService.getLatestIndexMetadata( @@ -421,11 +421,11 @@ public void testReadLatestMetadataManifestSuccess() throws IOException { .build(); mockBlobContainer(mockBlobStoreObjects(), expectedManifest, new HashMap<>()); - remoteClusterStateService.ensureRepositorySet(); + remoteClusterStateService.start(); final ClusterMetadataManifest manifest = remoteClusterStateService.getLatestClusterMetadataManifest( clusterState.getClusterName().value(), clusterState.metadata().clusterUUID() - ); + ).get(); assertThat(manifest.getIndices().size(), is(1)); assertThat(manifest.getIndices().get(0).getIndexName(), is(uploadedIndexMetadata.getIndexName())); @@ -439,7 +439,7 @@ public void testReadLatestMetadataManifestSuccess() throws IOException { public void testReadLatestIndexMetadataSuccess() throws IOException { final ClusterState clusterState = generateClusterStateWithOneIndex().nodes(nodesWithLocalNodeClusterManager()).build(); - remoteClusterStateService.ensureRepositorySet(); + remoteClusterStateService.start(); final Index index = new Index("test-index", "index-uuid"); String fileName = "metadata-" + index.getUUID(); @@ -480,7 +480,7 @@ public void testReadLatestIndexMetadataSuccess() throws IOException { public void testMarkLastStateAsCommittedSuccess() throws IOException { final ClusterState clusterState = generateClusterStateWithOneIndex().nodes(nodesWithLocalNodeClusterManager()).build(); mockBlobStoreObjects(); - remoteClusterStateService.ensureRepositorySet(); + remoteClusterStateService.start(); final UploadedIndexMetadata uploadedIndexMetadata = new UploadedIndexMetadata("test-index", "index-uuid", "metadata-filename"); List indices = List.of(uploadedIndexMetadata); final ClusterMetadataManifest previousManifest = ClusterMetadataManifest.builder().indices(indices).build(); @@ -506,6 +506,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.start(); + 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.start(); + 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); } diff --git a/test/framework/src/main/java/org/opensearch/gateway/MockGatewayMetaState.java b/test/framework/src/main/java/org/opensearch/gateway/MockGatewayMetaState.java index 14d646cf84a0a..dea205619ce95 100644 --- a/test/framework/src/main/java/org/opensearch/gateway/MockGatewayMetaState.java +++ b/test/framework/src/main/java/org/opensearch/gateway/MockGatewayMetaState.java @@ -46,14 +46,12 @@ import org.opensearch.core.xcontent.NamedXContentRegistry; import org.opensearch.env.NodeEnvironment; import org.opensearch.gateway.remote.RemoteClusterStateService; +import org.opensearch.index.recovery.RemoteStoreRestoreService; import org.opensearch.plugins.MetadataUpgrader; -import org.opensearch.repositories.RepositoriesService; import org.opensearch.threadpool.ThreadPool; import org.opensearch.transport.TransportService; import java.io.IOException; -import java.util.Collections; -import java.util.function.Supplier; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -67,10 +65,26 @@ public class MockGatewayMetaState extends GatewayMetaState { private final DiscoveryNode localNode; private final BigArrays bigArrays; + private final RemoteClusterStateService remoteClusterStateService; + private final RemoteStoreRestoreService remoteStoreRestoreService; public MockGatewayMetaState(DiscoveryNode localNode, BigArrays bigArrays) { this.localNode = localNode; this.bigArrays = bigArrays; + this.remoteClusterStateService = mock(RemoteClusterStateService.class); + this.remoteStoreRestoreService = mock(RemoteStoreRestoreService.class); + } + + public MockGatewayMetaState( + DiscoveryNode localNode, + BigArrays bigArrays, + RemoteClusterStateService remoteClusterStateService, + RemoteStoreRestoreService remoteStoreRestoreService + ) { + this.localNode = localNode; + this.bigArrays = bigArrays; + this.remoteClusterStateService = remoteClusterStateService; + this.remoteStoreRestoreService = remoteStoreRestoreService; } @Override @@ -107,26 +121,6 @@ public void start( } catch (IOException e) { throw new AssertionError(e); } - Supplier remoteClusterStateServiceSupplier = () -> { - if (RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING.get(settings) == true) { - return new RemoteClusterStateService( - nodeEnvironment.nodeId(), - () -> new RepositoriesService( - settings, - clusterService, - transportService, - Collections.emptyMap(), - Collections.emptyMap(), - transportService.getThreadPool() - ), - settings, - new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), - () -> 0L - ); - } else { - return null; - } - }; start( settings, transportService, @@ -141,8 +135,9 @@ public void start( new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS), () -> 0L ), - remoteClusterStateServiceSupplier.get(), - persistedStateRegistry + remoteClusterStateService, + persistedStateRegistry, + remoteStoreRestoreService ); } }