Add prefix support to hashed prefix & infix path types on remote store (

opensearch-project#15557) * Add prefix support to hashed prefix & infix path types on remote store Signed-off-by: Ashish Singh <[email protected]>
dk2k · Oct 17, 2024 · 5ee411b · 5ee411b
1 parent cde0822
commit 5ee411b
Show file tree

Hide file tree

Showing 37 changed files with 617 additions and 101 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -45,6 +45,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add support to upload snapshot shard blobs with hashed prefix ([#15426](https://github.com/opensearch-project/OpenSearch/pull/15426))
 - [Remote Publication] Add remote download stats ([#15291](https://github.com/opensearch-project/OpenSearch/pull/15291)))
 - Add support for comma-separated list of index names to be used with Snapshot Status API ([#15409](https://github.com/opensearch-project/OpenSearch/pull/15409))
+- Add prefix support to hashed prefix & infix path types on remote store ([#15557](https://github.com/opensearch-project/OpenSearch/pull/15557))
 
 ### Dependencies
 - Bump `netty` from 4.1.111.Final to 4.1.112.Final ([#15081](https://github.com/opensearch-project/OpenSearch/pull/15081))

diff --git a/...internalClusterTest/java/org/opensearch/remotemigration/RemotePrimaryLocalRecoveryIT.java b/...internalClusterTest/java/org/opensearch/remotemigration/RemotePrimaryLocalRecoveryIT.java
@@ -19,6 +19,7 @@
 import org.opensearch.core.util.FileSystemUtils;
 import org.opensearch.index.remote.RemoteSegmentStats;
 import org.opensearch.index.translog.RemoteTranslogStats;
+import org.opensearch.indices.RemoteStoreSettings;
 import org.opensearch.test.InternalTestCluster;
 import org.opensearch.test.OpenSearchIntegTestCase;
 
@@ -67,14 +68,16 @@ public void testLocalRecoveryRollingRestartAndNodeFailure() throws Exception {
                 assertTrue(remoteSegmentStats.getUploadBytesSucceeded() > 0);
             }
 
+            String segmentsPathFixedPrefix = RemoteStoreSettings.CLUSTER_REMOTE_STORE_SEGMENTS_PATH_PREFIX.get(getNodeSettings());
             assertBusy(() -> {
                 String shardPath = getShardLevelBlobPath(
                     client(),
                     indexName,
                     new BlobPath(),
                     String.valueOf(shardRouting.getId()),
                     SEGMENTS,
-                    DATA
+                    DATA,
+                    segmentsPathFixedPrefix
                 ).buildAsString();
                 Path segmentDataRepoPath = segmentRepoPath.resolve(shardPath);
                 List<String> segmentsNFilesInRepo = Arrays.stream(FileSystemUtils.files(segmentDataRepoPath))

diff --git a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteRestoreSnapshotIT.java b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteRestoreSnapshotIT.java
@@ -443,13 +443,15 @@ public void testRestoreInSameRemoteStoreEnabledIndex() throws IOException {
 
     void assertRemoteSegmentsAndTranslogUploaded(String idx) throws IOException {
         Client client = client();
-        String path = getShardLevelBlobPath(client, idx, new BlobPath(), "0", TRANSLOG, METADATA).buildAsString();
+        String translogPathFixedPrefix = RemoteStoreSettings.CLUSTER_REMOTE_STORE_TRANSLOG_PATH_PREFIX.get(getNodeSettings());
+        String segmentsPathFixedPrefix = RemoteStoreSettings.CLUSTER_REMOTE_STORE_SEGMENTS_PATH_PREFIX.get(getNodeSettings());
+        String path = getShardLevelBlobPath(client, idx, new BlobPath(), "0", TRANSLOG, METADATA, translogPathFixedPrefix).buildAsString();
         Path remoteTranslogMetadataPath = Path.of(remoteRepoPath + "/" + path);
-        path = getShardLevelBlobPath(client, idx, new BlobPath(), "0", TRANSLOG, DATA).buildAsString();
+        path = getShardLevelBlobPath(client, idx, new BlobPath(), "0", TRANSLOG, DATA, translogPathFixedPrefix).buildAsString();
         Path remoteTranslogDataPath = Path.of(remoteRepoPath + "/" + path);
-        path = getShardLevelBlobPath(client, idx, new BlobPath(), "0", SEGMENTS, METADATA).buildAsString();
+        path = getShardLevelBlobPath(client, idx, new BlobPath(), "0", SEGMENTS, METADATA, segmentsPathFixedPrefix).buildAsString();
         Path segmentMetadataPath = Path.of(remoteRepoPath + "/" + path);
-        path = getShardLevelBlobPath(client, idx, new BlobPath(), "0", SEGMENTS, DATA).buildAsString();
+        path = getShardLevelBlobPath(client, idx, new BlobPath(), "0", SEGMENTS, DATA, segmentsPathFixedPrefix).buildAsString();
         Path segmentDataPath = Path.of(remoteRepoPath + "/" + path);
 
         try (

diff --git a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java
@@ -205,7 +205,16 @@ public void testStaleCommitDeletionWithInvokeFlush() throws Exception {
         createIndex(INDEX_NAME, remoteStoreIndexSettings(1, 10000l, -1));
         int numberOfIterations = randomIntBetween(5, 15);
         indexData(numberOfIterations, true, INDEX_NAME);
-        String shardPath = getShardLevelBlobPath(client(), INDEX_NAME, BlobPath.cleanPath(), "0", SEGMENTS, METADATA).buildAsString();
+        String segmentsPathFixedPrefix = RemoteStoreSettings.CLUSTER_REMOTE_STORE_SEGMENTS_PATH_PREFIX.get(getNodeSettings());
+        String shardPath = getShardLevelBlobPath(
+            client(),
+            INDEX_NAME,
+            BlobPath.cleanPath(),
+            "0",
+            SEGMENTS,
+            METADATA,
+            segmentsPathFixedPrefix
+        ).buildAsString();
         Path indexPath = Path.of(segmentRepoPath + "/" + shardPath);
         ;
         IndexShard indexShard = getIndexShard(dataNode, INDEX_NAME);
@@ -236,7 +245,16 @@ public void testStaleCommitDeletionWithoutInvokeFlush() throws Exception {
         createIndex(INDEX_NAME, remoteStoreIndexSettings(1, 10000l, -1));
         int numberOfIterations = randomIntBetween(5, 15);
         indexData(numberOfIterations, false, INDEX_NAME);
-        String shardPath = getShardLevelBlobPath(client(), INDEX_NAME, BlobPath.cleanPath(), "0", SEGMENTS, METADATA).buildAsString();
+        String segmentsPathFixedPrefix = RemoteStoreSettings.CLUSTER_REMOTE_STORE_SEGMENTS_PATH_PREFIX.get(getNodeSettings());
+        String shardPath = getShardLevelBlobPath(
+            client(),
+            INDEX_NAME,
+            BlobPath.cleanPath(),
+            "0",
+            SEGMENTS,
+            METADATA,
+            segmentsPathFixedPrefix
+        ).buildAsString();
         Path indexPath = Path.of(segmentRepoPath + "/" + shardPath);
         int actualFileCount = getFileCount(indexPath);
         // We also allow (numberOfIterations + 1) as index creation also triggers refresh.
@@ -247,11 +265,19 @@ public void testStaleCommitDeletionWithMinSegmentFiles_3() throws Exception {
         Settings.Builder settings = Settings.builder()
             .put(RemoteStoreSettings.CLUSTER_REMOTE_INDEX_SEGMENT_METADATA_RETENTION_MAX_COUNT_SETTING.getKey(), "3");
         internalCluster().startNode(settings);
-
+        String segmentsPathFixedPrefix = RemoteStoreSettings.CLUSTER_REMOTE_STORE_SEGMENTS_PATH_PREFIX.get(getNodeSettings());
         createIndex(INDEX_NAME, remoteStoreIndexSettings(1, 10000l, -1));
         int numberOfIterations = randomIntBetween(5, 15);
         indexData(numberOfIterations, true, INDEX_NAME);
-        String shardPath = getShardLevelBlobPath(client(), INDEX_NAME, BlobPath.cleanPath(), "0", SEGMENTS, METADATA).buildAsString();
+        String shardPath = getShardLevelBlobPath(
+            client(),
+            INDEX_NAME,
+            BlobPath.cleanPath(),
+            "0",
+            SEGMENTS,
+            METADATA,
+            segmentsPathFixedPrefix
+        ).buildAsString();
         Path indexPath = Path.of(segmentRepoPath + "/" + shardPath);
         int actualFileCount = getFileCount(indexPath);
         // We also allow (numberOfIterations + 1) as index creation also triggers refresh.
@@ -271,7 +297,16 @@ public void testStaleCommitDeletionWithMinSegmentFiles_Disabled() throws Excepti
         createIndex(INDEX_NAME, remoteStoreIndexSettings(1, 10000l, -1));
         int numberOfIterations = randomIntBetween(12, 18);
         indexData(numberOfIterations, true, INDEX_NAME);
-        String shardPath = getShardLevelBlobPath(client(), INDEX_NAME, BlobPath.cleanPath(), "0", SEGMENTS, METADATA).buildAsString();
+        String segmentsPathFixedPrefix = RemoteStoreSettings.CLUSTER_REMOTE_STORE_SEGMENTS_PATH_PREFIX.get(getNodeSettings());
+        String shardPath = getShardLevelBlobPath(
+            client(),
+            INDEX_NAME,
+            BlobPath.cleanPath(),
+            "0",
+            SEGMENTS,
+            METADATA,
+            segmentsPathFixedPrefix
+        ).buildAsString();
         Path indexPath = Path.of(segmentRepoPath + "/" + shardPath);
         ;
         int actualFileCount = getFileCount(indexPath);
@@ -604,8 +639,10 @@ public void testFallbackToNodeToNodeSegmentCopy() throws Exception {
         indexBulk(INDEX_NAME, 50);
         flushAndRefresh(INDEX_NAME);
 
+        String segmentsPathFixedPrefix = RemoteStoreSettings.CLUSTER_REMOTE_STORE_SEGMENTS_PATH_PREFIX.get(getNodeSettings());
         // 3. Delete data from remote segment store
-        String shardPath = getShardLevelBlobPath(client(), INDEX_NAME, BlobPath.cleanPath(), "0", SEGMENTS, DATA).buildAsString();
+        String shardPath = getShardLevelBlobPath(client(), INDEX_NAME, BlobPath.cleanPath(), "0", SEGMENTS, DATA, segmentsPathFixedPrefix)
+            .buildAsString();
         Path segmentDataPath = Path.of(segmentRepoPath + "/" + shardPath);
 
         try (Stream<Path> files = Files.list(segmentDataPath)) {
@@ -844,7 +881,16 @@ public void testLocalOnlyTranslogCleanupOnNodeRestart() throws Exception {
             .get()
             .getSetting(INDEX_NAME, IndexMetadata.SETTING_INDEX_UUID);
 
-        String shardPath = getShardLevelBlobPath(client(), INDEX_NAME, BlobPath.cleanPath(), "0", TRANSLOG, METADATA).buildAsString();
+        String translogPathFixedPrefix = RemoteStoreSettings.CLUSTER_REMOTE_STORE_TRANSLOG_PATH_PREFIX.get(getNodeSettings());
+        String shardPath = getShardLevelBlobPath(
+            client(),
+            INDEX_NAME,
+            BlobPath.cleanPath(),
+            "0",
+            TRANSLOG,
+            METADATA,
+            translogPathFixedPrefix
+        ).buildAsString();
         Path translogMetaDataPath = Path.of(translogRepoPath + "/" + shardPath);
 
         try (Stream<Path> files = Files.list(translogMetaDataPath)) {

diff --git a/...src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreRefreshListenerIT.java b/...src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreRefreshListenerIT.java
@@ -13,6 +13,7 @@
 import org.opensearch.action.admin.indices.stats.IndicesStatsResponse;
 import org.opensearch.common.blobstore.BlobPath;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.indices.RemoteStoreSettings;
 import org.opensearch.test.OpenSearchIntegTestCase;
 
 import java.nio.file.Path;
@@ -50,7 +51,10 @@ public void testRemoteRefreshRetryOnFailure() throws Exception {
 
         String indexName = response.getShards()[0].getShardRouting().index().getName();
         String indexUuid = response.getShards()[0].getShardRouting().index().getUUID();
-        String shardPath = getShardLevelBlobPath(client(), indexName, new BlobPath(), "0", SEGMENTS, DATA).buildAsString();
+
+        String segmentsPathFixedPrefix = RemoteStoreSettings.CLUSTER_REMOTE_STORE_SEGMENTS_PATH_PREFIX.get(getNodeSettings());
+        String shardPath = getShardLevelBlobPath(client(), indexName, new BlobPath(), "0", SEGMENTS, DATA, segmentsPathFixedPrefix)
+            .buildAsString();
         Path segmentDataRepoPath = location.resolve(shardPath);
         String segmentDataLocalPath = String.format(Locale.ROOT, "%s/indices/%s/0/index", response.getShards()[0].getDataPath(), indexUuid);
 

diff --git a/server/src/internalClusterTest/java/org/opensearch/snapshots/DeleteSnapshotIT.java b/server/src/internalClusterTest/java/org/opensearch/snapshots/DeleteSnapshotIT.java
@@ -324,13 +324,15 @@ public void testRemoteStoreCleanupForDeletedIndex() throws Exception {
 
         final RepositoriesService repositoriesService = internalCluster().getCurrentClusterManagerNodeInstance(RepositoriesService.class);
         final BlobStoreRepository remoteStoreRepository = (BlobStoreRepository) repositoriesService.repository(REMOTE_REPO_NAME);
+        String segmentsPathFixedPrefix = RemoteStoreSettings.CLUSTER_REMOTE_STORE_SEGMENTS_PATH_PREFIX.get(getNodeSettings());
         BlobPath shardLevelBlobPath = getShardLevelBlobPath(
             client(),
             remoteStoreEnabledIndexName,
             remoteStoreRepository.basePath(),
             "0",
             SEGMENTS,
-            LOCK_FILES
+            LOCK_FILES,
+            segmentsPathFixedPrefix
         );
         BlobContainer blobContainer = remoteStoreRepository.blobStore().blobContainer(shardLevelBlobPath);
         String[] lockFiles;

diff --git a/...pensearch/action/admin/cluster/repositories/cleanup/TransportCleanupRepositoryAction.java b/...pensearch/action/admin/cluster/repositories/cleanup/TransportCleanupRepositoryAction.java
@@ -53,6 +53,7 @@
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.index.store.lockmanager.RemoteStoreLockManagerFactory;
+import org.opensearch.indices.RemoteStoreSettings;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.repositories.Repository;
 import org.opensearch.repositories.RepositoryCleanupResult;
@@ -109,7 +110,8 @@ public TransportCleanupRepositoryAction(
         SnapshotsService snapshotsService,
         ThreadPool threadPool,
         ActionFilters actionFilters,
-        IndexNameExpressionResolver indexNameExpressionResolver
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        RemoteStoreSettings remoteStoreSettings
     ) {
         super(
             CleanupRepositoryAction.NAME,
@@ -122,7 +124,10 @@ public TransportCleanupRepositoryAction(
         );
         this.repositoriesService = repositoriesService;
         this.snapshotsService = snapshotsService;
-        this.remoteStoreLockManagerFactory = new RemoteStoreLockManagerFactory(() -> repositoriesService);
+        this.remoteStoreLockManagerFactory = new RemoteStoreLockManagerFactory(
+            () -> repositoriesService,
+            remoteStoreSettings.getSegmentsPathFixedPrefix()
+        );
         // We add a state applier that will remove any dangling repository cleanup actions on cluster-manager failover.
         // This is safe to do since cleanups will increment the repository state id before executing any operations to prevent concurrent
         // operations from corrupting the repository. This is the same safety mechanism used by snapshot deletes.

diff --git a/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java b/server/src/main/java/org/opensearch/common/settings/ClusterSettings.java
@@ -766,6 +766,8 @@ public void apply(Settings value, Settings current, Settings previous) {
                 RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_SCHEDULER_INTERVAL,
                 RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_LOOKBACK_INTERVAL,
                 RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED,
+                RemoteStoreSettings.CLUSTER_REMOTE_STORE_SEGMENTS_PATH_PREFIX,
+                RemoteStoreSettings.CLUSTER_REMOTE_STORE_TRANSLOG_PATH_PREFIX,
 
                 SearchService.CLUSTER_ALLOW_DERIVED_FIELD_SETTING,
 

diff --git a/server/src/main/java/org/opensearch/index/remote/RemoteIndexPath.java b/server/src/main/java/org/opensearch/index/remote/RemoteIndexPath.java
@@ -20,6 +20,7 @@
 import org.opensearch.index.remote.RemoteStoreEnums.PathType;
 import org.opensearch.index.remote.RemoteStorePathStrategy.PathInput;
 import org.opensearch.index.remote.RemoteStorePathStrategy.ShardDataPathInput;
+import org.opensearch.indices.RemoteStoreSettings;
 
 import java.io.IOException;
 import java.util.Collections;
@@ -68,6 +69,7 @@ public class RemoteIndexPath implements ToXContentFragment {
     private final Iterable<String> basePath;
     private final PathType pathType;
     private final PathHashAlgorithm pathHashAlgorithm;
+    private final RemoteStoreSettings remoteStoreSettings;
 
     /**
      * This keeps the map of paths that would be present in the content of the index path file. For eg - It is possible
@@ -82,7 +84,8 @@ public RemoteIndexPath(
         Iterable<String> basePath,
         PathType pathType,
         PathHashAlgorithm pathHashAlgorithm,
-        Map<DataCategory, List<DataType>> pathCreationMap
+        Map<DataCategory, List<DataType>> pathCreationMap,
+        RemoteStoreSettings remoteStoreSettings
     ) {
         if (Objects.isNull(pathCreationMap)
             || Objects.isNull(pathType)
@@ -119,6 +122,7 @@ public RemoteIndexPath(
         this.pathType = pathType;
         this.pathHashAlgorithm = pathHashAlgorithm;
         this.pathCreationMap = pathCreationMap;
+        this.remoteStoreSettings = remoteStoreSettings;
     }
 
     @Override
@@ -148,6 +152,11 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
                         .shardId(Integer.toString(shardNo))
                         .dataCategory(dataCategory)
                         .dataType(type)
+                        .fixedPrefix(
+                            dataCategory == TRANSLOG
+                                ? remoteStoreSettings.getTranslogPathFixedPrefix()
+                                : remoteStoreSettings.getSegmentsPathFixedPrefix()
+                        )
                         .build();
                     builder.value(pathType.path(pathInput, pathHashAlgorithm).buildAsString());
                 }

diff --git a/server/src/main/java/org/opensearch/index/remote/RemoteIndexPathUploader.java b/server/src/main/java/org/opensearch/index/remote/RemoteIndexPathUploader.java
@@ -25,6 +25,7 @@
 import org.opensearch.gateway.remote.IndexMetadataUploadListener;
 import org.opensearch.gateway.remote.RemoteStateTransferException;
 import org.opensearch.index.remote.RemoteStoreEnums.PathType;
+import org.opensearch.indices.RemoteStoreSettings;
 import org.opensearch.node.Node;
 import org.opensearch.node.remotestore.RemoteStoreNodeAttribute;
 import org.opensearch.repositories.RepositoriesService;
@@ -79,6 +80,7 @@ public class RemoteIndexPathUploader extends IndexMetadataUploadListener {
     private final Settings settings;
     private final boolean isRemoteDataAttributePresent;
     private final boolean isTranslogSegmentRepoSame;
+    private final RemoteStoreSettings remoteStoreSettings;
     private final Supplier<RepositoriesService> repositoriesService;
     private volatile TimeValue metadataUploadTimeout;
 
@@ -89,7 +91,8 @@ public RemoteIndexPathUploader(
         ThreadPool threadPool,
         Settings settings,
         Supplier<RepositoriesService> repositoriesService,
-        ClusterSettings clusterSettings
+        ClusterSettings clusterSettings,
+        RemoteStoreSettings remoteStoreSettings
     ) {
         super(threadPool, ThreadPool.Names.GENERIC);
         this.settings = Objects.requireNonNull(settings);
@@ -100,6 +103,7 @@ public RemoteIndexPathUploader(
         Objects.requireNonNull(clusterSettings);
         metadataUploadTimeout = clusterSettings.get(GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING);
         clusterSettings.addSettingsUpdateConsumer(GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING, this::setMetadataUploadTimeout);
+        this.remoteStoreSettings = remoteStoreSettings;
     }
 
     @Override
@@ -208,7 +212,8 @@ private void writePathToRemoteStore(
                 basePath,
                 pathType,
                 hashAlgorithm,
-                pathCreationMap
+                pathCreationMap,
+                remoteStoreSettings
             );
             String fileName = generateFileName(indexUUID, idxMD.getVersion(), remoteIndexPath.getVersion());
             REMOTE_INDEX_PATH_FORMAT.writeAsyncWithUrgentPriority(remoteIndexPath, blobContainer, fileName, actionListener);

diff --git a/server/src/main/java/org/opensearch/index/remote/RemoteStoreEnums.java b/server/src/main/java/org/opensearch/index/remote/RemoteStoreEnums.java
@@ -13,6 +13,7 @@
 import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.common.blobstore.BlobPath;
 import org.opensearch.common.hash.FNV1a;
+import org.opensearch.core.common.Strings;
 import org.opensearch.index.remote.RemoteStorePathStrategy.PathInput;
 
 import java.util.HashMap;
@@ -107,7 +108,11 @@ boolean requiresHashAlgorithm() {
             @Override
             public BlobPath generatePath(PathInput pathInput, PathHashAlgorithm hashAlgorithm) {
                 assert Objects.nonNull(hashAlgorithm) : "hashAlgorithm is expected to be non-null";
-                return BlobPath.cleanPath().add(hashAlgorithm.hash(pathInput)).add(pathInput.basePath()).add(pathInput.fixedSubPath());
+                String fixedPrefix = pathInput.fixedPrefix();
+                return BlobPath.cleanPath()
+                    .add(Strings.isNullOrEmpty(fixedPrefix) ? hashAlgorithm.hash(pathInput) : fixedPrefix + hashAlgorithm.hash(pathInput))
+                    .add(pathInput.basePath())
+                    .add(pathInput.fixedSubPath());
             }
 
             @Override
@@ -119,7 +124,10 @@ boolean requiresHashAlgorithm() {
             @Override
             public BlobPath generatePath(PathInput pathInput, PathHashAlgorithm hashAlgorithm) {
                 assert Objects.nonNull(hashAlgorithm) : "hashAlgorithm is expected to be non-null";
-                return pathInput.basePath().add(hashAlgorithm.hash(pathInput)).add(pathInput.fixedSubPath());
+                String fixedPrefix = pathInput.fixedPrefix();
+                return pathInput.basePath()
+                    .add(Strings.isNullOrEmpty(fixedPrefix) ? hashAlgorithm.hash(pathInput) : fixedPrefix + hashAlgorithm.hash(pathInput))
+                    .add(pathInput.fixedSubPath());
             }
 
             @Override