-
Notifications
You must be signed in to change notification settings - Fork 1.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add Remote store as a segment replication source
Signed-off-by: Ankit Kala <[email protected]>
- Loading branch information
Showing
18 changed files
with
286 additions
and
55 deletions.
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
66 changes: 66 additions & 0 deletions
66
...rnalClusterTest/java/org/opensearch/remotestore/SegmentReplicationUsingRemoteStoreIT.java
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,66 @@ | ||
| /* | ||
| * 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; | ||
|
|
||
| /** | ||
| * The aim of this class is to run Segment Replication integ tests by enabling remote store specific settings. | ||
| * This makes sure that the constructs/flows that are being tested with Segment Replication, holds true after enabling | ||
| * remote store. | ||
| */ | ||
| @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0) | ||
| public class SegmentReplicationUsingRemoteStoreIT 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") | ||
| .put(FeatureFlags.SEGMENT_REPLICATION_EXPERIMENTAL, "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
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
101 changes: 101 additions & 0 deletions
101
server/src/main/java/org/opensearch/indices/replication/RemoteStoreReplicationSource.java
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,101 @@ | ||
| /* | ||
| * 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.indices.replication; | ||
|
|
||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.apache.lucene.store.FilterDirectory; | ||
| import org.apache.lucene.util.Version; | ||
| import org.opensearch.action.ActionListener; | ||
| import org.opensearch.index.shard.IndexShard; | ||
| import org.opensearch.index.store.RemoteSegmentStoreDirectory; | ||
| import org.opensearch.index.store.Store; | ||
| import org.opensearch.index.store.StoreFileMetadata; | ||
| import org.opensearch.indices.replication.checkpoint.ReplicationCheckpoint; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Collections; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| /** | ||
| * Implementation of a {@link SegmentReplicationSource} where the source is remote store. | ||
| * | ||
| * @opensearch.internal | ||
| */ | ||
| public class RemoteStoreReplicationSource implements SegmentReplicationSource { | ||
|
|
||
| private static final Logger logger = LogManager.getLogger(PrimaryShardReplicationSource.class); | ||
|
|
||
| private final IndexShard indexShard; | ||
|
|
||
| public RemoteStoreReplicationSource(IndexShard indexShard) { | ||
| this.indexShard = indexShard; | ||
| } | ||
|
|
||
| @Override | ||
| public void getCheckpointMetadata( | ||
| long replicationId, | ||
| ReplicationCheckpoint checkpoint, | ||
| ActionListener<CheckpointInfoResponse> listener | ||
| ) { | ||
| FilterDirectory remoteStoreDirectory = (FilterDirectory) indexShard.remoteStore().directory(); | ||
| FilterDirectory byteSizeCachingStoreDirectory = (FilterDirectory) remoteStoreDirectory.getDelegate(); | ||
| RemoteSegmentStoreDirectory remoteDirectory = (RemoteSegmentStoreDirectory) byteSizeCachingStoreDirectory.getDelegate(); | ||
|
|
||
| Map<String, StoreFileMetadata> metadataMap; | ||
| // TODO: Need to figure out a way to pass this information for segment metadata via remote store. | ||
| final Version version = indexShard.getSegmentInfosSnapshot().get().getCommitLuceneVersion(); | ||
| try { | ||
| metadataMap = remoteDirectory.readLatestMetadataFile() | ||
| .entrySet() | ||
| .stream() | ||
| .collect( | ||
| Collectors.toMap( | ||
| e -> e.getKey(), | ||
| e -> new StoreFileMetadata( | ||
| e.getValue().getOriginalFilename(), | ||
| e.getValue().getLength(), | ||
| Store.digestToString(Long.valueOf(e.getValue().getChecksum())), | ||
| version, | ||
| null | ||
| ) | ||
| ) | ||
| ); | ||
| // TODO: GET current checkpoint from remote store. | ||
| listener.onResponse(new CheckpointInfoResponse(checkpoint, metadataMap, null)); | ||
| } catch (IOException e) { | ||
| logger.error("Error fetching checkpoint metadata from remote store", e); | ||
| listener.onFailure(e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void getSegmentFiles( | ||
| long replicationId, | ||
| ReplicationCheckpoint checkpoint, | ||
| List<StoreFileMetadata> filesToFetch, | ||
| IndexShard indexShard, | ||
| ActionListener<GetSegmentFilesResponse> listener | ||
| ) { | ||
| try { | ||
| indexShard.syncSegmentsFromRemoteSegmentStore(false, true, false); | ||
| listener.onResponse(new GetSegmentFilesResponse(Collections.emptyList())); | ||
| } catch (Exception e) { | ||
| logger.error("Failed to sync segments", e); | ||
| listener.onFailure(e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public String getDescription() { | ||
| return "remote store"; | ||
| } | ||
| } |
Oops, something went wrong.