Add prefix mode verification setting for repository verification

CHANGELOG.md

@@ -20,6 +20,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add matchesPluginSystemIndexPattern to SystemIndexRegistry ([#14750](https://github.com/opensearch-project/OpenSearch/pull/14750))
 - Add Plugin interface for loading application based configuration templates (([#14659](https://github.com/opensearch-project/OpenSearch/issues/14659)))
 - Refactor remote-routing-table service inline with remote state interfaces([#14668](https://github.com/opensearch-project/OpenSearch/pull/14668))
+- Add prefix mode verification setting for repository verification (([#14790](https://github.com/opensearch-project/OpenSearch/pull/14790)))
 
 ### Dependencies
 - Bump `org.gradle.test-retry` from 1.5.8 to 1.5.9 ([#13442](https://github.com/opensearch-project/OpenSearch/pull/13442))

server/src/main/java/org/opensearch/repositories/blobstore/BlobStoreRepository.java

@@ -109,6 +109,7 @@
 import org.opensearch.core.xcontent.XContentParser;
 import org.opensearch.index.mapper.MapperService;
 import org.opensearch.index.remote.RemoteStorePathStrategy;
+import org.opensearch.index.remote.RemoteStorePathStrategy.PathInput;
 import org.opensearch.index.snapshots.IndexShardRestoreFailedException;
 import org.opensearch.index.snapshots.IndexShardSnapshotStatus;
 import org.opensearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot;
@@ -157,6 +158,7 @@
 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.BlockingQueue;
@@ -174,6 +176,8 @@
 import java.util.stream.LongStream;
 import java.util.stream.Stream;
 
+import static org.opensearch.index.remote.RemoteStoreEnums.PathHashAlgorithm.FNV_1A_COMPOSITE_1;
+import static org.opensearch.index.remote.RemoteStoreEnums.PathType.HASHED_PREFIX;
 import static org.opensearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot.FileInfo.canonicalName;
 import static org.opensearch.repositories.blobstore.ChecksumBlobStoreFormat.SNAPSHOT_ONLY_FORMAT_PARAMS;
 
@@ -302,6 +306,16 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp
         Setting.Property.NodeScope
     );
 
+    /**
+     * Setting to enable prefix mode verification. In this mode, a hashed string is prepended at the prefix of the base
+     * path during repository verification.
+     */
+    public static final Setting<Boolean> PREFIX_MODE_VERIFICATION_SETTING = Setting.boolSetting(
+        "prefix_mode_verification",
+        false,
+        Setting.Property.NodeScope
+    );
+
     protected volatile boolean supportURLRepo;
 
     private volatile int maxShardBlobDeleteBatch;
@@ -369,6 +383,8 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp
 
     private final boolean isSystemRepository;
 
+    private final boolean prefixModeVerification;
+
     private final Object lock = new Object();
 
     private final SetOnce<BlobContainer> blobContainer = new SetOnce<>();
@@ -426,6 +442,7 @@ protected BlobStoreRepository(
         readRepositoryMetadata(repositoryMetadata);
 
         isSystemRepository = SYSTEM_REPOSITORY_SETTING.get(metadata.settings());
+        prefixModeVerification = PREFIX_MODE_VERIFICATION_SETTING.get(metadata.settings());
         this.namedXContentRegistry = namedXContentRegistry;
         this.threadPool = clusterService.getClusterApplierService().threadPool();
         this.clusterService = clusterService;
@@ -767,6 +784,10 @@ protected BlobStore getBlobStore() {
         return blobStore.get();
     }
 
+    boolean getPrefixModeVerification() {
+        return prefixModeVerification;
+    }
+
     /**
      * maintains single lazy instance of {@link BlobContainer}
      */
@@ -1918,7 +1939,7 @@ public String startVerification() {
             } else {
                 String seed = UUIDs.randomBase64UUID();
                 byte[] testBytes = Strings.toUTF8Bytes(seed);
-                BlobContainer testContainer = blobStore().blobContainer(basePath().add(testBlobPrefix(seed)));
+                BlobContainer testContainer = testContainer(seed);
                 BytesArray bytes = new BytesArray(testBytes);
                 if (isSystemRepository == false) {
                     try (InputStream stream = bytes.streamInput()) {
@@ -1936,12 +1957,26 @@ public String startVerification() {
         }
     }
 
+    /**
+     * Returns the blobContainer depending on the seed and {@code prefixModeVerification}.
+     */
+    private BlobContainer testContainer(String seed) {
+        BlobPath testBlobPath;
+        if (prefixModeVerification == true) {
+            PathInput pathInput = PathInput.builder().basePath(basePath()).indexUUID(seed).build();
+            testBlobPath = HASHED_PREFIX.path(pathInput, FNV_1A_COMPOSITE_1);
+        } else {
+            testBlobPath = basePath();
+        }
+        assert Objects.nonNull(testBlobPath);
+        return blobStore().blobContainer(testBlobPath.add(testBlobPrefix(seed)));
+    }
+
     @Override
     public void endVerification(String seed) {
         if (isReadOnly() == false) {
             try {
-                final String testPrefix = testBlobPrefix(seed);
-                blobStore().blobContainer(basePath().add(testPrefix)).delete();
+                testContainer(seed).delete();
             } catch (Exception exp) {
                 throw new RepositoryVerificationException(metadata.name(), "cannot delete test data at " + basePath(), exp);
             }
@@ -3266,7 +3301,7 @@ public void verify(String seed, DiscoveryNode localNode) {
                 );
             }
         } else {
-            BlobContainer testBlobContainer = blobStore().blobContainer(basePath().add(testBlobPrefix(seed)));
+            BlobContainer testBlobContainer = testContainer(seed);
             try {
                 BytesArray bytes = new BytesArray(seed);
                 try (InputStream stream = bytes.streamInput()) {

server/src/test/java/org/opensearch/repositories/blobstore/BlobStoreRepositoryTests.java

@@ -255,6 +255,28 @@ public void testBadChunksize() throws Exception {
         );
     }
 
+    public void testPrefixModeVerification() throws Exception {
+        final Client client = client();
+        final Path location = OpenSearchIntegTestCase.randomRepoPath(node().settings());
+        final String repositoryName = "test-repo";
+        AcknowledgedResponse putRepositoryResponse = client.admin()
+            .cluster()
+            .preparePutRepository(repositoryName)
+            .setType(REPO_TYPE)
+            .setSettings(
+                Settings.builder()
+                    .put(node().settings())
+                    .put("location", location)
+                    .put(BlobStoreRepository.PREFIX_MODE_VERIFICATION_SETTING.getKey(), true)
+            )
+            .get();
+        assertTrue(putRepositoryResponse.isAcknowledged());
+
+        final RepositoriesService repositoriesService = getInstanceFromNode(RepositoriesService.class);
+        final BlobStoreRepository repository = (BlobStoreRepository) repositoriesService.repository(repositoryName);
+        assertTrue(repository.getPrefixModeVerification());
+    }
+
     public void testFsRepositoryCompressDeprecatedIgnored() {
         final Path location = OpenSearchIntegTestCase.randomRepoPath(node().settings());
         final Settings settings = Settings.builder().put(node().settings()).put("location", location).build();

test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java

@@ -152,6 +152,7 @@
 import org.opensearch.node.remotestore.RemoteStoreNodeService;
 import org.opensearch.plugins.NetworkPlugin;
 import org.opensearch.plugins.Plugin;
+import org.opensearch.repositories.blobstore.BlobStoreRepository;
 import org.opensearch.repositories.fs.ReloadableFsRepository;
 import org.opensearch.script.MockScriptService;
 import org.opensearch.search.MockSearchService;
@@ -386,6 +387,8 @@ public abstract class OpenSearchIntegTestCase extends OpenSearchTestCase {
 
     protected static final String REMOTE_BACKED_STORAGE_REPOSITORY_NAME = "test-remote-store-repo";
 
+    private static Boolean prefixModeVerificationEnable;
+
     private Path remoteStoreRepositoryPath;
 
     private ReplicationType randomReplicationType;
@@ -394,6 +397,7 @@ public abstract class OpenSearchIntegTestCase extends OpenSearchTestCase {
 
     @BeforeClass
     public static void beforeClass() throws Exception {
+        prefixModeVerificationEnable = randomBoolean();
         testClusterRule.beforeClass();
     }
 
@@ -2645,16 +2649,21 @@ private static Settings buildRemoteStoreNodeAttributes(
             segmentRepoName
         );
 
+        String prefixModeVerificationSuffix = BlobStoreRepository.PREFIX_MODE_VERIFICATION_SETTING.getKey();
+
         Settings.Builder settings = Settings.builder()
             .put("node.attr." + REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY, segmentRepoName)
             .put(segmentRepoTypeAttributeKey, segmentRepoType)
             .put(segmentRepoSettingsAttributeKeyPrefix + "location", segmentRepoPath)
+            .put(segmentRepoSettingsAttributeKeyPrefix + prefixModeVerificationSuffix, prefixModeVerificationEnable)
             .put("node.attr." + REMOTE_STORE_TRANSLOG_REPOSITORY_NAME_ATTRIBUTE_KEY, translogRepoName)
             .put(translogRepoTypeAttributeKey, translogRepoType)
             .put(translogRepoSettingsAttributeKeyPrefix + "location", translogRepoPath)
+            .put(translogRepoSettingsAttributeKeyPrefix + prefixModeVerificationSuffix, prefixModeVerificationEnable)
             .put("node.attr." + REMOTE_STORE_CLUSTER_STATE_REPOSITORY_NAME_ATTRIBUTE_KEY, segmentRepoName)
             .put(stateRepoTypeAttributeKey, segmentRepoType)
-            .put(stateRepoSettingsAttributeKeyPrefix + "location", segmentRepoPath);
+            .put(stateRepoSettingsAttributeKeyPrefix + "location", segmentRepoPath)
+            .put(stateRepoSettingsAttributeKeyPrefix + prefixModeVerificationSuffix, prefixModeVerificationEnable);
 
         if (withRateLimiterAttributes) {
             settings.put(segmentRepoSettingsAttributeKeyPrefix + "compress", randomBoolean())