Refactor translog download flow and Add support to run SegRep integ tests using remote store settings #6405
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| /* | ||
| * 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.remotestore; | ||
|
|
||
| import org.junit.After; | ||
| import org.junit.Before; | ||
| import org.opensearch.cluster.metadata.IndexMetadata; | ||
| import org.opensearch.common.settings.Settings; | ||
| import org.opensearch.common.util.FeatureFlags; | ||
| import org.opensearch.indices.replication.SegmentReplicationIT; | ||
| import org.opensearch.test.OpenSearchIntegTestCase; | ||
|
|
||
| import java.nio.file.Path; | ||
|
|
||
| import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked; | ||
|
|
||
| @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0) | ||
| public class SegmentReplicationRemoteStoreIT extends SegmentReplicationIT { | ||
|
|
||
| private static final String REPOSITORY_NAME = "test-remore-store-repo"; | ||
|
|
||
| @Override | ||
| public Settings indexSettings() { | ||
| return Settings.builder() | ||
| .put(super.indexSettings()) | ||
| .put(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, true) | ||
| .put(IndexMetadata.SETTING_REMOTE_STORE_REPOSITORY, REPOSITORY_NAME) | ||
| .put(IndexMetadata.SETTING_REMOTE_TRANSLOG_STORE_ENABLED, true) | ||
| .put(IndexMetadata.SETTING_REMOTE_TRANSLOG_STORE_REPOSITORY, REPOSITORY_NAME) | ||
| .build(); | ||
| } | ||
|
|
||
| @Override | ||
| protected Settings featureFlagSettings() { | ||
| return Settings.builder().put(super.featureFlagSettings()).put(FeatureFlags.REMOTE_STORE, "true").build(); | ||
| } | ||
|
|
||
| @Before | ||
| public void setup() { | ||
| internalCluster().startClusterManagerOnlyNode(); | ||
| Path absolutePath = randomRepoPath().toAbsolutePath(); | ||
| assertAcked( | ||
| clusterAdmin().preparePutRepository(REPOSITORY_NAME).setType("fs").setSettings(Settings.builder().put("location", absolutePath)) | ||
| ); | ||
| } | ||
|
|
||
| @After | ||
| public void teardown() { | ||
| assertAcked(clusterAdmin().prepareDeleteRepository(REPOSITORY_NAME)); | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -22,12 +22,14 @@ | |
| import org.opensearch.index.translog.transfer.TranslogTransferManager; | ||
| import org.opensearch.index.translog.transfer.TranslogTransferMetadata; | ||
| import org.opensearch.index.translog.transfer.listener.TranslogTransferListener; | ||
| import org.opensearch.repositories.Repository; | ||
| import org.opensearch.repositories.blobstore.BlobStoreRepository; | ||
| import org.opensearch.threadpool.ThreadPool; | ||
|
|
||
| import java.io.IOException; | ||
| import java.nio.file.Files; | ||
| import java.nio.file.Path; | ||
| import java.util.Arrays; | ||
| import java.util.HashSet; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
|
|
@@ -46,7 +48,6 @@ | |
| */ | ||
| public class RemoteFsTranslog extends Translog { | ||
|
|
||
| private final TranslogTransferManager translogTransferManager; | ||
| private final FileTransferTracker fileTransferTracker; | ||
| private final BooleanSupplier primaryModeSupplier; | ||
|
|
@@ -77,7 +78,6 @@ public RemoteFsTranslog( | |
| BooleanSupplier primaryModeSupplier | ||
| ) throws IOException { | ||
| super(config, translogUUID, deletionPolicy, globalCheckpointSupplier, primaryTermSupplier, persistedSequenceNumberConsumer); | ||
| this.primaryModeSupplier = primaryModeSupplier; | ||
| fileTransferTracker = new FileTransferTracker(shardId); | ||
| this.translogTransferManager = buildTranslogTransferManager(blobStoreRepository, threadPool, shardId, fileTransferTracker); | ||
|
|
@@ -116,22 +116,35 @@ public RemoteFsTranslog( | |
| } | ||
| } | ||
|
|
||
| public static void download(Repository repository, ShardId shardId, ThreadPool threadPool, Path location) throws IOException { | ||
| assert repository instanceof BlobStoreRepository : "repository should be instance of BlobStoreRepository"; | ||
| BlobStoreRepository blobStoreRepository = (BlobStoreRepository) repository; | ||
| FileTransferTracker fileTransferTracker = new FileTransferTracker(shardId); | ||
| TranslogTransferManager translogTransferManager = buildTranslogTransferManager( | ||
| blobStoreRepository, | ||
| threadPool, | ||
| shardId, | ||
| fileTransferTracker | ||
| ); | ||
| RemoteFsTranslog.download(translogTransferManager, location); | ||
| } | ||
|
|
||
| private static void download(TranslogTransferManager translogTransferManager, Path location) throws IOException { | ||
| TranslogTransferMetadata translogMetadata = translogTransferManager.readMetadata(); | ||
| if (translogMetadata != null) { | ||
| if (Files.notExists(location)) { | ||
| Files.createDirectories(location); | ||
| } else { | ||
| deleteTranslogFilesNotUploaded(location, translogMetadata.getGeneration()); | ||
| } | ||
| Map<String, String> generationToPrimaryTermMapper = translogMetadata.getGenerationToPrimaryTermMapper(); | ||
| for (long i = translogMetadata.getGeneration(); i >= translogMetadata.getMinTranslogGeneration(); i--) { | ||
| String generation = Long.toString(i); | ||
| translogTransferManager.downloadTranslog(generationToPrimaryTermMapper.get(generation), generation, location); | ||
| } | ||
| if (Files.exists(location.resolve(Translog.CHECKPOINT_FILE_NAME))) { | ||
| Files.delete(location.resolve(Translog.CHECKPOINT_FILE_NAME)); | ||
| } | ||
| // We copy the latest generation .ckp file to translog.ckp so that flows that depend on | ||
| // existence of translog.ckp file work in the same way | ||
| Files.copy( | ||
|
|
@@ -141,6 +154,25 @@ public static void download(TranslogTransferManager translogTransferManager, Pat | |
| } | ||
| } | ||
|
|
||
| private static void deleteTranslogFilesNotUploaded(Path location, long uploadedGeneration) throws IOException { | ||
|
There was a problem hiding this comment. If I recall the earlier check on deleting files before downloading translog files was added as otherwise the download used to fail, right? Now we are deleting files that are greater than the max generation referenced by the remote translog metadata file. While this is really a good optimisation, I think we should check the local file's checksum and the expected checksum. If there is a story on the checksum part already, then do let me know. If not, let's follow it up in next PR. There was a problem hiding this comment. On high level, comparing checksum makes sense. This is what we do for segments and can be added here as well. There was a problem hiding this comment. Created tracking issue: #6896 |
||
| // Delete translog files with generation > translogMetadata.getGeneration() | ||
| Arrays.stream(FileSystemUtils.files(location)) | ||
| .map(filePath -> filePath.getFileName().toString()) | ||
| .filter(filename -> filename.endsWith(TRANSLOG_FILE_SUFFIX)) | ||
| .map( | ||
| filename -> Long.parseLong( | ||
| filename.substring(TRANSLOG_FILE_PREFIX.length(), filename.length() - TRANSLOG_FILE_SUFFIX.length()) | ||
| ) | ||
| ) | ||
| .filter(generation -> generation > uploadedGeneration) | ||
| .forEach( | ||
| generation -> IOUtils.deleteFilesIgnoringExceptions( | ||
| location.resolve(Translog.getCommitCheckpointFileName(generation)), | ||
| location.resolve(Translog.getFilename(generation)) | ||
| ) | ||
| ); | ||
| } | ||
|
|
||
| public static TranslogTransferManager buildTranslogTransferManager( | ||
| BlobStoreRepository blobStoreRepository, | ||
| ThreadPool threadPool, | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Just a question - what
numDataNodes = 0means here? I checked the code base and saw multiple such occurrences.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We want to control number of nodes in the test cluster. That is why we provide 0 and start nodes as per the test requirement.
More on
numDataNodes: https://github.com/opensearch-project/OpenSearch/blob/main/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java#L1722