Add support to run RemoteStoreIT with S3 integration

plugins/repository-s3/build.gradle

@@ -141,6 +141,7 @@ test {
   // this is tested explicitly in separate test tasks
   exclude '**/RepositoryCredentialsTests.class'
   exclude '**/S3RepositoryThirdPartyTests.class'
+  exclude '**/S3RemoteStoreIT.class'
 }
 
 boolean useFixture = false
@@ -252,6 +253,7 @@ processYamlRestTestResources {
 internalClusterTest {
   // this is tested explicitly in a separate test task
   exclude '**/S3RepositoryThirdPartyTests.class'
+  exclude '**/S3RemoteStoreIT.class'
 }
 
 yamlRestTest {
@@ -408,6 +410,7 @@ TaskProvider s3ThirdPartyTest = tasks.register("s3ThirdPartyTest", Test) {
   setTestClassesDirs(internalTestSourceSet.getOutput().getClassesDirs())
   setClasspath(internalTestSourceSet.getRuntimeClasspath())
   include '**/S3RepositoryThirdPartyTests.class'
+  include '**/S3RemoteStoreIT.class'
   systemProperty 'test.s3.account', s3PermanentAccessKey
   systemProperty 'test.s3.key', s3PermanentSecretKey
   systemProperty 'test.s3.bucket', s3PermanentBucket

plugins/repository-s3/src/internalClusterTest/java/org/opensearch/repositories/s3/S3RemoteStoreIT.java

@@ -0,0 +1,237 @@
+/*
+ * 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.repositories.s3;
+
+import com.carrotsearch.randomizedtesting.annotations.ThreadLeakScope;
+
+import org.opensearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.opensearch.common.SuppressForbidden;
+import org.opensearch.common.blobstore.BlobPath;
+import org.opensearch.common.settings.MockSecureSettings;
+import org.opensearch.common.settings.SecureSettings;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.unit.TimeValue;
+import org.opensearch.core.common.Strings;
+import org.opensearch.index.remote.RemoteStoreEnums;
+import org.opensearch.indices.RemoteStoreSettings;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.remotestore.RemoteStoreCoreTestCase;
+import org.opensearch.repositories.RepositoriesService;
+import org.opensearch.repositories.blobstore.BlobStoreRepository;
+import org.opensearch.threadpool.ThreadPool;
+import org.opensearch.threadpool.ThreadPoolStats;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Locale;
+import java.util.concurrent.ExecutionException;
+
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_CLUSTER_STATE_REPOSITORY_NAME_ATTRIBUTE_KEY;
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_REPOSITORY_SETTINGS_ATTRIBUTE_KEY_PREFIX;
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_REPOSITORY_TYPE_ATTRIBUTE_KEY_FORMAT;
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY;
+import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.REMOTE_STORE_TRANSLOG_REPOSITORY_NAME_ATTRIBUTE_KEY;
+import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
+
+@ThreadLeakScope(ThreadLeakScope.Scope.NONE)
+public class S3RemoteStoreIT extends RemoteStoreCoreTestCase {
+
+    @Override
+    @SuppressForbidden(reason = "Need to set system property here for AWS SDK v2")
+    public void setUp() throws Exception {
+        SocketAccess.doPrivileged(() -> System.setProperty("opensearch.path.conf", "config"));
+        super.setUp();
+    }
+
+    @Override
+    @SuppressForbidden(reason = "Need to reset system property here for AWS SDK v2")
+    public void tearDown() throws Exception {
+        SocketAccess.doPrivileged(() -> System.clearProperty("opensearch.path.conf"));
+        clearIndices();
+        waitForEmptyRemotePurgeQueue();
+        super.tearDown();
+    }
+
+    private void clearIndices() throws Exception {
+        assertAcked(client().admin().indices().delete(new DeleteIndexRequest("*")).get());
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Collections.singletonList(S3RepositoryPlugin.class);
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal)).setSecureSettings(credentials()).build();
+    }
+
+    private SecureSettings credentials() {
+        assertFalse(Strings.isNullOrEmpty(System.getProperty("test.s3.account")));
+        assertFalse(Strings.isNullOrEmpty(System.getProperty("test.s3.key")));
+        assertFalse(Strings.isNullOrEmpty(System.getProperty("test.s3.bucket")));
+
+        MockSecureSettings secureSettings = new MockSecureSettings();
+        secureSettings.setString("s3.client.default.access_key", System.getProperty("test.s3.account"));
+        secureSettings.setString("s3.client.default.secret_key", System.getProperty("test.s3.key"));
+        return secureSettings;
+    }
+
+    @Override
+    protected Settings remoteStoreRepoSettings() {
+
+        String segmentRepoName = REPOSITORY_NAME;
+        String translogRepoName = REPOSITORY_2_NAME;
+        String stateRepoName = REPOSITORY_3_NAME;
+        String segmentRepoTypeAttributeKey = String.format(
+            Locale.getDefault(),
+            "node.attr." + REMOTE_STORE_REPOSITORY_TYPE_ATTRIBUTE_KEY_FORMAT,
+            segmentRepoName
+        );
+        String segmentRepoSettingsAttributeKeyPrefix = String.format(
+            Locale.getDefault(),
+            "node.attr." + REMOTE_STORE_REPOSITORY_SETTINGS_ATTRIBUTE_KEY_PREFIX,
+            segmentRepoName
+        );
+        String translogRepoTypeAttributeKey = String.format(
+            Locale.getDefault(),
+            "node.attr." + REMOTE_STORE_REPOSITORY_TYPE_ATTRIBUTE_KEY_FORMAT,
+            translogRepoName
+        );
+        String translogRepoSettingsAttributeKeyPrefix = String.format(
+            Locale.getDefault(),
+            "node.attr." + REMOTE_STORE_REPOSITORY_SETTINGS_ATTRIBUTE_KEY_PREFIX,
+            translogRepoName
+        );
+        String stateRepoTypeAttributeKey = String.format(
+            Locale.getDefault(),
+            "node.attr." + REMOTE_STORE_REPOSITORY_TYPE_ATTRIBUTE_KEY_FORMAT,
+            stateRepoName
+        );
+        String stateRepoSettingsAttributeKeyPrefix = String.format(
+            Locale.getDefault(),
+            "node.attr." + REMOTE_STORE_REPOSITORY_SETTINGS_ATTRIBUTE_KEY_PREFIX,
+            stateRepoName
+        );
+
+        String prefixModeVerificationSuffix = BlobStoreRepository.PREFIX_MODE_VERIFICATION_SETTING.getKey();
+
+        String bucket = System.getProperty("test.s3.bucket");
+        String region = System.getProperty("test.s3.region", "us-west-2");
+        String basePath = System.getProperty("test.s3.base", "testpath");
+        String segmentBasePath = basePath + "-segments";
+        String translogBasePath = basePath + "-translog";
+        String stateBasePath = basePath + "-state";
+
+        Settings.Builder settings = Settings.builder()
+            .put("node.attr." + REMOTE_STORE_SEGMENT_REPOSITORY_NAME_ATTRIBUTE_KEY, segmentRepoName)
+            .put(segmentRepoTypeAttributeKey, S3Repository.TYPE)
+            .put(segmentRepoSettingsAttributeKeyPrefix + "bucket", bucket)
+            .put(segmentRepoSettingsAttributeKeyPrefix + "region", region)
+            .put(segmentRepoSettingsAttributeKeyPrefix + "base_path", segmentBasePath)
+            .put(segmentRepoSettingsAttributeKeyPrefix + prefixModeVerificationSuffix, prefixModeVerificationEnable)
+            .put("node.attr." + REMOTE_STORE_TRANSLOG_REPOSITORY_NAME_ATTRIBUTE_KEY, translogRepoName)
+            .put(translogRepoTypeAttributeKey, S3Repository.TYPE)
+            .put(translogRepoSettingsAttributeKeyPrefix + "bucket", bucket)
+            .put(translogRepoSettingsAttributeKeyPrefix + "region", region)
+            .put(translogRepoSettingsAttributeKeyPrefix + "base_path", translogBasePath)
+            .put(translogRepoSettingsAttributeKeyPrefix + prefixModeVerificationSuffix, prefixModeVerificationEnable)
+            .put("node.attr." + REMOTE_STORE_CLUSTER_STATE_REPOSITORY_NAME_ATTRIBUTE_KEY, stateRepoName)
+            .put(stateRepoTypeAttributeKey, S3Repository.TYPE)
+            .put(stateRepoSettingsAttributeKeyPrefix + "bucket", bucket)
+            .put(stateRepoSettingsAttributeKeyPrefix + "region", region)
+            .put(stateRepoSettingsAttributeKeyPrefix + "base_path", stateBasePath)
+            .put(stateRepoSettingsAttributeKeyPrefix + prefixModeVerificationSuffix, prefixModeVerificationEnable);
+
+        final String endpoint = System.getProperty("test.s3.endpoint");
+        if (endpoint != null) {
+            settings.put(segmentRepoSettingsAttributeKeyPrefix + "endpoint", endpoint);
+            settings.put(translogRepoSettingsAttributeKeyPrefix + "endpoint", endpoint);
+            settings.put(stateRepoSettingsAttributeKeyPrefix + "endpoint", endpoint);
+        }
+
+        settings.put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_PATH_TYPE_SETTING.getKey(), randomFrom(RemoteStoreEnums.PathType.values()));
+        settings.put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_TRANSLOG_METADATA.getKey(), randomBoolean());
+        settings.put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED.getKey(), randomBoolean());
+        settings.put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_SEGMENTS_PATH_PREFIX.getKey(), segmentsPathFixedPrefix ? "a" : "");
+        settings.put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_TRANSLOG_PATH_PREFIX.getKey(), translogPathFixedPrefix ? "b" : "");
+        settings.put(BlobStoreRepository.SNAPSHOT_SHARD_PATH_PREFIX_SETTING.getKey(), snapshotShardPathFixedPrefix ? "c" : "");
+
+        return settings.build();
+    }
+
+    @Override
+    @AwaitsFix(bugUrl = "assertion of cluster health timeout trips")
+    public void testNoMultipleWriterDuringPrimaryRelocation() throws ExecutionException, InterruptedException {
+        super.testNoMultipleWriterDuringPrimaryRelocation();
+    }
+
+    @Override
+    @AwaitsFix(bugUrl = "assertion of cluster health timeout trips")
+    public void testResumeUploadAfterFailedPrimaryRelocation() throws ExecutionException, InterruptedException, IOException {
+        super.testResumeUploadAfterFailedPrimaryRelocation();
+    }
+
+    @Override
+    @AwaitsFix(bugUrl = "Test times out due to too many translog upload")
+    public void testFlushOnTooManyRemoteTranslogFiles() throws Exception {
+        super.testFlushOnTooManyRemoteTranslogFiles();
+    }
+
+    @Override
+    protected boolean addMockIndexStorePlugin() {
+        return false;
+    }
+
+    protected BlobStoreRepository getRepository() {
+        return (BlobStoreRepository) internalCluster().getDataNodeInstance(RepositoriesService.class).repository(REPOSITORY_2_NAME);
+    }
+
+    @Override
+    protected int getActualFileCount(Path ignoredSegmentRepoPath, String shardPath) throws IOException {
+        BlobStoreRepository repository = getRepository();
+        return repository.blobStore().blobContainer(BlobPath.cleanPath().add(shardPath)).listBlobs().size();
+    }
+
+    @Override
+    protected void delete(Path baseRepoPath, String shardPath) throws IOException {
+        BlobStoreRepository repository = getRepository();
+        repository.blobStore().blobContainer(repository.basePath().add(shardPath)).delete();
+    }
+
+    private void waitForEmptyRemotePurgeQueue() throws Exception {
+        if (internalCluster().getDataNodeNames().isEmpty()) {
+            return;
+        }
+        assertBusyWithFixedSleepTime(() -> {
+            ThreadPoolStats.Stats remotePurgeThreadPoolStats = getRemotePurgeThreadPoolStats();
+            assertEquals(0, remotePurgeThreadPoolStats.getQueue());
+            assertEquals(0, remotePurgeThreadPoolStats.getQueue());
+        }, TimeValue.timeValueSeconds(60), TimeValue.timeValueMillis(500));
+    }
+
+    ThreadPoolStats.Stats getRemotePurgeThreadPoolStats() {
+        final ThreadPoolStats stats = internalCluster().getDataNodeInstance(ThreadPool.class).stats();
+        for (ThreadPoolStats.Stats s : stats) {
+            if (s.getName().equals(ThreadPool.Names.REMOTE_PURGE)) {
+                return s;
+            }
+        }
+        throw new AssertionError("refresh thread pool stats not found [" + stats + "]");
+    }
+
+    @Override
+    protected BlobPath getSegmentBasePath() {
+        String basePath = System.getProperty("test.s3.base", "testpath");
+        String segmentBasePath = basePath + "-segments";
+        return BlobPath.cleanPath().add(segmentBasePath);
+    }
+}

plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3BlobContainer.java

@@ -392,7 +392,7 @@ private <T> T getFutureValue(PlainActionFuture<T> future) throws IOException {
             return future.get();
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
-            throw new IllegalStateException("Future got interrupted", e);
+            throw new IOException("Future got interrupted", e);
         } catch (ExecutionException e) {
             if (e.getCause() instanceof IOException) {
                 throw (IOException) e.getCause();

plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3ClientSettings.java

@@ -63,7 +63,7 @@
 /**
  * A container for settings used to create an S3 client.
  */
-final class S3ClientSettings {
+public final class S3ClientSettings {
 
     private static final DeprecationLogger deprecationLogger = DeprecationLogger.getLogger(S3ClientSettings.class);
 

plugins/repository-s3/src/test/java/org/opensearch/repositories/s3/S3BlobStoreContainerTests.java

@@ -1970,7 +1970,7 @@ public void testDeleteWithInterruptedException() throws Exception {
 
         final S3BlobContainer blobContainer = new S3BlobContainer(blobPath, blobStore);
 
-        IllegalStateException e = expectThrows(IllegalStateException.class, blobContainer::delete);
+        IOException e = expectThrows(IOException.class, blobContainer::delete);
         assertEquals("Future got interrupted", e.getMessage());
         assertTrue(Thread.interrupted()); // Clear interrupted state
     }
@@ -2026,7 +2026,7 @@ public void testDeleteBlobsIgnoringIfNotExistsWithInterruptedException() throws
         final S3BlobContainer blobContainer = new S3BlobContainer(blobPath, blobStore);
         List<String> blobNames = Arrays.asList("test1", "test2");
 
-        IllegalStateException e = expectThrows(IllegalStateException.class, () -> blobContainer.deleteBlobsIgnoringIfNotExists(blobNames));
+        IOException e = expectThrows(IOException.class, () -> blobContainer.deleteBlobsIgnoringIfNotExists(blobNames));
         assertEquals("Future got interrupted", e.getMessage());
         assertTrue(Thread.interrupted()); // Clear interrupted state
     }