Add support to upload snapshot shard blobs with hashed prefix (#15426)

* Add snapshot shard blobs with hashed prefix Signed-off-by: Ashish Singh <ssashish@amazon.com> * Add UTs Signed-off-by: Ashish Singh <ssashish@amazon.com> * Address comments and add UTs Signed-off-by: Ashish Singh <ssashish@amazon.com> * Change default snapshot shard path type to hashed_prefix for testing Signed-off-by: Ashish Singh <ssashish@amazon.com> * Fix failing tests Signed-off-by: Ashish Singh <ssashish@amazon.com> * Introduce single method for all create/put repository calls Signed-off-by: Ashish Singh <ssashish@amazon.com> * Move the index shard path upload to sync upload Signed-off-by: Ashish Singh <ssashish@amazon.com> * Fix spotless errors Signed-off-by: Ashish Singh <ssashish@amazon.com> * Do minor code refactor Signed-off-by: Ashish Singh <ssashish@amazon.com> * Fix tests Signed-off-by: Ashish Singh <ssashish@amazon.com> * More tests to use common methods for repo creation Signed-off-by: Ashish Singh <ssashish@amazon.com> * Address comments and fix multiple failing tests Signed-off-by: Ashish Singh <ssashish@amazon.com> --------- Signed-off-by: Ashish Singh <ssashish@amazon.com>
opensearch-project · Sep 1, 2024 · c308b98 · c308b98
1 parent 71d122b
commit c308b98
Show file tree

Hide file tree

Showing 60 changed files with 1,975 additions and 927 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -39,6 +39,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add fieldType to AbstractQueryBuilder and FieldSortBuilder ([#15328](https://github.com/opensearch-project/OpenSearch/pull/15328)))
 - [Reader Writer Separation] Add searchOnly replica routing configuration ([#15410](https://github.com/opensearch-project/OpenSearch/pull/15410))
 - [Workload Management] Add query group level failure tracking ([#15227](https://github.com/opensearch-project/OpenSearch/pull/15527))
+- Add support to upload snapshot shard blobs with hashed prefix ([#15426](https://github.com/opensearch-project/OpenSearch/pull/15426))
 
 ### 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/...rl/src/internalClusterTest/java/org/opensearch/repositories/url/URLSnapshotRestoreIT.java b/...rl/src/internalClusterTest/java/org/opensearch/repositories/url/URLSnapshotRestoreIT.java
@@ -67,19 +67,11 @@ public void testUrlRepository() throws Exception {
 
         logger.info("-->  creating repository");
         Path repositoryLocation = randomRepoPath();
-        assertAcked(
-            client.admin()
-                .cluster()
-                .preparePutRepository("test-repo")
-                .setType(FsRepository.TYPE)
-                .setSettings(
-                    Settings.builder()
-                        .put(FsRepository.LOCATION_SETTING.getKey(), repositoryLocation)
-                        .put(FsRepository.COMPRESS_SETTING.getKey(), randomBoolean())
-                        .put(FsRepository.CHUNK_SIZE_SETTING.getKey(), randomIntBetween(100, 1000), ByteSizeUnit.BYTES)
-                )
-        );
-
+        Settings.Builder settings = Settings.builder()
+            .put(FsRepository.LOCATION_SETTING.getKey(), repositoryLocation)
+            .put(FsRepository.COMPRESS_SETTING.getKey(), randomBoolean())
+            .put(FsRepository.CHUNK_SIZE_SETTING.getKey(), randomIntBetween(100, 1000), ByteSizeUnit.BYTES);
+        createRepository("test-repo", FsRepository.TYPE, settings);
         createIndex("test-idx");
         ensureGreen();
 
@@ -115,17 +107,10 @@ public void testUrlRepository() throws Exception {
         cluster().wipeIndices("test-idx");
 
         logger.info("--> create read-only URL repository");
-        assertAcked(
-            client.admin()
-                .cluster()
-                .preparePutRepository("url-repo")
-                .setType(URLRepository.TYPE)
-                .setSettings(
-                    Settings.builder()
-                        .put(URLRepository.URL_SETTING.getKey(), repositoryLocation.toUri().toURL().toString())
-                        .put("list_directories", randomBoolean())
-                )
-        );
+        Settings.Builder settingsBuilder = Settings.builder()
+            .put(URLRepository.URL_SETTING.getKey(), repositoryLocation.toUri().toURL().toString())
+            .put("list_directories", randomBoolean());
+        createRepository("url-repo", URLRepository.TYPE, settingsBuilder);
         logger.info("--> restore index after deletion");
         RestoreSnapshotResponse restoreSnapshotResponse = client.admin()
             .cluster()

diff --git a/...lusterTest/java/org/opensearch/repositories/azure/AzureStorageCleanupThirdPartyTests.java b/...lusterTest/java/org/opensearch/repositories/azure/AzureStorageCleanupThirdPartyTests.java
@@ -38,7 +38,6 @@
 import com.azure.storage.blob.models.BlobStorageException;
 import org.opensearch.action.ActionRunnable;
 import org.opensearch.action.support.PlainActionFuture;
-import org.opensearch.action.support.master.AcknowledgedResponse;
 import org.opensearch.common.collect.Tuple;
 import org.opensearch.common.settings.MockSecureSettings;
 import org.opensearch.common.settings.SecureSettings;
@@ -47,6 +46,7 @@
 import org.opensearch.plugins.Plugin;
 import org.opensearch.repositories.AbstractThirdPartyRepositoryTestCase;
 import org.opensearch.repositories.blobstore.BlobStoreRepository;
+import org.opensearch.test.OpenSearchIntegTestCase;
 import org.junit.AfterClass;
 
 import java.net.HttpURLConnection;
@@ -56,7 +56,6 @@
 import reactor.core.scheduler.Schedulers;
 
 import static org.hamcrest.Matchers.blankOrNullString;
-import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.not;
 
 public class AzureStorageCleanupThirdPartyTests extends AbstractThirdPartyRepositoryTestCase {
@@ -103,17 +102,11 @@ protected SecureSettings credentials() {
 
     @Override
     protected void createRepository(String repoName) {
-        AcknowledgedResponse putRepositoryResponse = client().admin()
-            .cluster()
-            .preparePutRepository(repoName)
-            .setType("azure")
-            .setSettings(
-                Settings.builder()
-                    .put("container", System.getProperty("test.azure.container"))
-                    .put("base_path", System.getProperty("test.azure.base"))
-            )
-            .get();
-        assertThat(putRepositoryResponse.isAcknowledged(), equalTo(true));
+        Settings.Builder settings = Settings.builder()
+            .put("container", System.getProperty("test.azure.container"))
+            .put("base_path", System.getProperty("test.azure.base"));
+
+        OpenSearchIntegTestCase.putRepository(client().admin().cluster(), repoName, "azure", settings);
         if (Strings.hasText(System.getProperty("test.azure.sas_token"))) {
             ensureSasTokenPermissions();
         }

diff --git a/...alClusterTest/java/org/opensearch/repositories/gcs/GoogleCloudStorageThirdPartyTests.java b/...alClusterTest/java/org/opensearch/repositories/gcs/GoogleCloudStorageThirdPartyTests.java
@@ -32,19 +32,18 @@
 
 package org.opensearch.repositories.gcs;
 
-import org.opensearch.action.support.master.AcknowledgedResponse;
 import org.opensearch.common.settings.MockSecureSettings;
 import org.opensearch.common.settings.SecureSettings;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.core.common.Strings;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.repositories.AbstractThirdPartyRepositoryTestCase;
+import org.opensearch.test.OpenSearchIntegTestCase;
 
 import java.util.Base64;
 import java.util.Collection;
 
 import static org.hamcrest.Matchers.blankOrNullString;
-import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.not;
 
 public class GoogleCloudStorageThirdPartyTests extends AbstractThirdPartyRepositoryTestCase {
@@ -84,16 +83,9 @@ protected SecureSettings credentials() {
 
     @Override
     protected void createRepository(final String repoName) {
-        AcknowledgedResponse putRepositoryResponse = client().admin()
-            .cluster()
-            .preparePutRepository("test-repo")
-            .setType("gcs")
-            .setSettings(
-                Settings.builder()
-                    .put("bucket", System.getProperty("test.google.bucket"))
-                    .put("base_path", System.getProperty("test.google.base", "/"))
-            )
-            .get();
-        assertThat(putRepositoryResponse.isAcknowledged(), equalTo(true));
+        Settings.Builder settings = Settings.builder()
+            .put("bucket", System.getProperty("test.google.bucket"))
+            .put("base_path", System.getProperty("test.google.base", "/"));
+        OpenSearchIntegTestCase.putRepository(client().admin().cluster(), "test-repo", "gcs", settings);
     }
 }
diff --git a/...s/repository-hdfs/src/test/java/org/opensearch/repositories/hdfs/HdfsRepositoryTests.java b/...s/repository-hdfs/src/test/java/org/opensearch/repositories/hdfs/HdfsRepositoryTests.java
@@ -34,12 +34,12 @@
 import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
 
 import org.opensearch.action.admin.cluster.repositories.cleanup.CleanupRepositoryResponse;
-import org.opensearch.action.support.master.AcknowledgedResponse;
 import org.opensearch.common.settings.MockSecureSettings;
 import org.opensearch.common.settings.SecureSettings;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.repositories.AbstractThirdPartyRepositoryTestCase;
+import org.opensearch.test.OpenSearchIntegTestCase;
 
 import java.util.Collection;
 
@@ -61,20 +61,13 @@ protected SecureSettings credentials() {
 
     @Override
     protected void createRepository(String repoName) {
-        AcknowledgedResponse putRepositoryResponse = client().admin()
-            .cluster()
-            .preparePutRepository(repoName)
-            .setType("hdfs")
-            .setSettings(
-                Settings.builder()
-                    .put("uri", "hdfs:///")
-                    .put("conf.fs.AbstractFileSystem.hdfs.impl", TestingFs.class.getName())
-                    .put("path", "foo")
-                    .put("chunk_size", randomIntBetween(100, 1000) + "k")
-                    .put("compress", randomBoolean())
-            )
-            .get();
-        assertThat(putRepositoryResponse.isAcknowledged(), equalTo(true));
+        Settings.Builder settings = Settings.builder()
+            .put("uri", "hdfs:///")
+            .put("conf.fs.AbstractFileSystem.hdfs.impl", TestingFs.class.getName())
+            .put("path", "foo")
+            .put("chunk_size", randomIntBetween(100, 1000) + "k")
+            .put("compress", randomBoolean());
+        OpenSearchIntegTestCase.putRepository(client().admin().cluster(), repoName, "hdfs", settings);
     }
 
     // HDFS repository doesn't have precise cleanup stats so we only check whether or not any blobs were removed

diff --git a/plugins/repository-hdfs/src/test/java/org/opensearch/repositories/hdfs/HdfsTests.java b/plugins/repository-hdfs/src/test/java/org/opensearch/repositories/hdfs/HdfsTests.java
@@ -35,7 +35,6 @@
 
 import org.opensearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse;
 import org.opensearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
-import org.opensearch.action.support.master.AcknowledgedResponse;
 import org.opensearch.client.Client;
 import org.opensearch.cluster.ClusterState;
 import org.opensearch.common.settings.Settings;
@@ -45,6 +44,7 @@
 import org.opensearch.repositories.blobstore.BlobStoreRepository;
 import org.opensearch.repositories.blobstore.BlobStoreTestUtil;
 import org.opensearch.snapshots.SnapshotState;
+import org.opensearch.test.OpenSearchIntegTestCase;
 import org.opensearch.test.OpenSearchSingleNodeTestCase;
 import org.opensearch.threadpool.ThreadPool;
 
@@ -63,21 +63,13 @@ protected Collection<Class<? extends Plugin>> getPlugins() {
 
     public void testSimpleWorkflow() {
         Client client = client();
-
-        AcknowledgedResponse putRepositoryResponse = client.admin()
-            .cluster()
-            .preparePutRepository("test-repo")
-            .setType("hdfs")
-            .setSettings(
-                Settings.builder()
-                    .put("uri", "hdfs:///")
-                    .put("conf.fs.AbstractFileSystem.hdfs.impl", TestingFs.class.getName())
-                    .put("path", "foo")
-                    .put("chunk_size", randomIntBetween(100, 1000) + "k")
-                    .put("compress", randomBoolean())
-            )
-            .get();
-        assertThat(putRepositoryResponse.isAcknowledged(), equalTo(true));
+        Settings.Builder settings = Settings.builder()
+            .put("uri", "hdfs:///")
+            .put("conf.fs.AbstractFileSystem.hdfs.impl", TestingFs.class.getName())
+            .put("path", "foo")
+            .put("chunk_size", randomIntBetween(100, 1000) + "k")
+            .put("compress", randomBoolean());
+        OpenSearchIntegTestCase.putRepository(client.admin().cluster(), "test-repo", "hdfs", settings);
 
         createIndex("test-idx-1");
         createIndex("test-idx-2");
@@ -168,7 +160,7 @@ public void testSimpleWorkflow() {
 
     public void testMissingUri() {
         try {
-            client().admin().cluster().preparePutRepository("test-repo").setType("hdfs").setSettings(Settings.EMPTY).get();
+            OpenSearchIntegTestCase.putRepository(client().admin().cluster(), "test-repo", "hdfs", Settings.builder());
             fail();
         } catch (RepositoryException e) {
             assertTrue(e.getCause() instanceof IllegalArgumentException);
@@ -178,12 +170,8 @@ public void testMissingUri() {
 
     public void testEmptyUri() {
         try {
-            client().admin()
-                .cluster()
-                .preparePutRepository("test-repo")
-                .setType("hdfs")
-                .setSettings(Settings.builder().put("uri", "/path").build())
-                .get();
+            Settings.Builder settings = Settings.builder().put("uri", "/path");
+            OpenSearchIntegTestCase.putRepository(client().admin().cluster(), "test-repo", "hdfs", settings);
             fail();
         } catch (RepositoryException e) {
             assertTrue(e.getCause() instanceof IllegalArgumentException);
@@ -193,12 +181,8 @@ public void testEmptyUri() {
 
     public void testNonHdfsUri() {
         try {
-            client().admin()
-                .cluster()
-                .preparePutRepository("test-repo")
-                .setType("hdfs")
-                .setSettings(Settings.builder().put("uri", "file:///").build())
-                .get();
+            Settings.Builder settings = Settings.builder().put("uri", "file:///");
+            OpenSearchIntegTestCase.putRepository(client().admin().cluster(), "test-repo", "hdfs", settings);
             fail();
         } catch (RepositoryException e) {
             assertTrue(e.getCause() instanceof IllegalArgumentException);
@@ -208,12 +192,8 @@ public void testNonHdfsUri() {
 
     public void testPathSpecifiedInHdfs() {
         try {
-            client().admin()
-                .cluster()
-                .preparePutRepository("test-repo")
-                .setType("hdfs")
-                .setSettings(Settings.builder().put("uri", "hdfs:///some/path").build())
-                .get();
+            Settings.Builder settings = Settings.builder().put("uri", "hdfs:///some/path");
+            OpenSearchIntegTestCase.putRepository(client().admin().cluster(), "test-repo", "hdfs", settings);
             fail();
         } catch (RepositoryException e) {
             assertTrue(e.getCause() instanceof IllegalArgumentException);
@@ -223,12 +203,8 @@ public void testPathSpecifiedInHdfs() {
 
     public void testMissingPath() {
         try {
-            client().admin()
-                .cluster()
-                .preparePutRepository("test-repo")
-                .setType("hdfs")
-                .setSettings(Settings.builder().put("uri", "hdfs:///").build())
-                .get();
+            Settings.Builder settings = Settings.builder().put("uri", "hdfs:///");
+            OpenSearchIntegTestCase.putRepository(client().admin().cluster(), "test-repo", "hdfs", settings);
             fail();
         } catch (RepositoryException e) {
             assertTrue(e.getCause() instanceof IllegalArgumentException);

diff --git a/.../internalClusterTest/java/org/opensearch/repositories/s3/S3RepositoryThirdPartyTests.java b/.../internalClusterTest/java/org/opensearch/repositories/s3/S3RepositoryThirdPartyTests.java
@@ -33,7 +33,6 @@
 
 import software.amazon.awssdk.services.s3.model.StorageClass;
 
-import org.opensearch.action.support.master.AcknowledgedResponse;
 import org.opensearch.common.SuppressForbidden;
 import org.opensearch.common.blobstore.BlobMetadata;
 import org.opensearch.common.blobstore.BlobPath;
@@ -43,6 +42,7 @@
 import org.opensearch.plugins.Plugin;
 import org.opensearch.repositories.AbstractThirdPartyRepositoryTestCase;
 import org.opensearch.repositories.blobstore.BlobStoreRepository;
+import org.opensearch.test.OpenSearchIntegTestCase;
 import org.junit.Before;
 
 import java.util.Collection;
@@ -51,7 +51,6 @@
 import java.util.concurrent.TimeUnit;
 
 import static org.hamcrest.Matchers.blankOrNullString;
-import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.not;
 
 public class S3RepositoryThirdPartyTests extends AbstractThirdPartyRepositoryTestCase {
@@ -111,13 +110,7 @@ protected void createRepository(String repoName) {
                 settings.put("storage_class", storageClass);
             }
         }
-        AcknowledgedResponse putRepositoryResponse = client().admin()
-            .cluster()
-            .preparePutRepository("test-repo")
-            .setType("s3")
-            .setSettings(settings)
-            .get();
-        assertThat(putRepositoryResponse.isAcknowledged(), equalTo(true));
+        OpenSearchIntegTestCase.putRepository(client().admin().cluster(), "test-repo", "s3", settings);
     }
 
     @Override

diff --git a/...epository-s3/src/test/java/org/opensearch/repositories/s3/RepositoryCredentialsTests.java b/...epository-s3/src/test/java/org/opensearch/repositories/s3/RepositoryCredentialsTests.java
@@ -55,6 +55,7 @@
 import org.opensearch.rest.RestRequest;
 import org.opensearch.rest.RestResponse;
 import org.opensearch.rest.action.admin.cluster.RestGetRepositoriesAction;
+import org.opensearch.test.OpenSearchIntegTestCase;
 import org.opensearch.test.OpenSearchSingleNodeTestCase;
 import org.opensearch.test.rest.FakeRestRequest;
 
@@ -68,7 +69,6 @@
 
 import static org.opensearch.repositories.s3.S3ClientSettings.ACCESS_KEY_SETTING;
 import static org.opensearch.repositories.s3.S3ClientSettings.SECRET_KEY_SETTING;
-import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
@@ -277,14 +277,8 @@ public void sendResponse(RestResponse response) {
     }
 
     private void createRepository(final String name, final Settings repositorySettings) {
-        assertAcked(
-            client().admin()
-                .cluster()
-                .preparePutRepository(name)
-                .setType(S3Repository.TYPE)
-                .setVerify(false)
-                .setSettings(repositorySettings)
-        );
+        Settings.Builder settings = Settings.builder().put(repositorySettings);
+        OpenSearchIntegTestCase.putRepository(client().admin().cluster(), name, S3Repository.TYPE, false, settings);
     }
 
     /**