From 25a4f62103df4cc6451e6fd166521a04bc94a438 Mon Sep 17 00:00:00 2001 From: Shivansh Arora Date: Wed, 3 Apr 2024 14:13:40 +0530 Subject: [PATCH] Refactored RemoteClusterStateService into smaller files Signed-off-by: Shivansh Arora --- .../remote/RemoteClusterStateServiceIT.java | 12 +- .../RemoteStoreClusterStateRestoreIT.java | 7 +- .../cluster/metadata/TemplatesMetadata.java | 3 + .../common/settings/ClusterSettings.java | 9 +- .../remote/RemoteClusterStateService.java | 930 ++---------------- .../remote/RemoteClusterStateUtils.java | 99 ++ .../remote/RemoteGlobalMetadataManager.java | 334 +++++++ .../remote/RemoteIndexMetadataManager.java | 182 ++++ .../gateway/remote/RemoteManifestManager.java | 357 +++++++ .../remote/ClusterMetadataManifestTests.java | 26 +- .../RemoteClusterStateServiceTests.java | 190 +--- .../RemoteGlobalMetadataManagerTests.java | 52 + .../RemoteIndexMetadataManagerTests.java | 52 + .../remote/RemoteManifestManagerTests.java | 143 +++ 14 files changed, 1384 insertions(+), 1012 deletions(-) create mode 100644 server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateUtils.java create mode 100644 server/src/main/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManager.java create mode 100644 server/src/main/java/org/opensearch/gateway/remote/RemoteIndexMetadataManager.java create mode 100644 server/src/main/java/org/opensearch/gateway/remote/RemoteManifestManager.java create mode 100644 server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java create mode 100644 server/src/test/java/org/opensearch/gateway/remote/RemoteIndexMetadataManagerTests.java create mode 100644 server/src/test/java/org/opensearch/gateway/remote/RemoteManifestManagerTests.java diff --git a/server/src/internalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateServiceIT.java b/server/src/internalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateServiceIT.java index 61b34af5be3ba..4fc6550f2a3a6 100644 --- a/server/src/internalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateServiceIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/gateway/remote/RemoteClusterStateServiceIT.java @@ -27,13 +27,13 @@ import java.util.stream.Collectors; import static org.opensearch.cluster.metadata.IndexMetadata.SETTING_NUMBER_OF_REPLICAS; -import static org.opensearch.gateway.remote.RemoteClusterStateService.COORDINATION_METADATA; -import static org.opensearch.gateway.remote.RemoteClusterStateService.CUSTOM_METADATA; -import static org.opensearch.gateway.remote.RemoteClusterStateService.DELIMITER; -import static org.opensearch.gateway.remote.RemoteClusterStateService.METADATA_FILE_PREFIX; import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING; -import static org.opensearch.gateway.remote.RemoteClusterStateService.SETTING_METADATA; -import static org.opensearch.gateway.remote.RemoteClusterStateService.TEMPLATES_METADATA; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.METADATA_FILE_PREFIX; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.COORDINATION_METADATA; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.CUSTOM_METADATA; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.SETTING_METADATA; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.TEMPLATES_METADATA; @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0) public class RemoteClusterStateServiceIT extends RemoteStoreBaseIntegTestCase { diff --git a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreClusterStateRestoreIT.java b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreClusterStateRestoreIT.java index 8c8209b80bfd8..34d223f1dd14f 100644 --- a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreClusterStateRestoreIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreClusterStateRestoreIT.java @@ -36,6 +36,7 @@ import org.opensearch.gateway.remote.ClusterMetadataManifest; import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedIndexMetadata; import org.opensearch.gateway.remote.RemoteClusterStateService; +import org.opensearch.gateway.remote.RemoteClusterStateUtils; import org.opensearch.test.InternalTestCluster; import org.opensearch.test.OpenSearchIntegTestCase; @@ -320,9 +321,7 @@ public void testFullClusterRestoreManifestFilePointsToInvalidIndexMetadataPathTh // Step - 3 Delete index metadata file in remote try { Files.move( - segmentRepoPath.resolve( - RemoteClusterStateService.encodeString(clusterName) + "/cluster-state/" + prevClusterUUID + "/index" - ), + segmentRepoPath.resolve(RemoteClusterStateUtils.encodeString(clusterName) + "/cluster-state/" + prevClusterUUID + "/index"), segmentRepoPath.resolve("cluster-state/") ); } catch (IOException e) { @@ -348,7 +347,7 @@ public void testRemoteStateFullRestart() throws Exception { try { Files.move( segmentRepoPath.resolve( - RemoteClusterStateService.encodeString(clusterService().state().getClusterName().value()) + RemoteClusterStateUtils.encodeString(clusterService().state().getClusterName().value()) + "/cluster-state/" + prevClusterUUID + "/manifest" diff --git a/server/src/main/java/org/opensearch/cluster/metadata/TemplatesMetadata.java b/server/src/main/java/org/opensearch/cluster/metadata/TemplatesMetadata.java index 84c62c22799e8..2ffcd7fb6988e 100644 --- a/server/src/main/java/org/opensearch/cluster/metadata/TemplatesMetadata.java +++ b/server/src/main/java/org/opensearch/cluster/metadata/TemplatesMetadata.java @@ -9,6 +9,7 @@ package org.opensearch.cluster.metadata; import org.opensearch.cluster.AbstractDiffable; +import org.opensearch.common.annotation.PublicApi; import org.opensearch.core.common.io.stream.StreamOutput; import org.opensearch.core.xcontent.ToXContentFragment; import org.opensearch.core.xcontent.XContentBuilder; @@ -25,6 +26,7 @@ * * @opensearch.internal */ +@PublicApi(since = "1.0.0") public class TemplatesMetadata extends AbstractDiffable implements ToXContentFragment { public static TemplatesMetadata EMPTY_METADATA = builder().build(); private final Map templates; @@ -83,6 +85,7 @@ public int hashCode() { * * @opensearch.api */ + @PublicApi(since = "1.0.0") public static class Builder { private final Map templates; diff --git a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java index 9d763c970c3e7..81ee55fcc31d4 100644 --- a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java +++ b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java @@ -102,6 +102,9 @@ import org.opensearch.gateway.GatewayService; import org.opensearch.gateway.PersistedClusterStateService; import org.opensearch.gateway.remote.RemoteClusterStateService; +import org.opensearch.gateway.remote.RemoteGlobalMetadataManager; +import org.opensearch.gateway.remote.RemoteIndexMetadataManager; +import org.opensearch.gateway.remote.RemoteManifestManager; import org.opensearch.http.HttpTransportSettings; import org.opensearch.index.IndexModule; import org.opensearch.index.IndexSettings; @@ -703,9 +706,9 @@ public void apply(Settings value, Settings current, Settings previous) { // Remote cluster state settings RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING, - RemoteClusterStateService.INDEX_METADATA_UPLOAD_TIMEOUT_SETTING, - RemoteClusterStateService.GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING, - RemoteClusterStateService.METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING, + RemoteIndexMetadataManager.INDEX_METADATA_UPLOAD_TIMEOUT_SETTING, + RemoteGlobalMetadataManager.GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING, + RemoteManifestManager.METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING, RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING, RemoteStoreNodeService.MIGRATION_DIRECTION_SETTING, IndicesService.CLUSTER_REMOTE_INDEX_RESTRICT_ASYNC_DURABILITY_SETTING, 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 ac9a14d0ba74f..1f7f40b8fb598 100644 --- a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java +++ b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java @@ -11,13 +11,10 @@ import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.message.ParameterizedMessage; -import org.opensearch.Version; import org.opensearch.action.LatchedActionListener; import org.opensearch.cluster.ClusterState; -import org.opensearch.cluster.coordination.CoordinationMetadata; import org.opensearch.cluster.metadata.IndexMetadata; import org.opensearch.cluster.metadata.Metadata; -import org.opensearch.cluster.metadata.TemplatesMetadata; import org.opensearch.common.CheckedRunnable; import org.opensearch.common.Nullable; import org.opensearch.common.blobstore.BlobContainer; @@ -30,44 +27,55 @@ import org.opensearch.common.unit.TimeValue; import org.opensearch.common.util.io.IOUtils; import org.opensearch.core.action.ActionListener; -import org.opensearch.core.xcontent.ToXContent; import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedIndexMetadata; import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedMetadataAttribute; -import org.opensearch.index.remote.RemoteStoreUtils; import org.opensearch.index.translog.transfer.BlobStoreTransferService; import org.opensearch.node.Node; import org.opensearch.node.remotestore.RemoteStoreNodeAttribute; import org.opensearch.repositories.RepositoriesService; import org.opensearch.repositories.Repository; import org.opensearch.repositories.blobstore.BlobStoreRepository; -import org.opensearch.repositories.blobstore.ChecksumBlobStoreFormat; import org.opensearch.threadpool.ThreadPool; import java.io.Closeable; import java.io.IOException; -import java.nio.charset.StandardCharsets; import java.util.ArrayList; -import java.util.Base64; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; 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.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import java.util.function.LongSupplier; import java.util.function.Supplier; import java.util.stream.Collectors; -import static java.util.Objects.requireNonNull; import static org.opensearch.gateway.PersistedClusterStateService.SLOW_WRITE_LOGGING_THRESHOLD; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.RemoteStateTransferException; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.UploadedMetadataResults; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.clusterUUIDContainer; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.getCusterMetadataBasePath; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.COORDINATION_METADATA; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.COORDINATION_METADATA_FORMAT; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.CUSTOM_DELIMITER; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.CUSTOM_METADATA; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.CUSTOM_METADATA_FORMAT; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.GLOBAL_METADATA_FORMAT; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.GLOBAL_METADATA_PATH_TOKEN; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.SETTINGS_METADATA_FORMAT; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.SETTING_METADATA; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.TEMPLATES_METADATA; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.TEMPLATES_METADATA_FORMAT; +import static org.opensearch.gateway.remote.RemoteIndexMetadataManager.INDEX_METADATA_FORMAT; +import static org.opensearch.gateway.remote.RemoteIndexMetadataManager.INDEX_PATH_TOKEN; +import static org.opensearch.gateway.remote.RemoteManifestManager.MANIFEST_PATH_TOKEN; import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteStoreClusterStateEnabled; /** @@ -76,102 +84,10 @@ * @opensearch.internal */ public class RemoteClusterStateService implements Closeable { - - public static final String METADATA_NAME_FORMAT = "%s.dat"; - - public static final String METADATA_MANIFEST_NAME_FORMAT = "%s"; - public static final int RETAINED_MANIFESTS = 10; - public static final String DELIMITER = "__"; - public static final String CUSTOM_DELIMITER = "--"; - private static final Logger logger = LogManager.getLogger(RemoteClusterStateService.class); - public static final TimeValue INDEX_METADATA_UPLOAD_TIMEOUT_DEFAULT = TimeValue.timeValueMillis(20000); - - public static final TimeValue GLOBAL_METADATA_UPLOAD_TIMEOUT_DEFAULT = TimeValue.timeValueMillis(20000); - - public static final TimeValue METADATA_MANIFEST_UPLOAD_TIMEOUT_DEFAULT = TimeValue.timeValueMillis(20000); - - public static final Setting INDEX_METADATA_UPLOAD_TIMEOUT_SETTING = Setting.timeSetting( - "cluster.remote_store.state.index_metadata.upload_timeout", - INDEX_METADATA_UPLOAD_TIMEOUT_DEFAULT, - Setting.Property.Dynamic, - Setting.Property.NodeScope - ); - - public static final Setting GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING = Setting.timeSetting( - "cluster.remote_store.state.global_metadata.upload_timeout", - GLOBAL_METADATA_UPLOAD_TIMEOUT_DEFAULT, - Setting.Property.Dynamic, - Setting.Property.NodeScope - ); - - public static final Setting METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING = Setting.timeSetting( - "cluster.remote_store.state.metadata_manifest.upload_timeout", - METADATA_MANIFEST_UPLOAD_TIMEOUT_DEFAULT, - Setting.Property.Dynamic, - Setting.Property.NodeScope - ); - - public static final ChecksumBlobStoreFormat INDEX_METADATA_FORMAT = new ChecksumBlobStoreFormat<>( - "index-metadata", - METADATA_NAME_FORMAT, - IndexMetadata::fromXContent - ); - - public static final ChecksumBlobStoreFormat GLOBAL_METADATA_FORMAT = new ChecksumBlobStoreFormat<>( - "metadata", - METADATA_NAME_FORMAT, - Metadata::fromXContent - ); - - public static final ChecksumBlobStoreFormat COORDINATION_METADATA_FORMAT = new ChecksumBlobStoreFormat<>( - "coordination", - METADATA_NAME_FORMAT, - CoordinationMetadata::fromXContent - ); - - public static final ChecksumBlobStoreFormat SETTINGS_METADATA_FORMAT = new ChecksumBlobStoreFormat<>( - "settings", - METADATA_NAME_FORMAT, - Settings::fromXContent - ); - - public static final ChecksumBlobStoreFormat TEMPLATES_METADATA_FORMAT = new ChecksumBlobStoreFormat<>( - "templates", - METADATA_NAME_FORMAT, - TemplatesMetadata::fromXContent - ); - - public static final ChecksumBlobStoreFormat CUSTOM_METADATA_FORMAT = new ChecksumBlobStoreFormat<>( - "custom", - METADATA_NAME_FORMAT, - Metadata.Custom::fromXContent - ); - - /** - * Manifest format compatible with older codec v0, where codec version was missing. - */ - public static final ChecksumBlobStoreFormat CLUSTER_METADATA_MANIFEST_FORMAT_V0 = - new ChecksumBlobStoreFormat<>("cluster-metadata-manifest", METADATA_MANIFEST_NAME_FORMAT, ClusterMetadataManifest::fromXContentV0); - - /** - * Manifest format compatible with older codec v1, where global metadata was missing. - */ - public static final ChecksumBlobStoreFormat CLUSTER_METADATA_MANIFEST_FORMAT_V1 = - new ChecksumBlobStoreFormat<>("cluster-metadata-manifest", METADATA_MANIFEST_NAME_FORMAT, ClusterMetadataManifest::fromXContentV1); - - /** - * Manifest format compatible with codec v2, where we introduced codec versions/global metadata. - */ - public static final ChecksumBlobStoreFormat CLUSTER_METADATA_MANIFEST_FORMAT = new ChecksumBlobStoreFormat<>( - "cluster-metadata-manifest", - METADATA_MANIFEST_NAME_FORMAT, - ClusterMetadataManifest::fromXContent - ); - /** * Used to specify if cluster state metadata should be published to remote store */ @@ -182,18 +98,6 @@ public class RemoteClusterStateService implements Closeable { Property.Final ); - public static final String CLUSTER_STATE_PATH_TOKEN = "cluster-state"; - public static final String INDEX_PATH_TOKEN = "index"; - public static final String GLOBAL_METADATA_PATH_TOKEN = "global-metadata"; - public static final String MANIFEST_PATH_TOKEN = "manifest"; - public static final String MANIFEST_FILE_PREFIX = "manifest"; - public static final String METADATA_FILE_PREFIX = "metadata"; - public static final String COORDINATION_METADATA = "coordination"; - public static final String SETTING_METADATA = "settings"; - public static final String TEMPLATES_METADATA = "templates"; - public static final String CUSTOM_METADATA = "custom"; - public static final int SPLITED_MANIFEST_FILE_LENGTH = 6; // file name manifest__term__version__C/P__timestamp__codecversion - private final String nodeId; private final Supplier repositoriesService; private final Settings settings; @@ -203,24 +107,13 @@ public class RemoteClusterStateService implements Closeable { private BlobStoreTransferService blobStoreTransferService; private volatile TimeValue slowWriteLoggingThreshold; - private volatile TimeValue indexMetadataUploadTimeout; - private volatile TimeValue globalMetadataUploadTimeout; - private volatile TimeValue metadataManifestUploadTimeout; - private final AtomicBoolean deleteStaleMetadataRunning = new AtomicBoolean(false); private final RemotePersistenceStats remoteStateStats; + private RemoteIndexMetadataManager remoteIndexMetadataManager; + private RemoteGlobalMetadataManager remoteGlobalMetadataManager; + private RemoteManifestManager remoteManifestManager; + private ClusterSettings clusterSettings; public static final int INDEX_METADATA_CURRENT_CODEC_VERSION = 1; - public static final int MANIFEST_CURRENT_CODEC_VERSION = ClusterMetadataManifest.CODEC_V2; - public static final int GLOBAL_METADATA_CURRENT_CODEC_VERSION = 1; - - // ToXContent Params with gateway mode. - // We are using gateway context mode to persist all custom metadata. - public static final ToXContent.Params FORMAT_PARAMS; - static { - Map params = new HashMap<>(1); - params.put(Metadata.CONTEXT_MODE_PARAM, Metadata.CONTEXT_MODE_GATEWAY); - FORMAT_PARAMS = new ToXContent.MapParams(params); - } public RemoteClusterStateService( String nodeId, @@ -236,14 +129,9 @@ public RemoteClusterStateService( this.settings = settings; this.relativeTimeNanosSupplier = relativeTimeNanosSupplier; this.threadpool = threadPool; + this.clusterSettings = clusterSettings; this.slowWriteLoggingThreshold = clusterSettings.get(SLOW_WRITE_LOGGING_THRESHOLD); - this.indexMetadataUploadTimeout = clusterSettings.get(INDEX_METADATA_UPLOAD_TIMEOUT_SETTING); - this.globalMetadataUploadTimeout = clusterSettings.get(GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING); - this.metadataManifestUploadTimeout = clusterSettings.get(METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING); clusterSettings.addSettingsUpdateConsumer(SLOW_WRITE_LOGGING_THRESHOLD, this::setSlowWriteLoggingThreshold); - clusterSettings.addSettingsUpdateConsumer(INDEX_METADATA_UPLOAD_TIMEOUT_SETTING, this::setIndexMetadataUploadTimeout); - clusterSettings.addSettingsUpdateConsumer(GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING, this::setGlobalMetadataUploadTimeout); - clusterSettings.addSettingsUpdateConsumer(METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING, this::setMetadataManifestUploadTimeout); this.remoteStateStats = new RemotePersistenceStats(); } @@ -276,7 +164,7 @@ public ClusterMetadataManifest writeFullMetadata(ClusterState clusterState, Stri true, true ); - final ClusterMetadataManifest manifest = uploadManifest( + final ClusterMetadataManifest manifest = remoteManifestManager.uploadManifest( clusterState, uploadedMetadataResults.uploadedIndexMetadata, previousClusterUUID, @@ -345,7 +233,10 @@ public ClusterMetadataManifest writeIncrementalMetadata( clusterState.metadata() ) == false; final Map previousStateCustomMap = new HashMap<>(previousManifest.getCustomMetadataMap()); - final Map customsToUpload = getUpdatedCustoms(clusterState, previousClusterState); + final Map customsToUpload = remoteGlobalMetadataManager.getUpdatedCustoms( + clusterState, + previousClusterState + ); final Map allUploadedCustomMap = new HashMap<>(previousManifest.getCustomMetadataMap()); for (final String custom : clusterState.metadata().customs().keySet()) { // remove all the customs which are present currently @@ -407,7 +298,7 @@ public ClusterMetadataManifest writeIncrementalMetadata( // remove the data for removed custom/indices previousStateCustomMap.keySet().forEach(allUploadedCustomMap::remove); previousStateIndexMetadataVersionByName.keySet().forEach(allUploadedIndexMetadata::remove); - final ClusterMetadataManifest manifest = uploadManifest( + final ClusterMetadataManifest manifest = remoteManifestManager.uploadManifest( clusterState, new ArrayList<>(allUploadedIndexMetadata.values()), previousManifest.getPreviousClusterUUID(), @@ -493,7 +384,7 @@ private UploadedMetadataResults writeMetadataInParallel( if (uploadSettingsMetadata) { uploadTasks.put( SETTING_METADATA, - getAsyncMetadataWriteAction( + remoteGlobalMetadataManager.getAsyncMetadataWriteAction( clusterState, SETTING_METADATA, SETTINGS_METADATA_FORMAT, @@ -505,7 +396,7 @@ private UploadedMetadataResults writeMetadataInParallel( if (uploadCoordinationMetadata) { uploadTasks.put( COORDINATION_METADATA, - getAsyncMetadataWriteAction( + remoteGlobalMetadataManager.getAsyncMetadataWriteAction( clusterState, COORDINATION_METADATA, COORDINATION_METADATA_FORMAT, @@ -517,7 +408,7 @@ private UploadedMetadataResults writeMetadataInParallel( if (uploadTemplateMetadata) { uploadTasks.put( TEMPLATES_METADATA, - getAsyncMetadataWriteAction( + remoteGlobalMetadataManager.getAsyncMetadataWriteAction( clusterState, TEMPLATES_METADATA, TEMPLATES_METADATA_FORMAT, @@ -530,11 +421,20 @@ private UploadedMetadataResults writeMetadataInParallel( String customComponent = String.join(CUSTOM_DELIMITER, CUSTOM_METADATA, key); uploadTasks.put( customComponent, - getAsyncMetadataWriteAction(clusterState, customComponent, CUSTOM_METADATA_FORMAT, value, listener) + remoteGlobalMetadataManager.getAsyncMetadataWriteAction( + clusterState, + customComponent, + CUSTOM_METADATA_FORMAT, + value, + listener + ) ); }); indexToUpload.forEach(indexMetadata -> { - uploadTasks.put(indexMetadata.getIndexName(), getIndexMetadataAsyncAction(clusterState, indexMetadata, listener)); + uploadTasks.put( + indexMetadata.getIndexName(), + remoteIndexMetadataManager.getIndexMetadataAsyncAction(clusterState, indexMetadata, listener) + ); }); // start async upload of all required metadata files @@ -543,7 +443,7 @@ private UploadedMetadataResults writeMetadataInParallel( } try { - if (latch.await(getGlobalMetadataUploadTimeout().millis(), TimeUnit.MILLISECONDS) == false) { + if (latch.await(remoteGlobalMetadataManager.getGlobalMetadataUploadTimeout().millis(), TimeUnit.MILLISECONDS) == false) { // TODO: We should add metrics where transfer is timing out. [Issue: #10687] RemoteStateTransferException ex = new RemoteStateTransferException( String.format( @@ -603,75 +503,6 @@ private UploadedMetadataResults writeMetadataInParallel( return response; } - /** - * 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 - */ - private CheckedRunnable getIndexMetadataAsyncAction( - ClusterState clusterState, - IndexMetadata indexMetadata, - LatchedActionListener latchedActionListener - ) { - final BlobContainer indexMetadataContainer = indexMetadataContainer( - clusterState.getClusterName().value(), - clusterState.metadata().clusterUUID(), - indexMetadata.getIndexUUID() - ); - final String indexMetadataFilename = indexMetadataFileName(indexMetadata); - ActionListener completionListener = ActionListener.wrap( - resp -> latchedActionListener.onResponse( - new UploadedIndexMetadata( - indexMetadata.getIndex().getName(), - indexMetadata.getIndexUUID(), - indexMetadataContainer.path().buildAsString() + indexMetadataFilename - ) - ), - ex -> latchedActionListener.onFailure(new RemoteStateTransferException(indexMetadata.getIndex().toString(), ex)) - ); - - return () -> INDEX_METADATA_FORMAT.writeAsyncWithUrgentPriority( - indexMetadata, - indexMetadataContainer, - indexMetadataFilename, - blobStoreRepository.getCompressor(), - completionListener, - FORMAT_PARAMS - ); - } - - /** - * Allows async upload of Metadata components to remote - */ - - private CheckedRunnable getAsyncMetadataWriteAction( - ClusterState clusterState, - String component, - ChecksumBlobStoreFormat componentMetadataBlobStore, - ToXContent componentMetadata, - LatchedActionListener latchedActionListener - ) { - final BlobContainer globalMetadataContainer = globalMetadataContainer( - clusterState.getClusterName().value(), - clusterState.metadata().clusterUUID() - ); - final String componentMetadataFilename = metadataAttributeFileName(component, clusterState.metadata().version()); - ActionListener completionListener = ActionListener.wrap( - resp -> latchedActionListener.onResponse(new UploadedMetadataAttribute(component, componentMetadataFilename)), - ex -> latchedActionListener.onFailure(new RemoteStateTransferException(component, ex)) - ); - return () -> componentMetadataBlobStore.writeAsyncWithUrgentPriority( - componentMetadata, - globalMetadataContainer, - componentMetadataFilename, - blobStoreRepository.getCompressor(), - completionListener, - FORMAT_PARAMS - ); - } - @Nullable public ClusterMetadataManifest markLastStateAsCommitted(ClusterState clusterState, ClusterMetadataManifest previousManifest) throws IOException { @@ -681,7 +512,7 @@ public ClusterMetadataManifest markLastStateAsCommitted(ClusterState clusterStat return null; } assert previousManifest != null : "Last cluster metadata manifest is not set"; - ClusterMetadataManifest committedManifest = uploadManifest( + ClusterMetadataManifest committedManifest = remoteManifestManager.uploadManifest( clusterState, previousManifest.getIndices(), previousManifest.getPreviousClusterUUID(), @@ -695,6 +526,17 @@ public ClusterMetadataManifest markLastStateAsCommitted(ClusterState clusterStat return committedManifest; } + /** + * 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 + */ + public Optional getLatestClusterMetadataManifest(String clusterName, String clusterUUID) { + return remoteManifestManager.getLatestClusterMetadataManifest(clusterName, clusterUUID); + } + @Override public void close() throws IOException { if (blobStoreRepository != null) { @@ -711,131 +553,16 @@ public void start() { final Repository repository = repositoriesService.get().repository(remoteStoreRepo); assert repository instanceof BlobStoreRepository : "Repository should be instance of BlobStoreRepository"; blobStoreRepository = (BlobStoreRepository) repository; - } - - private ClusterMetadataManifest uploadV1Manifest( - ClusterState clusterState, - List uploadedIndexMetadata, - String previousClusterUUID, - String globalMetadataFileName, - boolean committed - ) throws IOException { - synchronized (this) { - final String manifestFileName = getManifestFileName( - clusterState.term(), - clusterState.version(), - committed, - ClusterMetadataManifest.CODEC_V1 - ); - ClusterMetadataManifest manifest = ClusterMetadataManifest.builder() - .clusterTerm(clusterState.term()) - .stateVersion(clusterState.getVersion()) - .clusterUUID(clusterState.metadata().clusterUUID()) - .stateUUID(clusterState.stateUUID()) - .opensearchVersion(Version.CURRENT) - .nodeId(nodeId) - .committed(committed) - .codecVersion(ClusterMetadataManifest.CODEC_V1) - .globalMetadataFileName(globalMetadataFileName) - .indices(uploadedIndexMetadata) - .previousClusterUUID(previousClusterUUID) - .clusterUUIDCommitted(clusterState.metadata().clusterUUIDCommitted()) - .build(); - writeMetadataManifest(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID(), manifest, manifestFileName); - return manifest; - } - } - - private ClusterMetadataManifest uploadManifest( - ClusterState clusterState, - List uploadedIndexMetadata, - String previousClusterUUID, - UploadedMetadataAttribute uploadedCoordinationMetadata, - UploadedMetadataAttribute uploadedSettingsMetadata, - UploadedMetadataAttribute uploadedTemplatesMetadata, - Map uploadedCustomMetadataMap, - boolean committed - ) throws IOException { - synchronized (this) { - final String manifestFileName = getManifestFileName( - clusterState.term(), - clusterState.version(), - committed, - MANIFEST_CURRENT_CODEC_VERSION - ); - final ClusterMetadataManifest manifest = new ClusterMetadataManifest( - clusterState.term(), - clusterState.getVersion(), - clusterState.metadata().clusterUUID(), - clusterState.stateUUID(), - Version.CURRENT, - nodeId, - committed, - MANIFEST_CURRENT_CODEC_VERSION, - null, - uploadedIndexMetadata, - previousClusterUUID, - clusterState.metadata().clusterUUIDCommitted(), - uploadedCoordinationMetadata, - uploadedSettingsMetadata, - uploadedTemplatesMetadata, - uploadedCustomMetadataMap - ); - writeMetadataManifest(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID(), manifest, manifestFileName); - return manifest; - } - } - - private void writeMetadataManifest(String clusterName, String clusterUUID, ClusterMetadataManifest uploadManifest, String fileName) - throws IOException { - AtomicReference result = new AtomicReference(); - AtomicReference exceptionReference = new AtomicReference(); - - final BlobContainer metadataManifestContainer = manifestContainer(clusterName, clusterUUID); - - // latch to wait until upload is not finished - CountDownLatch latch = new CountDownLatch(1); - - LatchedActionListener completionListener = new LatchedActionListener<>(ActionListener.wrap(resp -> { - logger.trace(String.format(Locale.ROOT, "Manifest file uploaded successfully.")); - }, ex -> { exceptionReference.set(ex); }), latch); - - getClusterMetadataManifestBlobStoreFormat(fileName).writeAsyncWithUrgentPriority( - uploadManifest, - metadataManifestContainer, - fileName, - blobStoreRepository.getCompressor(), - completionListener, - FORMAT_PARAMS - ); - - try { - if (latch.await(getMetadataManifestUploadTimeout().millis(), TimeUnit.MILLISECONDS) == false) { - RemoteStateTransferException ex = new RemoteStateTransferException( - String.format(Locale.ROOT, "Timed out waiting for transfer of manifest file to complete") - ); - throw ex; - } - } catch (InterruptedException ex) { - RemoteStateTransferException exception = new RemoteStateTransferException( - String.format(Locale.ROOT, "Timed out waiting for transfer of manifest file to complete - %s"), - ex - ); - Thread.currentThread().interrupt(); - throw exception; - } - if (exceptionReference.get() != null) { - throw new RemoteStateTransferException(exceptionReference.get().getMessage(), exceptionReference.get()); - } - logger.debug( - "Metadata manifest file [{}] written during [{}] phase. ", - fileName, - uploadManifest.isCommitted() ? "commit" : "publish" - ); + remoteGlobalMetadataManager = new RemoteGlobalMetadataManager(blobStoreRepository, clusterSettings); + remoteIndexMetadataManager = new RemoteIndexMetadataManager(blobStoreRepository, clusterSettings); + remoteManifestManager = new RemoteManifestManager(blobStoreRepository, clusterSettings, nodeId); } private String fetchPreviousClusterUUID(String clusterName, String clusterUUID) { - final Optional latestManifest = getLatestClusterMetadataManifest(clusterName, clusterUUID); + final Optional latestManifest = remoteManifestManager.getLatestClusterMetadataManifest( + clusterName, + clusterUUID + ); if (!latestManifest.isPresent()) { final String previousClusterUUID = getLastKnownUUIDFromRemote(clusterName); assert !clusterUUID.equals(previousClusterUUID) : "Last cluster UUID is same current cluster UUID"; @@ -844,190 +571,10 @@ private String fetchPreviousClusterUUID(String clusterName, String clusterUUID) return latestManifest.get().getPreviousClusterUUID(); } - private BlobContainer indexMetadataContainer(String clusterName, String clusterUUID, String indexUUID) { - // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/index/ftqsCnn9TgOX - return blobStoreRepository.blobStore() - .blobContainer(getCusterMetadataBasePath(clusterName, clusterUUID).add(INDEX_PATH_TOKEN).add(indexUUID)); - } - - private BlobContainer globalMetadataContainer(String clusterName, String clusterUUID) { - // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/global-metadata/ - return blobStoreRepository.blobStore() - .blobContainer(getCusterMetadataBasePath(clusterName, clusterUUID).add(GLOBAL_METADATA_PATH_TOKEN)); - } - - private BlobContainer manifestContainer(String clusterName, String clusterUUID) { - // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/manifest - return blobStoreRepository.blobStore().blobContainer(getManifestFolderPath(clusterName, clusterUUID)); - } - - private BlobPath getCusterMetadataBasePath(String clusterName, String clusterUUID) { - return blobStoreRepository.basePath().add(encodeString(clusterName)).add(CLUSTER_STATE_PATH_TOKEN).add(clusterUUID); - } - - 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) - ); - } - private void setSlowWriteLoggingThreshold(TimeValue slowWriteLoggingThreshold) { this.slowWriteLoggingThreshold = slowWriteLoggingThreshold; } - private void setIndexMetadataUploadTimeout(TimeValue newIndexMetadataUploadTimeout) { - this.indexMetadataUploadTimeout = newIndexMetadataUploadTimeout; - } - - private void setGlobalMetadataUploadTimeout(TimeValue newGlobalMetadataUploadTimeout) { - this.globalMetadataUploadTimeout = newGlobalMetadataUploadTimeout; - } - - private void setMetadataManifestUploadTimeout(TimeValue newMetadataManifestUploadTimeout) { - this.metadataManifestUploadTimeout = newMetadataManifestUploadTimeout; - } - - private Map getUpdatedCustoms(ClusterState currentState, ClusterState previousState) { - if (Metadata.isCustomMetadataEqual(previousState.metadata(), currentState.metadata())) { - return new HashMap<>(); - } - Map updatedCustom = new HashMap<>(); - Set currentCustoms = new HashSet<>(currentState.metadata().customs().keySet()); - for (Map.Entry cursor : previousState.metadata().customs().entrySet()) { - if (cursor.getValue().context().contains(Metadata.XContentContext.GATEWAY)) { - if (currentCustoms.contains(cursor.getKey()) - && !cursor.getValue().equals(currentState.metadata().custom(cursor.getKey()))) { - // If the custom metadata is updated, we need to upload the new version. - updatedCustom.put(cursor.getKey(), currentState.metadata().custom(cursor.getKey())); - } - currentCustoms.remove(cursor.getKey()); - } - } - for (String custom : currentCustoms) { - Metadata.Custom cursor = currentState.metadata().custom(custom); - if (cursor.context().contains(Metadata.XContentContext.GATEWAY)) { - updatedCustom.put(custom, cursor); - } - } - return updatedCustom; - } - - public TimeValue getIndexMetadataUploadTimeout() { - return this.indexMetadataUploadTimeout; - } - - public TimeValue getGlobalMetadataUploadTimeout() { - return this.globalMetadataUploadTimeout; - } - - public TimeValue getMetadataManifestUploadTimeout() { - return this.metadataManifestUploadTimeout; - } - - static String getManifestFileName(long term, long version, boolean committed, int codecVersion) { - // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/manifest/manifest______C/P____ - return String.join( - DELIMITER, - MANIFEST_PATH_TOKEN, - RemoteStoreUtils.invertLong(term), - RemoteStoreUtils.invertLong(version), - (committed ? "C" : "P"), // C for committed and P for published - RemoteStoreUtils.invertLong(System.currentTimeMillis()), - String.valueOf(codecVersion) // Keep the codec version at last place only, during read we reads last place to - // determine codec version. - ); - } - - static String indexMetadataFileName(IndexMetadata indexMetadata) { - // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/index//metadata______ - return String.join( - DELIMITER, - METADATA_FILE_PREFIX, - RemoteStoreUtils.invertLong(indexMetadata.getVersion()), - RemoteStoreUtils.invertLong(System.currentTimeMillis()), - String.valueOf(INDEX_METADATA_CURRENT_CODEC_VERSION) // Keep the codec version at last place only, during read we reads last - // place to determine codec version. - ); - } - - private static String globalMetadataFileName(Metadata metadata) { - // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/global-metadata/metadata______ - return String.join( - DELIMITER, - METADATA_FILE_PREFIX, - RemoteStoreUtils.invertLong(metadata.version()), - RemoteStoreUtils.invertLong(System.currentTimeMillis()), - String.valueOf(GLOBAL_METADATA_CURRENT_CODEC_VERSION) - ); - } - - private static String metadataAttributeFileName(String componentPrefix, Long metadataVersion) { - // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/global-metadata/______ - return String.join( - DELIMITER, - componentPrefix, - RemoteStoreUtils.invertLong(metadataVersion), - RemoteStoreUtils.invertLong(System.currentTimeMillis()), - String.valueOf(GLOBAL_METADATA_CURRENT_CODEC_VERSION) - ); - } - - private BlobPath getManifestFolderPath(String clusterName, String clusterUUID) { - return getCusterMetadataBasePath(clusterName, clusterUUID).add(MANIFEST_PATH_TOKEN); - } - - /** - * 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 - * @param clusterMetadataManifest manifest file of cluster - * @return {@code Map} latest IndexUUID to IndexMetadata map - */ - private Map getIndexMetadataMap( - String clusterName, - String clusterUUID, - ClusterMetadataManifest clusterMetadataManifest - ) { - assert Objects.equals(clusterUUID, clusterMetadataManifest.getClusterUUID()) - : "Corrupt ClusterMetadataManifest found. Cluster UUID mismatch."; - Map remoteIndexMetadata = new HashMap<>(); - for (UploadedIndexMetadata uploadedIndexMetadata : clusterMetadataManifest.getIndices()) { - IndexMetadata indexMetadata = getIndexMetadata(clusterName, clusterUUID, uploadedIndexMetadata); - remoteIndexMetadata.put(uploadedIndexMetadata.getIndexUUID(), indexMetadata); - } - return remoteIndexMetadata; - } - - /** - * 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 - * @return {@link IndexMetadata} - */ - private IndexMetadata getIndexMetadata(String clusterName, String clusterUUID, UploadedIndexMetadata uploadedIndexMetadata) { - BlobContainer blobContainer = indexMetadataContainer(clusterName, clusterUUID, uploadedIndexMetadata.getIndexUUID()); - try { - String[] splitPath = uploadedIndexMetadata.getUploadedFilename().split("/"); - return INDEX_METADATA_FORMAT.read( - blobContainer, - splitPath[splitPath.length - 1], - blobStoreRepository.getNamedXContentRegistry() - ); - } catch (IOException e) { - throw new IllegalStateException( - String.format(Locale.ROOT, "Error while downloading IndexMetadata - %s", uploadedIndexMetadata.getUploadedFilename()), - e - ); - } - } - /** * Fetch latest ClusterState from remote, including global metadata, index metadata and cluster state version * @@ -1037,7 +584,10 @@ private IndexMetadata getIndexMetadata(String clusterName, String clusterUUID, U */ public ClusterState getLatestClusterState(String clusterName, String clusterUUID) { start(); - Optional clusterMetadataManifest = getLatestClusterMetadataManifest(clusterName, clusterUUID); + Optional clusterMetadataManifest = remoteManifestManager.getLatestClusterMetadataManifest( + clusterName, + clusterUUID + ); if (clusterMetadataManifest.isEmpty()) { throw new IllegalStateException( String.format(Locale.ROOT, "Latest cluster metadata manifest is not present for the provided clusterUUID: %s", clusterUUID) @@ -1045,10 +595,14 @@ public ClusterState getLatestClusterState(String clusterName, String clusterUUID } // Fetch Global Metadata - Metadata globalMetadata = getGlobalMetadata(clusterName, clusterUUID, clusterMetadataManifest.get()); + Metadata globalMetadata = remoteGlobalMetadataManager.getGlobalMetadata(clusterName, clusterUUID, clusterMetadataManifest.get()); // Fetch Index Metadata - Map indices = getIndexMetadataMap(clusterName, clusterUUID, clusterMetadataManifest.get()); + Map indices = remoteIndexMetadataManager.getIndexMetadataMap( + clusterName, + clusterUUID, + clusterMetadataManifest.get() + ); Map indexMetadataMap = new HashMap<>(); indices.values().forEach(indexMetadata -> { indexMetadataMap.put(indexMetadata.getIndex().getName(), indexMetadata); }); @@ -1059,154 +613,6 @@ public ClusterState getLatestClusterState(String clusterName, String clusterUUID .build(); } - private Metadata getGlobalMetadata(String clusterName, String clusterUUID, ClusterMetadataManifest clusterMetadataManifest) { - String globalMetadataFileName = clusterMetadataManifest.getGlobalMetadataFileName(); - try { - // Fetch Global metadata - if (globalMetadataFileName != null) { - String[] splitPath = globalMetadataFileName.split("/"); - return GLOBAL_METADATA_FORMAT.read( - globalMetadataContainer(clusterName, clusterUUID), - splitPath[splitPath.length - 1], - blobStoreRepository.getNamedXContentRegistry() - ); - } else if (clusterMetadataManifest.hasMetadataAttributesFiles()) { - CoordinationMetadata coordinationMetadata = getCoordinationMetadata( - clusterName, - clusterUUID, - clusterMetadataManifest.getCoordinationMetadata().getUploadedFilename() - ); - Settings settingsMetadata = getSettingsMetadata( - clusterName, - clusterUUID, - clusterMetadataManifest.getSettingsMetadata().getUploadedFilename() - ); - TemplatesMetadata templatesMetadata = getTemplatesMetadata( - clusterName, - clusterUUID, - clusterMetadataManifest.getTemplatesMetadata().getUploadedFilename() - ); - Metadata.Builder builder = new Metadata.Builder(); - builder.coordinationMetadata(coordinationMetadata); - builder.persistentSettings(settingsMetadata); - builder.templates(templatesMetadata); - clusterMetadataManifest.getCustomMetadataMap() - .forEach( - (key, value) -> builder.putCustom( - key, - getCustomsMetadata(clusterName, clusterUUID, value.getUploadedFilename(), key) - ) - ); - return builder.build(); - } else { - return Metadata.EMPTY_METADATA; - } - } catch (IOException e) { - throw new IllegalStateException( - String.format(Locale.ROOT, "Error while downloading Global Metadata - %s", globalMetadataFileName), - e - ); - } - } - - private CoordinationMetadata getCoordinationMetadata(String clusterName, String clusterUUID, String coordinationMetadataFileName) { - try { - // Fetch Coordination metadata - if (coordinationMetadataFileName != null) { - String[] splitPath = coordinationMetadataFileName.split("/"); - return COORDINATION_METADATA_FORMAT.read( - globalMetadataContainer(clusterName, clusterUUID), - splitPath[splitPath.length - 1], - blobStoreRepository.getNamedXContentRegistry() - ); - } else { - return CoordinationMetadata.EMPTY_METADATA; - } - } catch (IOException e) { - throw new IllegalStateException( - String.format(Locale.ROOT, "Error while downloading Coordination Metadata - %s", coordinationMetadataFileName), - e - ); - } - } - - private Settings getSettingsMetadata(String clusterName, String clusterUUID, String settingsMetadataFileName) { - try { - // Fetch Settings metadata - if (settingsMetadataFileName != null) { - String[] splitPath = settingsMetadataFileName.split("/"); - return SETTINGS_METADATA_FORMAT.read( - globalMetadataContainer(clusterName, clusterUUID), - splitPath[splitPath.length - 1], - blobStoreRepository.getNamedXContentRegistry() - ); - } else { - return Settings.EMPTY; - } - } catch (IOException e) { - throw new IllegalStateException( - String.format(Locale.ROOT, "Error while downloading Settings Metadata - %s", settingsMetadataFileName), - e - ); - } - } - - private TemplatesMetadata getTemplatesMetadata(String clusterName, String clusterUUID, String templatesMetadataFileName) { - try { - // Fetch Templates metadata - if (templatesMetadataFileName != null) { - String[] splitPath = templatesMetadataFileName.split("/"); - return TEMPLATES_METADATA_FORMAT.read( - globalMetadataContainer(clusterName, clusterUUID), - splitPath[splitPath.length - 1], - blobStoreRepository.getNamedXContentRegistry() - ); - } else { - return TemplatesMetadata.EMPTY_METADATA; - } - } catch (IOException e) { - throw new IllegalStateException( - String.format(Locale.ROOT, "Error while downloading Templates Metadata - %s", templatesMetadataFileName), - e - ); - } - } - - private Metadata.Custom getCustomsMetadata(String clusterName, String clusterUUID, String customMetadataFileName, String custom) { - requireNonNull(customMetadataFileName); - try { - // Fetch Custom metadata - String[] splitPath = customMetadataFileName.split("/"); - ChecksumBlobStoreFormat customChecksumBlobStoreFormat = new ChecksumBlobStoreFormat<>( - "custom", - METADATA_NAME_FORMAT, - (parser -> Metadata.Custom.fromXContent(parser, custom)) - ); - return customChecksumBlobStoreFormat.read( - globalMetadataContainer(clusterName, clusterUUID), - splitPath[splitPath.length - 1], - blobStoreRepository.getNamedXContentRegistry() - ); - } catch (IOException e) { - throw new IllegalStateException( - String.format(Locale.ROOT, "Error while downloading Custom Metadata - %s", customMetadataFileName), - e - ); - } - } - - /** - * 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 - */ - public Optional getLatestClusterMetadataManifest(String clusterName, String clusterUUID) { - Optional latestManifestFileName = getLatestManifestFileName(clusterName, clusterUUID); - return latestManifestFileName.map(s -> fetchRemoteClusterMetadataManifest(clusterName, clusterUUID, s)); - } - /** * Fetch the previous cluster UUIDs from remote state store and return the most recent valid cluster UUID * @@ -1216,7 +622,10 @@ public Optional getLatestClusterMetadataManifest(String public String getLastKnownUUIDFromRemote(String clusterName) { try { Set clusterUUIDs = getAllClusterUUIDs(clusterName); - Map latestManifests = getLatestManifestForAllClusterUUIDs(clusterName, clusterUUIDs); + Map latestManifests = remoteManifestManager.getLatestManifestForAllClusterUUIDs( + clusterName, + clusterUUIDs + ); List validChain = createClusterChain(latestManifests, clusterName); if (validChain.isEmpty()) { return ClusterState.UNKNOWN_UUID; @@ -1231,29 +640,13 @@ public String getLastKnownUUIDFromRemote(String clusterName) { } private Set getAllClusterUUIDs(String clusterName) throws IOException { - Map clusterUUIDMetadata = clusterUUIDContainer(clusterName).children(); + Map clusterUUIDMetadata = clusterUUIDContainer(blobStoreRepository, 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 { - Optional manifest = getLatestClusterMetadataManifest(clusterName, clusterUUID); - manifest.ifPresent(clusterMetadataManifest -> manifestsByClusterUUID.put(clusterUUID, clusterMetadataManifest)); - } catch (Exception e) { - throw new IllegalStateException( - String.format(Locale.ROOT, "Exception in fetching manifest for clusterUUID: %s", clusterUUID), - e - ); - } - } - return manifestsByClusterUUID; - } - /** * This method creates a valid cluster UUID chain. * @@ -1331,7 +724,7 @@ private Map trimClusterUUIDs( if (!ClusterState.UNKNOWN_UUID.equals(currentManifest.getPreviousClusterUUID())) { ClusterMetadataManifest previousManifest = trimmedUUIDs.get(currentManifest.getPreviousClusterUUID()); if (isMetadataEqual(currentManifest, previousManifest, clusterName) - && isGlobalMetadataEqual(currentManifest, previousManifest, clusterName)) { + && remoteGlobalMetadataManager.isGlobalMetadataEqual(currentManifest, previousManifest, clusterName)) { trimmedUUIDs.remove(clusterUUID); } } @@ -1348,12 +741,20 @@ private boolean isMetadataEqual(ClusterMetadataManifest first, ClusterMetadataMa .stream() .collect(Collectors.toMap(md -> md.getIndexName(), Function.identity())); for (UploadedIndexMetadata uploadedIndexMetadata : first.getIndices()) { - final IndexMetadata firstIndexMetadata = getIndexMetadata(clusterName, first.getClusterUUID(), uploadedIndexMetadata); + final IndexMetadata firstIndexMetadata = remoteIndexMetadataManager.getIndexMetadata( + clusterName, + first.getClusterUUID(), + uploadedIndexMetadata + ); final UploadedIndexMetadata secondUploadedIndexMetadata = secondIndices.get(uploadedIndexMetadata.getIndexName()); if (secondUploadedIndexMetadata == null) { return false; } - final IndexMetadata secondIndexMetadata = getIndexMetadata(clusterName, second.getClusterUUID(), secondUploadedIndexMetadata); + final IndexMetadata secondIndexMetadata = remoteIndexMetadataManager.getIndexMetadata( + clusterName, + second.getClusterUUID(), + secondUploadedIndexMetadata + ); if (firstIndexMetadata.equals(secondIndexMetadata) == false) { return false; } @@ -1361,125 +762,14 @@ private boolean isMetadataEqual(ClusterMetadataManifest first, ClusterMetadataMa return true; } - private boolean isGlobalMetadataEqual(ClusterMetadataManifest first, ClusterMetadataManifest second, String clusterName) { - Metadata secondGlobalMetadata = getGlobalMetadata(clusterName, second.getClusterUUID(), second); - Metadata firstGlobalMetadata = getGlobalMetadata(clusterName, first.getClusterUUID(), first); - return Metadata.isGlobalResourcesMetadataEquals(firstGlobalMetadata, secondGlobalMetadata); - } - private boolean isValidClusterUUID(ClusterMetadataManifest manifest) { return manifest.isClusterUUIDCommitted(); } - /** - * Fetch ClusterMetadataManifest files from remote state store in order - * - * @param clusterUUID uuid of cluster state to refer to in remote - * @param clusterName name of the cluster - * @param limit max no of files to fetch - * @return all manifest file names - */ - private List getManifestFileNames(String clusterName, String clusterUUID, int limit) throws IllegalStateException { - try { - - /* - {@link BlobContainer#listBlobsByPrefixInSortedOrder} will list the latest manifest file first - as the manifest file name generated via {@link RemoteClusterStateService#getManifestFileName} ensures - when sorted in LEXICOGRAPHIC order the latest uploaded manifest file comes on top. - */ - return manifestContainer(clusterName, clusterUUID).listBlobsByPrefixInSortedOrder( - MANIFEST_FILE_PREFIX + DELIMITER, - limit, - BlobContainer.BlobNameSortOrder.LEXICOGRAPHIC - ); - } catch (IOException e) { - throw new IllegalStateException("Error while fetching latest manifest file for remote cluster state", e); - } - } - - /** - * 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 - */ - private Optional getLatestManifestFileName(String clusterName, String clusterUUID) throws IllegalStateException { - List manifestFilesMetadata = getManifestFileNames(clusterName, clusterUUID, 1); - if (manifestFilesMetadata != null && !manifestFilesMetadata.isEmpty()) { - return Optional.of(manifestFilesMetadata.get(0).name()); - } - logger.info("No manifest file present in remote store for cluster name: {}, cluster UUID: {}", clusterName, clusterUUID); - return Optional.empty(); - } - - /** - * 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 - */ - private ClusterMetadataManifest fetchRemoteClusterMetadataManifest(String clusterName, String clusterUUID, String filename) - throws IllegalStateException { - try { - return getClusterMetadataManifestBlobStoreFormat(filename).read( - manifestContainer(clusterName, clusterUUID), - filename, - blobStoreRepository.getNamedXContentRegistry() - ); - } catch (IOException e) { - throw new IllegalStateException(String.format(Locale.ROOT, "Error while downloading cluster metadata - %s", filename), e); - } - } - - private ChecksumBlobStoreFormat getClusterMetadataManifestBlobStoreFormat(String fileName) { - long codecVersion = getManifestCodecVersion(fileName); - if (codecVersion == MANIFEST_CURRENT_CODEC_VERSION) { - return CLUSTER_METADATA_MANIFEST_FORMAT; - } else if (codecVersion == ClusterMetadataManifest.CODEC_V1) { - return CLUSTER_METADATA_MANIFEST_FORMAT_V1; - } else if (codecVersion == ClusterMetadataManifest.CODEC_V0) { - return CLUSTER_METADATA_MANIFEST_FORMAT_V0; - } - - throw new IllegalArgumentException("Cluster metadata manifest file is corrupted, don't have valid codec version"); - } - - private int getManifestCodecVersion(String fileName) { - String[] splitName = fileName.split(DELIMITER); - if (splitName.length == SPLITED_MANIFEST_FILE_LENGTH) { - return Integer.parseInt(splitName[splitName.length - 1]); // Last value would be codec version. - } else if (splitName.length < SPLITED_MANIFEST_FILE_LENGTH) { // Where codec is not part of file name, i.e. default codec version 0 - // is used. - return ClusterMetadataManifest.CODEC_V0; - } else { - throw new IllegalArgumentException("Manifest file name is corrupted"); - } - } - - public static String encodeString(String content) { - return Base64.getUrlEncoder().withoutPadding().encodeToString(content.getBytes(StandardCharsets.UTF_8)); - } - public void writeMetadataFailed() { getStats().stateFailed(); } - /** - * Exception for Remote state transfer. - */ - static class RemoteStateTransferException extends RuntimeException { - - public RemoteStateTransferException(String errorDesc) { - super(errorDesc); - } - - public RemoteStateTransferException(String errorDesc, Throwable cause) { - super(errorDesc, cause); - } - } - /** * Purges all remote cluster state against provided cluster UUIDs * @@ -1490,7 +780,7 @@ void deleteStaleUUIDsClusterMetadata(String clusterName, List clusterUUI clusterUUIDs.forEach(clusterUUID -> { getBlobStoreTransferService().deleteAsync( ThreadPool.Names.REMOTE_PURGE, - getCusterMetadataBasePath(clusterName, clusterUUID), + getCusterMetadataBasePath(blobStoreRepository, clusterName, clusterUUID), new ActionListener<>() { @Override public void onResponse(Void unused) { @@ -1529,7 +819,7 @@ void deleteStaleClusterMetadata(String clusterName, String clusterUUID, int mani try { getBlobStoreTransferService().listAllInSortedOrderAsync( ThreadPool.Names.REMOTE_PURGE, - getManifestFolderPath(clusterName, clusterUUID), + remoteManifestManager.getManifestFolderPath(clusterName, clusterUUID), "manifest", Integer.MAX_VALUE, new ActionListener<>() { @@ -1576,7 +866,7 @@ private void deleteClusterMetadata( Set staleIndexMetadataPaths = new HashSet<>(); Set staleGlobalMetadataPaths = new HashSet<>(); activeManifestBlobMetadata.forEach(blobMetadata -> { - ClusterMetadataManifest clusterMetadataManifest = fetchRemoteClusterMetadataManifest( + ClusterMetadataManifest clusterMetadataManifest = remoteManifestManager.fetchRemoteClusterMetadataManifest( clusterName, clusterUUID, blobMetadata.name() @@ -1594,7 +884,7 @@ private void deleteClusterMetadata( } }); staleManifestBlobMetadata.forEach(blobMetadata -> { - ClusterMetadataManifest clusterMetadataManifest = fetchRemoteClusterMetadataManifest( + ClusterMetadataManifest clusterMetadataManifest = remoteManifestManager.fetchRemoteClusterMetadataManifest( clusterName, clusterUUID, blobMetadata.name() @@ -1681,7 +971,7 @@ private void deleteClusterMetadata( private void deleteStalePaths(String clusterName, String clusterUUID, List stalePaths) throws IOException { logger.debug(String.format(Locale.ROOT, "Deleting stale files from remote - %s", stalePaths)); - getBlobStoreTransferService().deleteBlobs(getCusterMetadataBasePath(clusterName, clusterUUID), stalePaths); + getBlobStoreTransferService().deleteBlobs(getCusterMetadataBasePath(blobStoreRepository, clusterName, clusterUUID), stalePaths); } /** @@ -1710,34 +1000,4 @@ public void deleteStaleClusterUUIDs(ClusterState clusterState, ClusterMetadataMa public RemotePersistenceStats getStats() { return remoteStateStats; } - - private static class UploadedMetadataResults { - List uploadedIndexMetadata; - Map uploadedCustomMetadataMap; - UploadedMetadataAttribute uploadedCoordinationMetadata; - UploadedMetadataAttribute uploadedSettingsMetadata; - UploadedMetadataAttribute uploadedTemplatesMetadata; - - public UploadedMetadataResults( - List uploadedIndexMetadata, - Map uploadedCustomMetadataMap, - UploadedMetadataAttribute uploadedCoordinationMetadata, - UploadedMetadataAttribute uploadedSettingsMetadata, - UploadedMetadataAttribute uploadedTemplatesMetadata - ) { - this.uploadedIndexMetadata = uploadedIndexMetadata; - this.uploadedCustomMetadataMap = uploadedCustomMetadataMap; - this.uploadedCoordinationMetadata = uploadedCoordinationMetadata; - this.uploadedSettingsMetadata = uploadedSettingsMetadata; - this.uploadedTemplatesMetadata = uploadedTemplatesMetadata; - } - - public UploadedMetadataResults() { - this.uploadedIndexMetadata = new ArrayList<>(); - this.uploadedCustomMetadataMap = new HashMap<>(); - this.uploadedCoordinationMetadata = null; - this.uploadedSettingsMetadata = null; - this.uploadedTemplatesMetadata = null; - } - } } diff --git a/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateUtils.java b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateUtils.java new file mode 100644 index 0000000000000..4e8fb7cf351ea --- /dev/null +++ b/server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateUtils.java @@ -0,0 +1,99 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.gateway.remote; + +import org.opensearch.cluster.metadata.Metadata; +import org.opensearch.common.blobstore.BlobContainer; +import org.opensearch.common.blobstore.BlobPath; +import org.opensearch.core.xcontent.ToXContent; +import org.opensearch.repositories.blobstore.BlobStoreRepository; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Base64; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class RemoteClusterStateUtils { + public static final String METADATA_NAME_FORMAT = "%s.dat"; + public static final String METADATA_FILE_PREFIX = "metadata"; + public static final String CLUSTER_STATE_PATH_TOKEN = "cluster-state"; + public static final String DELIMITER = "__"; + + // ToXContent Params with gateway mode. + // We are using gateway context mode to persist all custom metadata. + public static final ToXContent.Params FORMAT_PARAMS; + static { + Map params = new HashMap<>(1); + params.put(Metadata.CONTEXT_MODE_PARAM, Metadata.CONTEXT_MODE_GATEWAY); + FORMAT_PARAMS = new ToXContent.MapParams(params); + } + + public static BlobPath getCusterMetadataBasePath(BlobStoreRepository blobStoreRepository, String clusterName, String clusterUUID) { + return blobStoreRepository.basePath().add(encodeString(clusterName)).add(CLUSTER_STATE_PATH_TOKEN).add(clusterUUID); + } + + public static String encodeString(String content) { + return Base64.getUrlEncoder().withoutPadding().encodeToString(content.getBytes(StandardCharsets.UTF_8)); + } + + static BlobContainer clusterUUIDContainer(BlobStoreRepository blobStoreRepository, String clusterName) { + return blobStoreRepository.blobStore() + .blobContainer( + blobStoreRepository.basePath() + .add(Base64.getUrlEncoder().withoutPadding().encodeToString(clusterName.getBytes(StandardCharsets.UTF_8))) + .add(CLUSTER_STATE_PATH_TOKEN) + ); + } + + /** + * Exception for Remote state transfer. + */ + static class RemoteStateTransferException extends RuntimeException { + + public RemoteStateTransferException(String errorDesc) { + super(errorDesc); + } + + public RemoteStateTransferException(String errorDesc, Throwable cause) { + super(errorDesc, cause); + } + } + + static class UploadedMetadataResults { + List uploadedIndexMetadata; + Map uploadedCustomMetadataMap; + ClusterMetadataManifest.UploadedMetadataAttribute uploadedCoordinationMetadata; + ClusterMetadataManifest.UploadedMetadataAttribute uploadedSettingsMetadata; + ClusterMetadataManifest.UploadedMetadataAttribute uploadedTemplatesMetadata; + + public UploadedMetadataResults( + List uploadedIndexMetadata, + Map uploadedCustomMetadataMap, + ClusterMetadataManifest.UploadedMetadataAttribute uploadedCoordinationMetadata, + ClusterMetadataManifest.UploadedMetadataAttribute uploadedSettingsMetadata, + ClusterMetadataManifest.UploadedMetadataAttribute uploadedTemplatesMetadata + ) { + this.uploadedIndexMetadata = uploadedIndexMetadata; + this.uploadedCustomMetadataMap = uploadedCustomMetadataMap; + this.uploadedCoordinationMetadata = uploadedCoordinationMetadata; + this.uploadedSettingsMetadata = uploadedSettingsMetadata; + this.uploadedTemplatesMetadata = uploadedTemplatesMetadata; + } + + public UploadedMetadataResults() { + this.uploadedIndexMetadata = new ArrayList<>(); + this.uploadedCustomMetadataMap = new HashMap<>(); + this.uploadedCoordinationMetadata = null; + this.uploadedSettingsMetadata = null; + this.uploadedTemplatesMetadata = null; + } + } +} diff --git a/server/src/main/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManager.java b/server/src/main/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManager.java new file mode 100644 index 0000000000000..9141085b6d5a2 --- /dev/null +++ b/server/src/main/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManager.java @@ -0,0 +1,334 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.gateway.remote; + +import org.opensearch.action.LatchedActionListener; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.coordination.CoordinationMetadata; +import org.opensearch.cluster.metadata.Metadata; +import org.opensearch.cluster.metadata.TemplatesMetadata; +import org.opensearch.common.CheckedRunnable; +import org.opensearch.common.blobstore.BlobContainer; +import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.common.settings.Setting; +import org.opensearch.common.settings.Settings; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.action.ActionListener; +import org.opensearch.core.xcontent.ToXContent; +import org.opensearch.index.remote.RemoteStoreUtils; +import org.opensearch.repositories.blobstore.BlobStoreRepository; +import org.opensearch.repositories.blobstore.ChecksumBlobStoreFormat; + +import java.io.IOException; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Locale; +import java.util.Map; +import java.util.Set; + +import static java.util.Objects.requireNonNull; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.FORMAT_PARAMS; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.METADATA_FILE_PREFIX; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.METADATA_NAME_FORMAT; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.RemoteStateTransferException; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.getCusterMetadataBasePath; + +public class RemoteGlobalMetadataManager { + public static final String GLOBAL_METADATA_PATH_TOKEN = "global-metadata"; + public static final String COORDINATION_METADATA = "coordination"; + public static final String SETTING_METADATA = "settings"; + public static final String TEMPLATES_METADATA = "templates"; + public static final String CUSTOM_METADATA = "custom"; + public static final String CUSTOM_DELIMITER = "--"; + + public static final TimeValue GLOBAL_METADATA_UPLOAD_TIMEOUT_DEFAULT = TimeValue.timeValueMillis(20000); + + public static final Setting GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING = Setting.timeSetting( + "cluster.remote_store.state.global_metadata.upload_timeout", + GLOBAL_METADATA_UPLOAD_TIMEOUT_DEFAULT, + Setting.Property.Dynamic, + Setting.Property.NodeScope + ); + + public static final ChecksumBlobStoreFormat GLOBAL_METADATA_FORMAT = new ChecksumBlobStoreFormat<>( + "metadata", + METADATA_NAME_FORMAT, + Metadata::fromXContent + ); + + public static final ChecksumBlobStoreFormat COORDINATION_METADATA_FORMAT = new ChecksumBlobStoreFormat<>( + "coordination", + METADATA_NAME_FORMAT, + CoordinationMetadata::fromXContent + ); + + public static final ChecksumBlobStoreFormat SETTINGS_METADATA_FORMAT = new ChecksumBlobStoreFormat<>( + "settings", + METADATA_NAME_FORMAT, + Settings::fromXContent + ); + + public static final ChecksumBlobStoreFormat TEMPLATES_METADATA_FORMAT = new ChecksumBlobStoreFormat<>( + "templates", + METADATA_NAME_FORMAT, + TemplatesMetadata::fromXContent + ); + + public static final ChecksumBlobStoreFormat CUSTOM_METADATA_FORMAT = new ChecksumBlobStoreFormat<>( + "custom", + METADATA_NAME_FORMAT, + Metadata.Custom::fromXContent + ); + public static final int GLOBAL_METADATA_CURRENT_CODEC_VERSION = 1; + + private final BlobStoreRepository blobStoreRepository; + + private volatile TimeValue globalMetadataUploadTimeout; + + RemoteGlobalMetadataManager(BlobStoreRepository blobStoreRepository, ClusterSettings clusterSettings) { + this.blobStoreRepository = blobStoreRepository; + this.globalMetadataUploadTimeout = clusterSettings.get(GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING); + clusterSettings.addSettingsUpdateConsumer(GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING, this::setGlobalMetadataUploadTimeout); + } + + /** + * Allows async upload of Metadata components to remote + */ + CheckedRunnable getAsyncMetadataWriteAction( + ClusterState clusterState, + String component, + ChecksumBlobStoreFormat componentMetadataBlobStore, + ToXContent componentMetadata, + LatchedActionListener latchedActionListener + ) { + final BlobContainer globalMetadataContainer = globalMetadataContainer( + clusterState.getClusterName().value(), + clusterState.metadata().clusterUUID() + ); + final String componentMetadataFilename = metadataAttributeFileName(component, clusterState.metadata().version()); + ActionListener completionListener = ActionListener.wrap( + resp -> latchedActionListener.onResponse( + new ClusterMetadataManifest.UploadedMetadataAttribute(component, componentMetadataFilename) + ), + ex -> latchedActionListener.onFailure(new RemoteStateTransferException(component, ex)) + ); + return () -> componentMetadataBlobStore.writeAsyncWithUrgentPriority( + componentMetadata, + globalMetadataContainer, + componentMetadataFilename, + blobStoreRepository.getCompressor(), + completionListener, + FORMAT_PARAMS + ); + } + + Metadata getGlobalMetadata(String clusterName, String clusterUUID, ClusterMetadataManifest clusterMetadataManifest) { + String globalMetadataFileName = clusterMetadataManifest.getGlobalMetadataFileName(); + try { + // Fetch Global metadata + if (globalMetadataFileName != null) { + String[] splitPath = globalMetadataFileName.split("/"); + return GLOBAL_METADATA_FORMAT.read( + globalMetadataContainer(clusterName, clusterUUID), + splitPath[splitPath.length - 1], + blobStoreRepository.getNamedXContentRegistry() + ); + } else if (clusterMetadataManifest.hasMetadataAttributesFiles()) { + CoordinationMetadata coordinationMetadata = getCoordinationMetadata( + clusterName, + clusterUUID, + clusterMetadataManifest.getCoordinationMetadata().getUploadedFilename() + ); + Settings settingsMetadata = getSettingsMetadata( + clusterName, + clusterUUID, + clusterMetadataManifest.getSettingsMetadata().getUploadedFilename() + ); + TemplatesMetadata templatesMetadata = getTemplatesMetadata( + clusterName, + clusterUUID, + clusterMetadataManifest.getTemplatesMetadata().getUploadedFilename() + ); + Metadata.Builder builder = new Metadata.Builder(); + builder.coordinationMetadata(coordinationMetadata); + builder.persistentSettings(settingsMetadata); + builder.templates(templatesMetadata); + clusterMetadataManifest.getCustomMetadataMap() + .forEach( + (key, value) -> builder.putCustom( + key, + getCustomsMetadata(clusterName, clusterUUID, value.getUploadedFilename(), key) + ) + ); + return builder.build(); + } else { + return Metadata.EMPTY_METADATA; + } + } catch (IOException e) { + throw new IllegalStateException( + String.format(Locale.ROOT, "Error while downloading Global Metadata - %s", globalMetadataFileName), + e + ); + } + } + + private CoordinationMetadata getCoordinationMetadata(String clusterName, String clusterUUID, String coordinationMetadataFileName) { + try { + // Fetch Coordination metadata + if (coordinationMetadataFileName != null) { + String[] splitPath = coordinationMetadataFileName.split("/"); + return COORDINATION_METADATA_FORMAT.read( + globalMetadataContainer(clusterName, clusterUUID), + splitPath[splitPath.length - 1], + blobStoreRepository.getNamedXContentRegistry() + ); + } else { + return CoordinationMetadata.EMPTY_METADATA; + } + } catch (IOException e) { + throw new IllegalStateException( + String.format(Locale.ROOT, "Error while downloading Coordination Metadata - %s", coordinationMetadataFileName), + e + ); + } + } + + private Settings getSettingsMetadata(String clusterName, String clusterUUID, String settingsMetadataFileName) { + try { + // Fetch Settings metadata + if (settingsMetadataFileName != null) { + String[] splitPath = settingsMetadataFileName.split("/"); + return SETTINGS_METADATA_FORMAT.read( + globalMetadataContainer(clusterName, clusterUUID), + splitPath[splitPath.length - 1], + blobStoreRepository.getNamedXContentRegistry() + ); + } else { + return Settings.EMPTY; + } + } catch (IOException e) { + throw new IllegalStateException( + String.format(Locale.ROOT, "Error while downloading Settings Metadata - %s", settingsMetadataFileName), + e + ); + } + } + + private TemplatesMetadata getTemplatesMetadata(String clusterName, String clusterUUID, String templatesMetadataFileName) { + try { + // Fetch Templates metadata + if (templatesMetadataFileName != null) { + String[] splitPath = templatesMetadataFileName.split("/"); + return TEMPLATES_METADATA_FORMAT.read( + globalMetadataContainer(clusterName, clusterUUID), + splitPath[splitPath.length - 1], + blobStoreRepository.getNamedXContentRegistry() + ); + } else { + return TemplatesMetadata.EMPTY_METADATA; + } + } catch (IOException e) { + throw new IllegalStateException( + String.format(Locale.ROOT, "Error while downloading Templates Metadata - %s", templatesMetadataFileName), + e + ); + } + } + + private Metadata.Custom getCustomsMetadata(String clusterName, String clusterUUID, String customMetadataFileName, String custom) { + requireNonNull(customMetadataFileName); + try { + // Fetch Custom metadata + String[] splitPath = customMetadataFileName.split("/"); + ChecksumBlobStoreFormat customChecksumBlobStoreFormat = new ChecksumBlobStoreFormat<>( + "custom", + METADATA_NAME_FORMAT, + (parser -> Metadata.Custom.fromXContent(parser, custom)) + ); + return customChecksumBlobStoreFormat.read( + globalMetadataContainer(clusterName, clusterUUID), + splitPath[splitPath.length - 1], + blobStoreRepository.getNamedXContentRegistry() + ); + } catch (IOException e) { + throw new IllegalStateException( + String.format(Locale.ROOT, "Error while downloading Custom Metadata - %s", customMetadataFileName), + e + ); + } + } + + Map getUpdatedCustoms(ClusterState currentState, ClusterState previousState) { + if (Metadata.isCustomMetadataEqual(previousState.metadata(), currentState.metadata())) { + return new HashMap<>(); + } + Map updatedCustom = new HashMap<>(); + Set currentCustoms = new HashSet<>(currentState.metadata().customs().keySet()); + for (Map.Entry cursor : previousState.metadata().customs().entrySet()) { + if (cursor.getValue().context().contains(Metadata.XContentContext.GATEWAY)) { + if (currentCustoms.contains(cursor.getKey()) + && !cursor.getValue().equals(currentState.metadata().custom(cursor.getKey()))) { + // If the custom metadata is updated, we need to upload the new version. + updatedCustom.put(cursor.getKey(), currentState.metadata().custom(cursor.getKey())); + } + currentCustoms.remove(cursor.getKey()); + } + } + for (String custom : currentCustoms) { + Metadata.Custom cursor = currentState.metadata().custom(custom); + if (cursor.context().contains(Metadata.XContentContext.GATEWAY)) { + updatedCustom.put(custom, cursor); + } + } + return updatedCustom; + } + + private BlobContainer globalMetadataContainer(String clusterName, String clusterUUID) { + // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/global-metadata/ + return blobStoreRepository.blobStore() + .blobContainer(getCusterMetadataBasePath(blobStoreRepository, clusterName, clusterUUID).add(GLOBAL_METADATA_PATH_TOKEN)); + } + + private static String globalMetadataFileName(Metadata metadata) { + // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/global-metadata/metadata______ + return String.join( + DELIMITER, + METADATA_FILE_PREFIX, + RemoteStoreUtils.invertLong(metadata.version()), + RemoteStoreUtils.invertLong(System.currentTimeMillis()), + String.valueOf(GLOBAL_METADATA_CURRENT_CODEC_VERSION) + ); + } + + private static String metadataAttributeFileName(String componentPrefix, Long metadataVersion) { + // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/global-metadata/______ + return String.join( + DELIMITER, + componentPrefix, + RemoteStoreUtils.invertLong(metadataVersion), + RemoteStoreUtils.invertLong(System.currentTimeMillis()), + String.valueOf(GLOBAL_METADATA_CURRENT_CODEC_VERSION) + ); + } + + boolean isGlobalMetadataEqual(ClusterMetadataManifest first, ClusterMetadataManifest second, String clusterName) { + Metadata secondGlobalMetadata = getGlobalMetadata(clusterName, second.getClusterUUID(), second); + Metadata firstGlobalMetadata = getGlobalMetadata(clusterName, first.getClusterUUID(), first); + return Metadata.isGlobalResourcesMetadataEquals(firstGlobalMetadata, secondGlobalMetadata); + } + + private void setGlobalMetadataUploadTimeout(TimeValue newGlobalMetadataUploadTimeout) { + this.globalMetadataUploadTimeout = newGlobalMetadataUploadTimeout; + } + + public TimeValue getGlobalMetadataUploadTimeout() { + return this.globalMetadataUploadTimeout; + } +} diff --git a/server/src/main/java/org/opensearch/gateway/remote/RemoteIndexMetadataManager.java b/server/src/main/java/org/opensearch/gateway/remote/RemoteIndexMetadataManager.java new file mode 100644 index 0000000000000..c034d597817ea --- /dev/null +++ b/server/src/main/java/org/opensearch/gateway/remote/RemoteIndexMetadataManager.java @@ -0,0 +1,182 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.gateway.remote; + +import org.opensearch.action.LatchedActionListener; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.metadata.IndexMetadata; +import org.opensearch.common.CheckedRunnable; +import org.opensearch.common.blobstore.BlobContainer; +import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.common.settings.Setting; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.action.ActionListener; +import org.opensearch.index.remote.RemoteStoreUtils; +import org.opensearch.repositories.blobstore.BlobStoreRepository; +import org.opensearch.repositories.blobstore.ChecksumBlobStoreFormat; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; + +import static org.opensearch.gateway.remote.RemoteClusterStateService.INDEX_METADATA_CURRENT_CODEC_VERSION; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.FORMAT_PARAMS; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.METADATA_FILE_PREFIX; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.METADATA_NAME_FORMAT; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.RemoteStateTransferException; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.getCusterMetadataBasePath; + +public class RemoteIndexMetadataManager { + public static final TimeValue INDEX_METADATA_UPLOAD_TIMEOUT_DEFAULT = TimeValue.timeValueMillis(20000); + + public static final Setting INDEX_METADATA_UPLOAD_TIMEOUT_SETTING = Setting.timeSetting( + "cluster.remote_store.state.index_metadata.upload_timeout", + INDEX_METADATA_UPLOAD_TIMEOUT_DEFAULT, + Setting.Property.Dynamic, + Setting.Property.NodeScope + ); + + public static final ChecksumBlobStoreFormat INDEX_METADATA_FORMAT = new ChecksumBlobStoreFormat<>( + "index-metadata", + METADATA_NAME_FORMAT, + IndexMetadata::fromXContent + ); + public static final String INDEX_PATH_TOKEN = "index"; + + private final BlobStoreRepository blobStoreRepository; + + private volatile TimeValue indexMetadataUploadTimeout; + + public RemoteIndexMetadataManager(BlobStoreRepository blobStoreRepository, ClusterSettings clusterSettings) { + this.blobStoreRepository = blobStoreRepository; + this.indexMetadataUploadTimeout = clusterSettings.get(INDEX_METADATA_UPLOAD_TIMEOUT_SETTING); + clusterSettings.addSettingsUpdateConsumer(INDEX_METADATA_UPLOAD_TIMEOUT_SETTING, this::setIndexMetadataUploadTimeout); + } + + /** + * 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 + */ + CheckedRunnable getIndexMetadataAsyncAction( + ClusterState clusterState, + IndexMetadata indexMetadata, + LatchedActionListener latchedActionListener + ) { + final BlobContainer indexMetadataContainer = indexMetadataContainer( + clusterState.getClusterName().value(), + clusterState.metadata().clusterUUID(), + indexMetadata.getIndexUUID() + ); + final String indexMetadataFilename = indexMetadataFileName(indexMetadata); + ActionListener completionListener = ActionListener.wrap( + resp -> latchedActionListener.onResponse( + new ClusterMetadataManifest.UploadedIndexMetadata( + indexMetadata.getIndex().getName(), + indexMetadata.getIndexUUID(), + indexMetadataContainer.path().buildAsString() + indexMetadataFilename + ) + ), + ex -> latchedActionListener.onFailure(new RemoteStateTransferException(indexMetadata.getIndex().toString(), ex)) + ); + + return () -> INDEX_METADATA_FORMAT.writeAsyncWithUrgentPriority( + indexMetadata, + indexMetadataContainer, + indexMetadataFilename, + blobStoreRepository.getCompressor(), + completionListener, + FORMAT_PARAMS + ); + } + + /** + * 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 ClusterMetadataManifest.UploadedIndexMetadata} contains details about remote location of index metadata + * @return {@link IndexMetadata} + */ + IndexMetadata getIndexMetadata( + String clusterName, + String clusterUUID, + ClusterMetadataManifest.UploadedIndexMetadata uploadedIndexMetadata + ) { + BlobContainer blobContainer = indexMetadataContainer(clusterName, clusterUUID, uploadedIndexMetadata.getIndexUUID()); + try { + String[] splitPath = uploadedIndexMetadata.getUploadedFilename().split("/"); + return INDEX_METADATA_FORMAT.read( + blobContainer, + splitPath[splitPath.length - 1], + blobStoreRepository.getNamedXContentRegistry() + ); + } catch (IOException e) { + throw new IllegalStateException( + String.format(Locale.ROOT, "Error while downloading IndexMetadata - %s", uploadedIndexMetadata.getUploadedFilename()), + e + ); + } + } + + /** + * 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 + * @param clusterMetadataManifest manifest file of cluster + * @return {@code Map} latest IndexUUID to IndexMetadata map + */ + Map getIndexMetadataMap( + String clusterName, + String clusterUUID, + ClusterMetadataManifest clusterMetadataManifest + ) { + assert Objects.equals(clusterUUID, clusterMetadataManifest.getClusterUUID()) + : "Corrupt ClusterMetadataManifest found. Cluster UUID mismatch."; + Map remoteIndexMetadata = new HashMap<>(); + for (ClusterMetadataManifest.UploadedIndexMetadata uploadedIndexMetadata : clusterMetadataManifest.getIndices()) { + IndexMetadata indexMetadata = getIndexMetadata(clusterName, clusterUUID, uploadedIndexMetadata); + remoteIndexMetadata.put(uploadedIndexMetadata.getIndexUUID(), indexMetadata); + } + return remoteIndexMetadata; + } + + private BlobContainer indexMetadataContainer(String clusterName, String clusterUUID, String indexUUID) { + // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/index/ftqsCnn9TgOX + return blobStoreRepository.blobStore() + .blobContainer(getCusterMetadataBasePath(blobStoreRepository, clusterName, clusterUUID).add(INDEX_PATH_TOKEN).add(indexUUID)); + } + + public TimeValue getIndexMetadataUploadTimeout() { + return this.indexMetadataUploadTimeout; + } + + private void setIndexMetadataUploadTimeout(TimeValue newIndexMetadataUploadTimeout) { + this.indexMetadataUploadTimeout = newIndexMetadataUploadTimeout; + } + + static String indexMetadataFileName(IndexMetadata indexMetadata) { + // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/index//metadata______ + return String.join( + DELIMITER, + METADATA_FILE_PREFIX, + RemoteStoreUtils.invertLong(indexMetadata.getVersion()), + RemoteStoreUtils.invertLong(System.currentTimeMillis()), + String.valueOf(INDEX_METADATA_CURRENT_CODEC_VERSION) // Keep the codec version at last place only, during read we reads last + // place to determine codec version. + ); + } +} diff --git a/server/src/main/java/org/opensearch/gateway/remote/RemoteManifestManager.java b/server/src/main/java/org/opensearch/gateway/remote/RemoteManifestManager.java new file mode 100644 index 0000000000000..c741989552462 --- /dev/null +++ b/server/src/main/java/org/opensearch/gateway/remote/RemoteManifestManager.java @@ -0,0 +1,357 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.gateway.remote; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.opensearch.Version; +import org.opensearch.action.LatchedActionListener; +import org.opensearch.cluster.ClusterState; +import org.opensearch.common.blobstore.BlobContainer; +import org.opensearch.common.blobstore.BlobMetadata; +import org.opensearch.common.blobstore.BlobPath; +import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.common.settings.Setting; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.action.ActionListener; +import org.opensearch.index.remote.RemoteStoreUtils; +import org.opensearch.repositories.blobstore.BlobStoreRepository; +import org.opensearch.repositories.blobstore.ChecksumBlobStoreFormat; + +import java.io.IOException; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.FORMAT_PARAMS; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.RemoteStateTransferException; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.getCusterMetadataBasePath; + +public class RemoteManifestManager { + public static final String MANIFEST_PATH_TOKEN = "manifest"; + public static final String MANIFEST_FILE_PREFIX = "manifest"; + public static final String METADATA_MANIFEST_NAME_FORMAT = "%s"; + public static final int MANIFEST_CURRENT_CODEC_VERSION = ClusterMetadataManifest.CODEC_V2; + public static final int SPLITED_MANIFEST_FILE_LENGTH = 6; // file name manifest__term__version__C/P__timestamp__codecversion + + public static final TimeValue METADATA_MANIFEST_UPLOAD_TIMEOUT_DEFAULT = TimeValue.timeValueMillis(20000); + + public static final Setting METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING = Setting.timeSetting( + "cluster.remote_store.state.metadata_manifest.upload_timeout", + METADATA_MANIFEST_UPLOAD_TIMEOUT_DEFAULT, + Setting.Property.Dynamic, + Setting.Property.NodeScope + ); + + /** + * Manifest format compatible with older codec v0, where codec version was missing. + */ + public static final ChecksumBlobStoreFormat CLUSTER_METADATA_MANIFEST_FORMAT_V0 = + new ChecksumBlobStoreFormat<>("cluster-metadata-manifest", METADATA_MANIFEST_NAME_FORMAT, ClusterMetadataManifest::fromXContentV0); + /** + * Manifest format compatible with older codec v1, where global metadata was missing. + */ + public static final ChecksumBlobStoreFormat CLUSTER_METADATA_MANIFEST_FORMAT_V1 = + new ChecksumBlobStoreFormat<>("cluster-metadata-manifest", METADATA_MANIFEST_NAME_FORMAT, ClusterMetadataManifest::fromXContentV1); + + /** + * Manifest format compatible with codec v2, where we introduced codec versions/global metadata. + */ + public static final ChecksumBlobStoreFormat CLUSTER_METADATA_MANIFEST_FORMAT = new ChecksumBlobStoreFormat<>( + "cluster-metadata-manifest", + METADATA_MANIFEST_NAME_FORMAT, + ClusterMetadataManifest::fromXContent + ); + + private final BlobStoreRepository blobStoreRepository; + private volatile TimeValue metadataManifestUploadTimeout; + private final String nodeId; + private static final Logger logger = LogManager.getLogger(RemoteManifestManager.class); + + RemoteManifestManager(BlobStoreRepository blobStoreRepository, ClusterSettings clusterSettings, String nodeId) { + this.blobStoreRepository = blobStoreRepository; + this.metadataManifestUploadTimeout = clusterSettings.get(METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING); + this.nodeId = nodeId; + clusterSettings.addSettingsUpdateConsumer(METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING, this::setMetadataManifestUploadTimeout); + } + + private ClusterMetadataManifest uploadV1Manifest( + ClusterState clusterState, + List uploadedIndexMetadata, + String previousClusterUUID, + String globalMetadataFileName, + boolean committed + ) throws IOException { + synchronized (this) { + final String manifestFileName = getManifestFileName( + clusterState.term(), + clusterState.version(), + committed, + ClusterMetadataManifest.CODEC_V1 + ); + ClusterMetadataManifest manifest = ClusterMetadataManifest.builder() + .clusterTerm(clusterState.term()) + .stateVersion(clusterState.getVersion()) + .clusterUUID(clusterState.metadata().clusterUUID()) + .stateUUID(clusterState.stateUUID()) + .opensearchVersion(Version.CURRENT) + .nodeId(nodeId) + .committed(committed) + .codecVersion(ClusterMetadataManifest.CODEC_V1) + .globalMetadataFileName(globalMetadataFileName) + .indices(uploadedIndexMetadata) + .previousClusterUUID(previousClusterUUID) + .clusterUUIDCommitted(clusterState.metadata().clusterUUIDCommitted()) + .build(); + writeMetadataManifest(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID(), manifest, manifestFileName); + return manifest; + } + } + + ClusterMetadataManifest uploadManifest( + ClusterState clusterState, + List uploadedIndexMetadata, + String previousClusterUUID, + ClusterMetadataManifest.UploadedMetadataAttribute uploadedCoordinationMetadata, + ClusterMetadataManifest.UploadedMetadataAttribute uploadedSettingsMetadata, + ClusterMetadataManifest.UploadedMetadataAttribute uploadedTemplatesMetadata, + Map uploadedCustomMetadataMap, + boolean committed + ) throws IOException { + synchronized (this) { + final String manifestFileName = getManifestFileName( + clusterState.term(), + clusterState.version(), + committed, + MANIFEST_CURRENT_CODEC_VERSION + ); + final ClusterMetadataManifest manifest = new ClusterMetadataManifest( + clusterState.term(), + clusterState.getVersion(), + clusterState.metadata().clusterUUID(), + clusterState.stateUUID(), + Version.CURRENT, + nodeId, + committed, + MANIFEST_CURRENT_CODEC_VERSION, + null, + uploadedIndexMetadata, + previousClusterUUID, + clusterState.metadata().clusterUUIDCommitted(), + uploadedCoordinationMetadata, + uploadedSettingsMetadata, + uploadedTemplatesMetadata, + uploadedCustomMetadataMap + ); + writeMetadataManifest(clusterState.getClusterName().value(), clusterState.metadata().clusterUUID(), manifest, manifestFileName); + return manifest; + } + } + + private void writeMetadataManifest(String clusterName, String clusterUUID, ClusterMetadataManifest uploadManifest, String fileName) + throws IOException { + AtomicReference result = new AtomicReference(); + AtomicReference exceptionReference = new AtomicReference(); + + final BlobContainer metadataManifestContainer = manifestContainer(clusterName, clusterUUID); + + // latch to wait until upload is not finished + CountDownLatch latch = new CountDownLatch(1); + + LatchedActionListener completionListener = new LatchedActionListener<>(ActionListener.wrap(resp -> { + logger.trace(String.format(Locale.ROOT, "Manifest file uploaded successfully.")); + }, ex -> { exceptionReference.set(ex); }), latch); + + getClusterMetadataManifestBlobStoreFormat(fileName).writeAsyncWithUrgentPriority( + uploadManifest, + metadataManifestContainer, + fileName, + blobStoreRepository.getCompressor(), + completionListener, + FORMAT_PARAMS + ); + + try { + if (latch.await(getMetadataManifestUploadTimeout().millis(), TimeUnit.MILLISECONDS) == false) { + RemoteStateTransferException ex = new RemoteStateTransferException( + String.format(Locale.ROOT, "Timed out waiting for transfer of manifest file to complete") + ); + throw ex; + } + } catch (InterruptedException ex) { + RemoteStateTransferException exception = new RemoteStateTransferException( + String.format(Locale.ROOT, "Timed out waiting for transfer of manifest file to complete - %s"), + ex + ); + Thread.currentThread().interrupt(); + throw exception; + } + if (exceptionReference.get() != null) { + throw new RemoteStateTransferException(exceptionReference.get().getMessage(), exceptionReference.get()); + } + logger.debug( + "Metadata manifest file [{}] written during [{}] phase. ", + fileName, + uploadManifest.isCommitted() ? "commit" : "publish" + ); + } + + /** + * 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 + */ + public Optional getLatestClusterMetadataManifest(String clusterName, String clusterUUID) { + Optional latestManifestFileName = getLatestManifestFileName(clusterName, clusterUUID); + return latestManifestFileName.map(s -> fetchRemoteClusterMetadataManifest(clusterName, clusterUUID, s)); + } + + /** + * 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 + */ + ClusterMetadataManifest fetchRemoteClusterMetadataManifest(String clusterName, String clusterUUID, String filename) + throws IllegalStateException { + try { + return getClusterMetadataManifestBlobStoreFormat(filename).read( + manifestContainer(clusterName, clusterUUID), + filename, + blobStoreRepository.getNamedXContentRegistry() + ); + } catch (IOException e) { + throw new IllegalStateException(String.format(Locale.ROOT, "Error while downloading cluster metadata - %s", filename), e); + } + } + + Map getLatestManifestForAllClusterUUIDs(String clusterName, Set clusterUUIDs) { + Map manifestsByClusterUUID = new HashMap<>(); + for (String clusterUUID : clusterUUIDs) { + try { + Optional manifest = getLatestClusterMetadataManifest(clusterName, clusterUUID); + manifest.ifPresent(clusterMetadataManifest -> manifestsByClusterUUID.put(clusterUUID, clusterMetadataManifest)); + } catch (Exception e) { + throw new IllegalStateException( + String.format(Locale.ROOT, "Exception in fetching manifest for clusterUUID: %s", clusterUUID), + e + ); + } + } + return manifestsByClusterUUID; + } + + private ChecksumBlobStoreFormat getClusterMetadataManifestBlobStoreFormat(String fileName) { + long codecVersion = getManifestCodecVersion(fileName); + if (codecVersion == MANIFEST_CURRENT_CODEC_VERSION) { + return CLUSTER_METADATA_MANIFEST_FORMAT; + } else if (codecVersion == ClusterMetadataManifest.CODEC_V1) { + return CLUSTER_METADATA_MANIFEST_FORMAT_V1; + } else if (codecVersion == ClusterMetadataManifest.CODEC_V0) { + return CLUSTER_METADATA_MANIFEST_FORMAT_V0; + } + + throw new IllegalArgumentException("Cluster metadata manifest file is corrupted, don't have valid codec version"); + } + + private int getManifestCodecVersion(String fileName) { + String[] splitName = fileName.split(DELIMITER); + if (splitName.length == SPLITED_MANIFEST_FILE_LENGTH) { + return Integer.parseInt(splitName[splitName.length - 1]); // Last value would be codec version. + } else if (splitName.length < SPLITED_MANIFEST_FILE_LENGTH) { // Where codec is not part of file name, i.e. default codec version 0 + // is used. + return ClusterMetadataManifest.CODEC_V0; + } else { + throw new IllegalArgumentException("Manifest file name is corrupted"); + } + } + + private BlobContainer manifestContainer(String clusterName, String clusterUUID) { + // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/manifest + return blobStoreRepository.blobStore().blobContainer(getManifestFolderPath(clusterName, clusterUUID)); + } + + BlobPath getManifestFolderPath(String clusterName, String clusterUUID) { + return getCusterMetadataBasePath(blobStoreRepository, clusterName, clusterUUID).add(MANIFEST_PATH_TOKEN); + } + + public TimeValue getMetadataManifestUploadTimeout() { + return this.metadataManifestUploadTimeout; + } + + private void setMetadataManifestUploadTimeout(TimeValue newMetadataManifestUploadTimeout) { + this.metadataManifestUploadTimeout = newMetadataManifestUploadTimeout; + } + + /** + * Fetch ClusterMetadataManifest files from remote state store in order + * + * @param clusterUUID uuid of cluster state to refer to in remote + * @param clusterName name of the cluster + * @param limit max no of files to fetch + * @return all manifest file names + */ + private List getManifestFileNames(String clusterName, String clusterUUID, int limit) throws IllegalStateException { + try { + + /* + {@link BlobContainer#listBlobsByPrefixInSortedOrder} will list the latest manifest file first + as the manifest file name generated via {@link RemoteClusterStateService#getManifestFileName} ensures + when sorted in LEXICOGRAPHIC order the latest uploaded manifest file comes on top. + */ + return manifestContainer(clusterName, clusterUUID).listBlobsByPrefixInSortedOrder( + MANIFEST_FILE_PREFIX + DELIMITER, + limit, + BlobContainer.BlobNameSortOrder.LEXICOGRAPHIC + ); + } catch (IOException e) { + throw new IllegalStateException("Error while fetching latest manifest file for remote cluster state", e); + } + } + + static String getManifestFileName(long term, long version, boolean committed, int codecVersion) { + // 123456789012_test-cluster/cluster-state/dsgYj10Nkso7/manifest/manifest______C/P____ + return String.join( + DELIMITER, + MANIFEST_PATH_TOKEN, + RemoteStoreUtils.invertLong(term), + RemoteStoreUtils.invertLong(version), + (committed ? "C" : "P"), // C for committed and P for published + RemoteStoreUtils.invertLong(System.currentTimeMillis()), + String.valueOf(codecVersion) // Keep the codec version at last place only, during read we reads last place to + // determine codec version. + ); + } + + /** + * 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 + */ + private Optional getLatestManifestFileName(String clusterName, String clusterUUID) throws IllegalStateException { + List manifestFilesMetadata = getManifestFileNames(clusterName, clusterUUID, 1); + if (manifestFilesMetadata != null && !manifestFilesMetadata.isEmpty()) { + return Optional.of(manifestFilesMetadata.get(0).name()); + } + logger.info("No manifest file present in remote store for cluster name: {}, cluster UUID: {}", clusterName, clusterUUID); + return Optional.empty(); + } +} 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 0b3cd49140939..94d71dc9eaa75 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/ClusterMetadataManifestTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/ClusterMetadataManifestTests.java @@ -104,21 +104,22 @@ public void testClusterMetadataManifestXContent() throws IOException { Collections.singletonList(uploadedIndexMetadata), "prev-cluster-uuid", true, - new UploadedMetadataAttribute(RemoteClusterStateService.COORDINATION_METADATA, "coordination-file"), - new UploadedMetadataAttribute(RemoteClusterStateService.SETTING_METADATA, "setting-file"), - new UploadedMetadataAttribute(RemoteClusterStateService.TEMPLATES_METADATA, "templates-file"), + new UploadedMetadataAttribute(RemoteGlobalMetadataManager.COORDINATION_METADATA, "coordination-file"), + new UploadedMetadataAttribute(RemoteGlobalMetadataManager.SETTING_METADATA, "setting-file"), + new UploadedMetadataAttribute(RemoteGlobalMetadataManager.TEMPLATES_METADATA, "templates-file"), Collections.unmodifiableList( Arrays.asList( new UploadedMetadataAttribute( - RemoteClusterStateService.CUSTOM_METADATA + RemoteClusterStateService.CUSTOM_DELIMITER + RepositoriesMetadata.TYPE, + RemoteGlobalMetadataManager.CUSTOM_METADATA + RemoteGlobalMetadataManager.CUSTOM_DELIMITER + + RepositoriesMetadata.TYPE, "custom--repositories-file" ), new UploadedMetadataAttribute( - RemoteClusterStateService.CUSTOM_METADATA + RemoteClusterStateService.CUSTOM_DELIMITER + IndexGraveyard.TYPE, + RemoteGlobalMetadataManager.CUSTOM_METADATA + RemoteGlobalMetadataManager.CUSTOM_DELIMITER + IndexGraveyard.TYPE, "custom--index_graveyard-file" ), new UploadedMetadataAttribute( - RemoteClusterStateService.CUSTOM_METADATA + RemoteClusterStateService.CUSTOM_DELIMITER + RemoteGlobalMetadataManager.CUSTOM_METADATA + RemoteGlobalMetadataManager.CUSTOM_DELIMITER + WeightedRoutingMetadata.TYPE, "custom--weighted_routing_netadata-file" ) @@ -150,21 +151,22 @@ public void testClusterMetadataManifestSerializationEqualsHashCode() { randomUploadedIndexMetadataList(), "yfObdx8KSMKKrXf8UyHhM", true, - new UploadedMetadataAttribute(RemoteClusterStateService.COORDINATION_METADATA, "coordination-file"), - new UploadedMetadataAttribute(RemoteClusterStateService.SETTING_METADATA, "setting-file"), - new UploadedMetadataAttribute(RemoteClusterStateService.TEMPLATES_METADATA, "templates-file"), + new UploadedMetadataAttribute(RemoteGlobalMetadataManager.COORDINATION_METADATA, "coordination-file"), + new UploadedMetadataAttribute(RemoteGlobalMetadataManager.SETTING_METADATA, "setting-file"), + new UploadedMetadataAttribute(RemoteGlobalMetadataManager.TEMPLATES_METADATA, "templates-file"), Collections.unmodifiableList( Arrays.asList( new UploadedMetadataAttribute( - RemoteClusterStateService.CUSTOM_METADATA + RemoteClusterStateService.CUSTOM_DELIMITER + RepositoriesMetadata.TYPE, + RemoteGlobalMetadataManager.CUSTOM_METADATA + RemoteGlobalMetadataManager.CUSTOM_DELIMITER + + RepositoriesMetadata.TYPE, "custom--repositories-file" ), new UploadedMetadataAttribute( - RemoteClusterStateService.CUSTOM_METADATA + RemoteClusterStateService.CUSTOM_DELIMITER + IndexGraveyard.TYPE, + RemoteGlobalMetadataManager.CUSTOM_METADATA + RemoteGlobalMetadataManager.CUSTOM_DELIMITER + IndexGraveyard.TYPE, "custom--index_graveyard-file" ), new UploadedMetadataAttribute( - RemoteClusterStateService.CUSTOM_METADATA + RemoteClusterStateService.CUSTOM_DELIMITER + RemoteGlobalMetadataManager.CUSTOM_METADATA + RemoteGlobalMetadataManager.CUSTOM_DELIMITER + WeightedRoutingMetadata.TYPE, "custom--weighted_routing_netadata-file" ) 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 e4b41d30ba677..0327ddc887f2e 100644 --- a/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java +++ b/server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateServiceTests.java @@ -41,7 +41,6 @@ import org.opensearch.core.xcontent.NamedXContentRegistry; import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedIndexMetadata; import org.opensearch.gateway.remote.ClusterMetadataManifest.UploadedMetadataAttribute; -import org.opensearch.index.remote.RemoteStoreUtils; import org.opensearch.indices.IndicesModule; import org.opensearch.repositories.FilterRepository; import org.opensearch.repositories.RepositoriesService; @@ -85,16 +84,15 @@ import org.mockito.ArgumentMatchers; import static java.util.stream.Collectors.toList; -import static org.opensearch.gateway.remote.RemoteClusterStateService.COORDINATION_METADATA; -import static org.opensearch.gateway.remote.RemoteClusterStateService.DELIMITER; -import static org.opensearch.gateway.remote.RemoteClusterStateService.FORMAT_PARAMS; -import static org.opensearch.gateway.remote.RemoteClusterStateService.INDEX_METADATA_CURRENT_CODEC_VERSION; -import static org.opensearch.gateway.remote.RemoteClusterStateService.MANIFEST_CURRENT_CODEC_VERSION; -import static org.opensearch.gateway.remote.RemoteClusterStateService.MANIFEST_FILE_PREFIX; -import static org.opensearch.gateway.remote.RemoteClusterStateService.METADATA_FILE_PREFIX; import static org.opensearch.gateway.remote.RemoteClusterStateService.RETAINED_MANIFESTS; -import static org.opensearch.gateway.remote.RemoteClusterStateService.SETTING_METADATA; -import static org.opensearch.gateway.remote.RemoteClusterStateService.TEMPLATES_METADATA; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.FORMAT_PARAMS; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.RemoteStateTransferException; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.COORDINATION_METADATA; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.SETTING_METADATA; +import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.TEMPLATES_METADATA; +import static org.opensearch.gateway.remote.RemoteManifestManager.MANIFEST_CURRENT_CODEC_VERSION; +import static org.opensearch.gateway.remote.RemoteManifestManager.MANIFEST_FILE_PREFIX; 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; @@ -294,7 +292,7 @@ public void testWriteFullMetadataInParallelSuccess() throws IOException { .provideStream(0) .getInputStream() .readAllBytes(); - IndexMetadata writtenIndexMetadata = RemoteClusterStateService.INDEX_METADATA_FORMAT.deserialize( + IndexMetadata writtenIndexMetadata = RemoteIndexMetadataManager.INDEX_METADATA_FORMAT.deserialize( capturedWriteContext.get("metadata").getFileName(), blobStoreRepository.getNamedXContentRegistry(), new BytesArray(writtenBytes) @@ -333,7 +331,7 @@ public void run() { remoteClusterStateService.start(); assertThrows( - RemoteClusterStateService.RemoteStateTransferException.class, + RemoteStateTransferException.class, () -> remoteClusterStateService.writeFullMetadata(clusterState, randomAlphaOfLength(10)) ); } @@ -366,7 +364,7 @@ public void testTimeoutWhileWritingManifestFile() throws IOException { try { remoteClusterStateService.writeFullMetadata(clusterState, randomAlphaOfLength(10)); } catch (Exception e) { - assertTrue(e instanceof RemoteClusterStateService.RemoteStateTransferException); + assertTrue(e instanceof RemoteStateTransferException); assertTrue(e.getMessage().contains("Timed out waiting for transfer of following metadata to complete")); } } @@ -387,7 +385,7 @@ public void testWriteFullMetadataInParallelFailureForIndexMetadata() throws IOEx remoteClusterStateService.start(); assertThrows( - RemoteClusterStateService.RemoteStateTransferException.class, + RemoteStateTransferException.class, () -> remoteClusterStateService.writeFullMetadata(clusterState, randomAlphaOfLength(10)) ); assertEquals(0, remoteClusterStateService.getStats().getSuccessCount()); @@ -828,24 +826,6 @@ private void verifyMetadataAttributeOnlyUpdated( assertions.accept(initialManifest, manifestAfterMetadataUpdate); } - public void testReadLatestMetadataManifestFailedIOException() throws IOException { - final ClusterState clusterState = generateClusterStateWithOneIndex().nodes(nodesWithLocalNodeClusterManager()).build(); - - BlobContainer blobContainer = mockBlobStoreObjects(); - when(blobContainer.listBlobsByPrefixInSortedOrder("manifest" + DELIMITER, 1, BlobContainer.BlobNameSortOrder.LEXICOGRAPHIC)) - .thenThrow(IOException.class); - - remoteClusterStateService.start(); - Exception e = assertThrows( - IllegalStateException.class, - () -> remoteClusterStateService.getLatestClusterMetadataManifest( - clusterState.getClusterName().value(), - clusterState.metadata().clusterUUID() - ) - ); - assertEquals(e.getMessage(), "Error while fetching latest manifest file for remote cluster state"); - } - public void testReadLatestMetadataManifestFailedNoManifestFileInRemote() throws IOException { final ClusterState clusterState = generateClusterStateWithOneIndex().nodes(nodesWithLocalNodeClusterManager()).build(); @@ -986,10 +966,10 @@ public void testReadGlobalMetadata() throws IOException { .stateUUID("state-uuid") .clusterUUID("cluster-uuid") .codecVersion(MANIFEST_CURRENT_CODEC_VERSION) - .coordinationMetadata(new UploadedMetadataAttribute(COORDINATION_METADATA, "mock-coordination-file")) - .settingMetadata(new UploadedMetadataAttribute(SETTING_METADATA, "mock-setting-file")) - .templatesMetadata(new UploadedMetadataAttribute(TEMPLATES_METADATA, "mock-templates-file")) - .put(IndexGraveyard.TYPE, new UploadedMetadataAttribute(IndexGraveyard.TYPE, "mock-custom-" +IndexGraveyard.TYPE+ "-file")) + .coordinationMetadata(new ClusterMetadataManifest.UploadedMetadataAttribute(COORDINATION_METADATA, "mock-coordination-file")) + .settingMetadata(new ClusterMetadataManifest.UploadedMetadataAttribute(SETTING_METADATA, "mock-setting-file")) + .templatesMetadata(new ClusterMetadataManifest.UploadedMetadataAttribute(TEMPLATES_METADATA, "mock-templates-file")) + .put(IndexGraveyard.TYPE, new ClusterMetadataManifest.UploadedMetadataAttribute(IndexGraveyard.TYPE, "mock-custom-" +IndexGraveyard.TYPE+ "-file")) .nodeId("nodeA") .opensearchVersion(VersionUtils.randomOpenSearchVersion(random())) .previousClusterUUID("prev-cluster-uuid") @@ -998,7 +978,7 @@ public void testReadGlobalMetadata() throws IOException { Metadata expactedMetadata = Metadata.builder().persistentSettings(Settings.builder().put("readonly", true).build()).build(); mockBlobContainerForGlobalMetadata(mockBlobStoreObjects(), expectedManifest, expactedMetadata); - ClusterState newClusterState = remoteClusterStateService.getLatestClusterState( + ClusterState newClusterState = remoteClusterStateService.getLatestClusterState( clusterState.getClusterName().value(), clusterState.metadata().clusterUUID() ); @@ -1036,7 +1016,7 @@ public void testReadGlobalMetadataIOException() throws IOException { BlobContainer blobContainer = mockBlobStoreObjects(); mockBlobContainerForGlobalMetadata(blobContainer, expectedManifest, expactedMetadata); - when(blobContainer.readBlob(RemoteClusterStateService.GLOBAL_METADATA_FORMAT.blobName(globalIndexMetadataName))).thenThrow( + when(blobContainer.readBlob(RemoteGlobalMetadataManager.GLOBAL_METADATA_FORMAT.blobName(globalIndexMetadataName))).thenThrow( FileNotFoundException.class ); @@ -1300,46 +1280,6 @@ public void testRemoteStateCleanupFailureStats() throws IOException { } } - public void testFileNames() { - final Index index = new Index("test-index", "index-uuid"); - final Settings idxSettings = Settings.builder() - .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT) - .put(IndexMetadata.SETTING_INDEX_UUID, index.getUUID()) - .build(); - final IndexMetadata indexMetadata = new IndexMetadata.Builder(index.getName()).settings(idxSettings) - .numberOfShards(1) - .numberOfReplicas(0) - .build(); - - String indexMetadataFileName = RemoteClusterStateService.indexMetadataFileName(indexMetadata); - String[] splittedIndexMetadataFileName = indexMetadataFileName.split(DELIMITER); - assertThat(indexMetadataFileName.split(DELIMITER).length, is(4)); - assertThat(splittedIndexMetadataFileName[0], is(METADATA_FILE_PREFIX)); - assertThat(splittedIndexMetadataFileName[1], is(RemoteStoreUtils.invertLong(indexMetadata.getVersion()))); - assertThat(splittedIndexMetadataFileName[3], is(String.valueOf(INDEX_METADATA_CURRENT_CODEC_VERSION))); - - verifyManifestFileNameWithCodec(MANIFEST_CURRENT_CODEC_VERSION); - verifyManifestFileNameWithCodec(ClusterMetadataManifest.CODEC_V1); - verifyManifestFileNameWithCodec(ClusterMetadataManifest.CODEC_V0); - } - - private void verifyManifestFileNameWithCodec(int codecVersion) { - int term = randomIntBetween(5, 10); - int version = randomIntBetween(5, 10); - String manifestFileName = RemoteClusterStateService.getManifestFileName(term, version, true, codecVersion); - assertThat(manifestFileName.split(DELIMITER).length, is(6)); - String[] splittedName = manifestFileName.split(DELIMITER); - assertThat(splittedName[0], is(MANIFEST_FILE_PREFIX)); - assertThat(splittedName[1], is(RemoteStoreUtils.invertLong(term))); - assertThat(splittedName[2], is(RemoteStoreUtils.invertLong(version))); - assertThat(splittedName[3], is("C")); - assertThat(splittedName[5], is(String.valueOf(codecVersion))); - - manifestFileName = RemoteClusterStateService.getManifestFileName(term, version, false, codecVersion); - splittedName = manifestFileName.split(DELIMITER); - assertThat(splittedName[3], is("P")); - } - public void testSingleConcurrentExecutionOfStaleManifestCleanup() throws Exception { BlobContainer blobContainer = mock(BlobContainer.class); BlobPath blobPath = new BlobPath().add("random-path"); @@ -1370,54 +1310,6 @@ public void testSingleConcurrentExecutionOfStaleManifestCleanup() throws Excepti assertBusy(() -> assertEquals(1, callCount.get())); } - public void testIndexMetadataUploadWaitTimeSetting() { - // verify default value - assertEquals( - RemoteClusterStateService.INDEX_METADATA_UPLOAD_TIMEOUT_DEFAULT, - remoteClusterStateService.getIndexMetadataUploadTimeout() - ); - - // verify update index metadata upload timeout - int indexMetadataUploadTimeout = randomIntBetween(1, 10); - Settings newSettings = Settings.builder() - .put("cluster.remote_store.state.index_metadata.upload_timeout", indexMetadataUploadTimeout + "s") - .build(); - clusterSettings.applySettings(newSettings); - assertEquals(indexMetadataUploadTimeout, remoteClusterStateService.getIndexMetadataUploadTimeout().seconds()); - } - - public void testMetadataManifestUploadWaitTimeSetting() { - // verify default value - assertEquals( - RemoteClusterStateService.METADATA_MANIFEST_UPLOAD_TIMEOUT_DEFAULT, - remoteClusterStateService.getMetadataManifestUploadTimeout() - ); - - // verify update metadata manifest upload timeout - int metadataManifestUploadTimeout = randomIntBetween(1, 10); - Settings newSettings = Settings.builder() - .put("cluster.remote_store.state.metadata_manifest.upload_timeout", metadataManifestUploadTimeout + "s") - .build(); - clusterSettings.applySettings(newSettings); - assertEquals(metadataManifestUploadTimeout, remoteClusterStateService.getMetadataManifestUploadTimeout().seconds()); - } - - public void testGlobalMetadataUploadWaitTimeSetting() { - // verify default value - assertEquals( - RemoteClusterStateService.GLOBAL_METADATA_UPLOAD_TIMEOUT_DEFAULT, - remoteClusterStateService.getGlobalMetadataUploadTimeout() - ); - - // verify update global metadata upload timeout - int globalMetadataUploadTimeout = randomIntBetween(1, 10); - Settings newSettings = Settings.builder() - .put("cluster.remote_store.state.global_metadata.upload_timeout", globalMetadataUploadTimeout + "s") - .build(); - clusterSettings.applySettings(newSettings); - assertEquals(globalMetadataUploadTimeout, remoteClusterStateService.getGlobalMetadataUploadTimeout().seconds()); - } - private void mockObjectsForGettingPreviousClusterUUID(Map clusterUUIDsPointers) throws IOException { mockObjectsForGettingPreviousClusterUUID(clusterUUIDsPointers, false, Collections.emptyMap()); } @@ -1664,7 +1556,7 @@ private void mockBlobContainer( when(blobContainer.listBlobsByPrefixInSortedOrder("manifest" + DELIMITER, 1, BlobContainer.BlobNameSortOrder.LEXICOGRAPHIC)) .thenReturn(Arrays.asList(blobMetadata)); - BytesReference bytes = RemoteClusterStateService.CLUSTER_METADATA_MANIFEST_FORMAT.serialize( + BytesReference bytes = RemoteManifestManager.CLUSTER_METADATA_MANIFEST_FORMAT.serialize( clusterMetadataManifest, manifestFileName, blobStoreRepository.getCompressor(), @@ -1680,7 +1572,7 @@ private void mockBlobContainer( } String fileName = uploadedIndexMetadata.getUploadedFilename(); when(blobContainer.readBlob(fileName + ".dat")).thenAnswer((invocationOnMock) -> { - BytesReference bytesIndexMetadata = RemoteClusterStateService.INDEX_METADATA_FORMAT.serialize( + BytesReference bytesIndexMetadata = RemoteIndexMetadataManager.INDEX_METADATA_FORMAT.serialize( indexMetadata, fileName, blobStoreRepository.getCompressor(), @@ -1702,15 +1594,10 @@ private void mockBlobContainerForGlobalMetadata( int codecVersion = clusterMetadataManifest.getCodecVersion(); String mockManifestFileName = "manifest__1__2__C__456__" + codecVersion; BlobMetadata blobMetadata = new PlainBlobMetadata(mockManifestFileName, 1); - when( - blobContainer.listBlobsByPrefixInSortedOrder( - "manifest" + RemoteClusterStateService.DELIMITER, - 1, - BlobContainer.BlobNameSortOrder.LEXICOGRAPHIC - ) - ).thenReturn(Arrays.asList(blobMetadata)); + when(blobContainer.listBlobsByPrefixInSortedOrder("manifest" + DELIMITER, 1, BlobContainer.BlobNameSortOrder.LEXICOGRAPHIC)) + .thenReturn(Arrays.asList(blobMetadata)); - BytesReference bytes = RemoteClusterStateService.CLUSTER_METADATA_MANIFEST_FORMAT.serialize( + BytesReference bytes = RemoteManifestManager.CLUSTER_METADATA_MANIFEST_FORMAT.serialize( clusterMetadataManifest, mockManifestFileName, blobStoreRepository.getCompressor(), @@ -1719,22 +1606,21 @@ private void mockBlobContainerForGlobalMetadata( when(blobContainer.readBlob(mockManifestFileName)).thenReturn(new ByteArrayInputStream(bytes.streamInput().readAllBytes())); if (codecVersion >= ClusterMetadataManifest.CODEC_V2) { String coordinationFileName = getFileNameFromPath(clusterMetadataManifest.getCoordinationMetadata().getUploadedFilename()); - when(blobContainer.readBlob(RemoteClusterStateService.COORDINATION_METADATA_FORMAT.blobName(coordinationFileName))).thenAnswer( - (invocationOnMock) -> { - BytesReference bytesReference = RemoteClusterStateService.COORDINATION_METADATA_FORMAT.serialize( + when(blobContainer.readBlob(RemoteGlobalMetadataManager.COORDINATION_METADATA_FORMAT.blobName(coordinationFileName))) + .thenAnswer((invocationOnMock) -> { + BytesReference bytesReference = RemoteGlobalMetadataManager.COORDINATION_METADATA_FORMAT.serialize( metadata.coordinationMetadata(), coordinationFileName, blobStoreRepository.getCompressor(), FORMAT_PARAMS ); return new ByteArrayInputStream(bytesReference.streamInput().readAllBytes()); - } - ); + }); String settingsFileName = getFileNameFromPath(clusterMetadataManifest.getSettingsMetadata().getUploadedFilename()); - when(blobContainer.readBlob(RemoteClusterStateService.SETTINGS_METADATA_FORMAT.blobName(settingsFileName))).thenAnswer( + when(blobContainer.readBlob(RemoteGlobalMetadataManager.SETTINGS_METADATA_FORMAT.blobName(settingsFileName))).thenAnswer( (invocationOnMock) -> { - BytesReference bytesReference = RemoteClusterStateService.SETTINGS_METADATA_FORMAT.serialize( + BytesReference bytesReference = RemoteGlobalMetadataManager.SETTINGS_METADATA_FORMAT.serialize( metadata.persistentSettings(), settingsFileName, blobStoreRepository.getCompressor(), @@ -1745,9 +1631,9 @@ private void mockBlobContainerForGlobalMetadata( ); String templatesFileName = getFileNameFromPath(clusterMetadataManifest.getTemplatesMetadata().getUploadedFilename()); - when(blobContainer.readBlob(RemoteClusterStateService.TEMPLATES_METADATA_FORMAT.blobName(templatesFileName))).thenAnswer( + when(blobContainer.readBlob(RemoteGlobalMetadataManager.TEMPLATES_METADATA_FORMAT.blobName(templatesFileName))).thenAnswer( (invocationOnMock) -> { - BytesReference bytesReference = RemoteClusterStateService.TEMPLATES_METADATA_FORMAT.serialize( + BytesReference bytesReference = RemoteGlobalMetadataManager.TEMPLATES_METADATA_FORMAT.serialize( metadata.templatesMetadata(), templatesFileName, blobStoreRepository.getCompressor(), @@ -1765,9 +1651,9 @@ private void mockBlobContainerForGlobalMetadata( for (Map.Entry entry : customFileMap.entrySet()) { String custom = entry.getKey(); String fileName = entry.getValue(); - when(blobContainer.readBlob(RemoteClusterStateService.CUSTOM_METADATA_FORMAT.blobName(fileName))).thenAnswer( + when(blobContainer.readBlob(RemoteGlobalMetadataManager.CUSTOM_METADATA_FORMAT.blobName(fileName))).thenAnswer( (invocation) -> { - BytesReference bytesReference = RemoteClusterStateService.CUSTOM_METADATA_FORMAT.serialize( + BytesReference bytesReference = RemoteGlobalMetadataManager.CUSTOM_METADATA_FORMAT.serialize( metadata.custom(custom), fileName, blobStoreRepository.getCompressor(), @@ -1779,9 +1665,9 @@ private void mockBlobContainerForGlobalMetadata( } } else if (codecVersion == ClusterMetadataManifest.CODEC_V1) { String[] splitPath = clusterMetadataManifest.getGlobalMetadataFileName().split("/"); - when(blobContainer.readBlob(RemoteClusterStateService.GLOBAL_METADATA_FORMAT.blobName(splitPath[splitPath.length - 1]))) + when(blobContainer.readBlob(RemoteGlobalMetadataManager.GLOBAL_METADATA_FORMAT.blobName(splitPath[splitPath.length - 1]))) .thenAnswer((invocationOnMock) -> { - BytesReference bytesGlobalMetadata = RemoteClusterStateService.GLOBAL_METADATA_FORMAT.serialize( + BytesReference bytesGlobalMetadata = RemoteGlobalMetadataManager.GLOBAL_METADATA_FORMAT.serialize( metadata, "global-metadata-file", blobStoreRepository.getCompressor(), @@ -1797,7 +1683,7 @@ private String getFileNameFromPath(String filePath) { return splitPath[splitPath.length - 1]; } - private static ClusterState.Builder generateClusterStateWithGlobalMetadata() { + static ClusterState.Builder generateClusterStateWithGlobalMetadata() { final Settings clusterSettings = Settings.builder().put("cluster.blocks.read_only", true).build(); final CoordinationMetadata coordinationMetadata = CoordinationMetadata.builder().term(1L).build(); @@ -1813,7 +1699,7 @@ private static ClusterState.Builder generateClusterStateWithGlobalMetadata() { ); } - private static ClusterState.Builder generateClusterStateWithOneIndex() { + static ClusterState.Builder generateClusterStateWithOneIndex() { final Index index = new Index("test-index", "index-uuid"); final Settings idxSettings = Settings.builder() .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT) @@ -1843,7 +1729,7 @@ private static ClusterState.Builder generateClusterStateWithOneIndex() { ); } - private static DiscoveryNodes nodesWithLocalNodeClusterManager() { + static DiscoveryNodes nodesWithLocalNodeClusterManager() { return DiscoveryNodes.builder().clusterManagerNodeId("cluster-manager-id").localNodeId("cluster-manager-id").build(); } diff --git a/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java b/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java new file mode 100644 index 0000000000000..c9fb7d13e678a --- /dev/null +++ b/server/src/test/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManagerTests.java @@ -0,0 +1,52 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.gateway.remote; + +import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.common.settings.Settings; +import org.opensearch.repositories.blobstore.BlobStoreRepository; +import org.opensearch.test.OpenSearchTestCase; +import org.junit.After; +import org.junit.Before; + +import static org.mockito.Mockito.mock; + +public class RemoteGlobalMetadataManagerTests extends OpenSearchTestCase { + private RemoteGlobalMetadataManager remoteGlobalMetadataManager; + private ClusterSettings clusterSettings; + private BlobStoreRepository blobStoreRepository; + + @Before + public void setup() { + clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + blobStoreRepository = mock(BlobStoreRepository.class); + remoteGlobalMetadataManager = new RemoteGlobalMetadataManager(blobStoreRepository, clusterSettings); + } + + @After + public void tearDown() throws Exception { + super.tearDown(); + } + + public void testGlobalMetadataUploadWaitTimeSetting() { + // verify default value + assertEquals( + RemoteGlobalMetadataManager.GLOBAL_METADATA_UPLOAD_TIMEOUT_DEFAULT, + remoteGlobalMetadataManager.getGlobalMetadataUploadTimeout() + ); + + // verify update global metadata upload timeout + int globalMetadataUploadTimeout = randomIntBetween(1, 10); + Settings newSettings = Settings.builder() + .put("cluster.remote_store.state.global_metadata.upload_timeout", globalMetadataUploadTimeout + "s") + .build(); + clusterSettings.applySettings(newSettings); + assertEquals(globalMetadataUploadTimeout, remoteGlobalMetadataManager.getGlobalMetadataUploadTimeout().seconds()); + } +} diff --git a/server/src/test/java/org/opensearch/gateway/remote/RemoteIndexMetadataManagerTests.java b/server/src/test/java/org/opensearch/gateway/remote/RemoteIndexMetadataManagerTests.java new file mode 100644 index 0000000000000..8ccc34e32f4b7 --- /dev/null +++ b/server/src/test/java/org/opensearch/gateway/remote/RemoteIndexMetadataManagerTests.java @@ -0,0 +1,52 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.gateway.remote; + +import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.common.settings.Settings; +import org.opensearch.repositories.blobstore.BlobStoreRepository; +import org.opensearch.test.OpenSearchTestCase; +import org.junit.After; +import org.junit.Before; + +import static org.mockito.Mockito.mock; + +public class RemoteIndexMetadataManagerTests extends OpenSearchTestCase { + private RemoteIndexMetadataManager remoteIndexMetadataManager; + private BlobStoreRepository blobStoreRepository; + private ClusterSettings clusterSettings; + + @Before + public void setup() { + clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + blobStoreRepository = mock(BlobStoreRepository.class); + remoteIndexMetadataManager = new RemoteIndexMetadataManager(blobStoreRepository, clusterSettings); + } + + @After + public void tearDown() throws Exception { + super.tearDown(); + } + + public void testIndexMetadataUploadWaitTimeSetting() { + // verify default value + assertEquals( + RemoteIndexMetadataManager.INDEX_METADATA_UPLOAD_TIMEOUT_DEFAULT, + remoteIndexMetadataManager.getIndexMetadataUploadTimeout() + ); + + // verify update index metadata upload timeout + int indexMetadataUploadTimeout = randomIntBetween(1, 10); + Settings newSettings = Settings.builder() + .put("cluster.remote_store.state.index_metadata.upload_timeout", indexMetadataUploadTimeout + "s") + .build(); + clusterSettings.applySettings(newSettings); + assertEquals(indexMetadataUploadTimeout, remoteIndexMetadataManager.getIndexMetadataUploadTimeout().seconds()); + } +} diff --git a/server/src/test/java/org/opensearch/gateway/remote/RemoteManifestManagerTests.java b/server/src/test/java/org/opensearch/gateway/remote/RemoteManifestManagerTests.java new file mode 100644 index 0000000000000..fb7f0612d824c --- /dev/null +++ b/server/src/test/java/org/opensearch/gateway/remote/RemoteManifestManagerTests.java @@ -0,0 +1,143 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +package org.opensearch.gateway.remote; + +import org.opensearch.Version; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.metadata.IndexMetadata; +import org.opensearch.common.blobstore.BlobContainer; +import org.opensearch.common.blobstore.BlobPath; +import org.opensearch.common.blobstore.BlobStore; +import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.common.settings.Settings; +import org.opensearch.core.index.Index; +import org.opensearch.index.remote.RemoteStoreUtils; +import org.opensearch.repositories.blobstore.BlobStoreRepository; +import org.opensearch.test.OpenSearchTestCase; +import org.junit.After; +import org.junit.Before; + +import java.io.IOException; + +import static org.opensearch.gateway.remote.RemoteClusterStateService.INDEX_METADATA_CURRENT_CODEC_VERSION; +import static org.opensearch.gateway.remote.RemoteClusterStateServiceTests.generateClusterStateWithOneIndex; +import static org.opensearch.gateway.remote.RemoteClusterStateServiceTests.nodesWithLocalNodeClusterManager; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.DELIMITER; +import static org.opensearch.gateway.remote.RemoteClusterStateUtils.METADATA_FILE_PREFIX; +import static org.opensearch.gateway.remote.RemoteManifestManager.MANIFEST_CURRENT_CODEC_VERSION; +import static org.opensearch.gateway.remote.RemoteManifestManager.MANIFEST_FILE_PREFIX; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class RemoteManifestManagerTests extends OpenSearchTestCase { + private RemoteManifestManager remoteManifestManager; + private ClusterSettings clusterSettings; + private BlobStoreRepository blobStoreRepository; + private BlobStore blobStore; + + @Before + public void setup() { + clusterSettings = new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS); + blobStoreRepository = mock(BlobStoreRepository.class); + remoteManifestManager = new RemoteManifestManager(blobStoreRepository, clusterSettings, "test-node-id"); + blobStore = mock(BlobStore.class); + when(blobStoreRepository.blobStore()).thenReturn(blobStore); + } + + @After + public void tearDown() throws Exception { + super.tearDown(); + } + + public void testMetadataManifestUploadWaitTimeSetting() { + // verify default value + assertEquals( + RemoteManifestManager.METADATA_MANIFEST_UPLOAD_TIMEOUT_DEFAULT, + remoteManifestManager.getMetadataManifestUploadTimeout() + ); + + // verify update metadata manifest upload timeout + int metadataManifestUploadTimeout = randomIntBetween(1, 10); + Settings newSettings = Settings.builder() + .put("cluster.remote_store.state.metadata_manifest.upload_timeout", metadataManifestUploadTimeout + "s") + .build(); + clusterSettings.applySettings(newSettings); + assertEquals(metadataManifestUploadTimeout, remoteManifestManager.getMetadataManifestUploadTimeout().seconds()); + } + + public void testFileNames() { + final Index index = new Index("test-index", "index-uuid"); + final Settings idxSettings = Settings.builder() + .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT) + .put(IndexMetadata.SETTING_INDEX_UUID, index.getUUID()) + .build(); + final IndexMetadata indexMetadata = new IndexMetadata.Builder(index.getName()).settings(idxSettings) + .numberOfShards(1) + .numberOfReplicas(0) + .build(); + + String indexMetadataFileName = RemoteIndexMetadataManager.indexMetadataFileName(indexMetadata); + String[] splittedIndexMetadataFileName = indexMetadataFileName.split(DELIMITER); + assertEquals(4, indexMetadataFileName.split(DELIMITER).length); + assertEquals(METADATA_FILE_PREFIX, splittedIndexMetadataFileName[0]); + assertEquals(RemoteStoreUtils.invertLong(indexMetadata.getVersion()), splittedIndexMetadataFileName[1]); + assertEquals(String.valueOf(INDEX_METADATA_CURRENT_CODEC_VERSION), splittedIndexMetadataFileName[3]); + + verifyManifestFileNameWithCodec(MANIFEST_CURRENT_CODEC_VERSION); + verifyManifestFileNameWithCodec(ClusterMetadataManifest.CODEC_V1); + verifyManifestFileNameWithCodec(ClusterMetadataManifest.CODEC_V0); + } + + private void verifyManifestFileNameWithCodec(int codecVersion) { + int term = randomIntBetween(5, 10); + int version = randomIntBetween(5, 10); + String manifestFileName = RemoteManifestManager.getManifestFileName(term, version, true, codecVersion); + assertEquals(6, manifestFileName.split(DELIMITER).length); + String[] splittedName = manifestFileName.split(DELIMITER); + assertEquals(MANIFEST_FILE_PREFIX, splittedName[0]); + assertEquals(RemoteStoreUtils.invertLong(term), splittedName[1]); + assertEquals(RemoteStoreUtils.invertLong(version), splittedName[2]); + assertEquals("C", splittedName[3]); + assertEquals(String.valueOf(codecVersion), splittedName[5]); + + manifestFileName = RemoteManifestManager.getManifestFileName(term, version, false, codecVersion); + splittedName = manifestFileName.split(DELIMITER); + assertEquals("P", splittedName[3]); + } + + public void testReadLatestMetadataManifestFailedIOException() throws IOException { + final ClusterState clusterState = generateClusterStateWithOneIndex().nodes(nodesWithLocalNodeClusterManager()).build(); + + BlobContainer blobContainer = mockBlobStoreObjects(); + when(blobContainer.listBlobsByPrefixInSortedOrder("manifest" + DELIMITER, 1, BlobContainer.BlobNameSortOrder.LEXICOGRAPHIC)) + .thenThrow(IOException.class); + + Exception e = assertThrows( + IllegalStateException.class, + () -> remoteManifestManager.getLatestClusterMetadataManifest( + clusterState.getClusterName().value(), + clusterState.metadata().clusterUUID() + ) + ); + assertEquals(e.getMessage(), "Error while fetching latest manifest file for remote cluster state"); + } + + private BlobContainer mockBlobStoreObjects() { + final BlobPath blobPath = mock(BlobPath.class); + when((blobStoreRepository.basePath())).thenReturn(blobPath); + when(blobPath.add(anyString())).thenReturn(blobPath); + when(blobPath.buildAsString()).thenReturn("/blob/path/"); + final BlobContainer blobContainer = mock(BlobContainer.class); + when(blobContainer.path()).thenReturn(blobPath); + when(blobStore.blobContainer(any())).thenReturn(blobContainer); + return blobContainer; + } +}