Passing in project-id when creating s3 client (elastic#129301)

Enables creating different s3 clients for different projects

Relates: elastic#127631

Author: ywangd

modules/repository-s3/qa/third-party/src/test/java/org/elasticsearch/repositories/s3/S3RepositoryThirdPartyTests.java

@@ -18,6 +18,7 @@
 import com.carrotsearch.randomizedtesting.annotations.ThreadLeakScope;
 
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.blobstore.OptionalBytesReference;
 import org.elasticsearch.common.bytes.BytesArray;
@@ -147,7 +148,7 @@ public long absoluteTimeInMillis() {
         // construct our own repo instance so we can inject a threadpool that allows to control the passage of time
         try (
             var repository = new S3Repository(
-                randomProjectIdOrDefault(),
+                ProjectId.DEFAULT,
                 node().injector().getInstance(RepositoriesService.class).repository(TEST_REPO_NAME).getMetadata(),
                 xContentRegistry(),
                 node().injector().getInstance(PluginsService.class).filterPlugins(S3RepositoryPlugin.class).findFirst().get().getService(),

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3BlobStore.java

@@ -140,7 +140,7 @@ class S3BlobStore implements BlobStore {
         this.bulkDeletionBatchSize = S3Repository.DELETION_BATCH_SIZE_SETTING.get(repositoryMetadata.settings());
         this.retryThrottledDeleteBackoffPolicy = retryThrottledDeleteBackoffPolicy;
         this.getRegisterRetryDelay = S3Repository.GET_REGISTER_RETRY_DELAY.get(repositoryMetadata.settings());
-        this.addPurposeCustomQueryParameter = service.settings(repositoryMetadata).addPurposeCustomQueryParameter;
+        this.addPurposeCustomQueryParameter = service.settings(projectId, repositoryMetadata).addPurposeCustomQueryParameter;
     }
 
     MetricPublisher getMetricPublisher(Operation operation, OperationPurpose purpose) {
@@ -263,12 +263,11 @@ public String toString() {
     }
 
     public AmazonS3Reference clientReference() {
-        // TODO: use service.client(ProjectId, RepositoryMetadata), see https://github.com/elastic/elasticsearch/pull/127631
-        return service.client(repositoryMetadata);
+        return service.client(projectId, repositoryMetadata);
     }
 
     final int getMaxRetries() {
-        return service.settings(repositoryMetadata).maxRetries;
+        return service.settings(projectId, repositoryMetadata).maxRetries;
     }
 
     public String bucket() {
@@ -441,7 +440,7 @@ private static DeleteObjectsRequest bulkDelete(OperationPurpose purpose, S3BlobS
 
     @Override
     public void close() throws IOException {
-        service.onBlobStoreClose();
+        service.onBlobStoreClose(projectId);
     }
 
     @Override

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3Service.java

@@ -49,7 +49,6 @@
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.RunOnce;
-import org.elasticsearch.core.FixForMultiProject;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
@@ -157,15 +156,6 @@ public synchronized void refreshAndClearCache(Map<String, S3ClientSettings> clie
         s3ClientsManager.refreshAndClearCacheForClusterClients(clientsSettings);
     }
 
-    /**
-     * Attempts to retrieve a client by its repository metadata and settings from the cache.
-     * If the client does not exist it will be created.
-     */
-    @FixForMultiProject(description = "can be removed once blobstore is project aware")
-    public AmazonS3Reference client(RepositoryMetadata repositoryMetadata) {
-        return client(ProjectId.DEFAULT, repositoryMetadata);
-    }
-
     /**
      * Attempts to retrieve either a cluster or project client from the client manager. Throws if project-id or
      * the client name does not exist. The client maybe initialized lazily.
@@ -196,11 +186,6 @@ private AmazonS3Reference buildClientReference(final S3ClientSettings clientSett
         }
     }
 
-    @FixForMultiProject(description = "can be removed once blobstore is project aware")
-    S3ClientSettings settings(RepositoryMetadata repositoryMetadata) {
-        return settings(ProjectId.DEFAULT, repositoryMetadata);
-    }
-
     S3ClientSettings settings(@Nullable ProjectId projectId, RepositoryMetadata repositoryMetadata) {
         return s3ClientsManager.settingsForClient(effectiveProjectId(projectId), repositoryMetadata);
     }
@@ -420,11 +405,6 @@ static AwsCredentialsProvider buildCredentials(
         }
     }
 
-    @FixForMultiProject(description = "can be removed once blobstore is project aware")
-    public void onBlobStoreClose() {
-        onBlobStoreClose(ProjectId.DEFAULT);
-    }
-
     /**
      * Release clients for the specified project.
      * @param projectId The project associated with the client, or null if the client is cluster level

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/AwsS3ServiceImplTests.java

@@ -21,6 +21,7 @@
 
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.util.Supplier;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.project.TestProjectResolvers;
 import org.elasticsearch.common.settings.MockSecureSettings;
@@ -246,7 +247,10 @@ public void testEndPointAndRegionOverrides() throws IOException {
             s3Service.start();
             final String endpointOverride = "http://first";
             final Settings settings = Settings.builder().put("endpoint", endpointOverride).build();
-            final AmazonS3Reference reference = s3Service.client(new RepositoryMetadata("first", "s3", settings));
+            final AmazonS3Reference reference = s3Service.client(
+                randomFrom(ProjectId.DEFAULT, null),
+                new RepositoryMetadata("first", "s3", settings)
+            );
 
             assertEquals(endpointOverride, reference.client().serviceClientConfiguration().endpointOverride().get().toString());
             assertEquals("es-test-region", reference.client().serviceClientConfiguration().region().toString());

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3BlobContainerRetriesTests.java

@@ -21,6 +21,7 @@
 import org.apache.http.conn.DnsResolver;
 import org.apache.logging.log4j.Level;
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.project.TestProjectResolvers;
 import org.elasticsearch.common.BackoffPolicy;
@@ -239,7 +240,7 @@ protected BlobContainer createBlobContainer(
         final RepositoryMetadata repositoryMetadata = new RepositoryMetadata("repository", S3Repository.TYPE, repositorySettings.build());
 
         final S3BlobStore s3BlobStore = new S3BlobStore(
-            randomProjectIdOrDefault(),
+            ProjectId.DEFAULT,
             service,
             "bucket",
             S3Repository.SERVER_SIDE_ENCRYPTION_SETTING.getDefault(Settings.EMPTY),

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3RepositoryTests.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.MockBigArrays;
 import org.elasticsearch.common.util.concurrent.DeterministicTaskQueue;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.indices.recovery.RecoverySettings;
 import org.elasticsearch.repositories.RepositoryException;
@@ -71,7 +72,7 @@ private static class DummyS3Service extends S3Service {
         }
 
         @Override
-        public AmazonS3Reference client(RepositoryMetadata repositoryMetadata) {
+        public AmazonS3Reference client(@Nullable ProjectId projectId, RepositoryMetadata repositoryMetadata) {
             return new AmazonS3Reference(new DummyS3Client(), mock(SdkHttpClient.class));
         }
 

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3ServiceTests.java

@@ -19,6 +19,7 @@
 import software.amazon.awssdk.services.s3.model.S3Exception;
 
 import org.apache.logging.log4j.Level;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.project.TestProjectResolvers;
 import org.elasticsearch.common.Strings;
@@ -53,17 +54,17 @@ public void testCachedClientsAreReleased() throws IOException {
         final Settings settings = Settings.builder().put("endpoint", "http://first").build();
         final RepositoryMetadata metadata1 = new RepositoryMetadata("first", "s3", settings);
         final RepositoryMetadata metadata2 = new RepositoryMetadata("second", "s3", settings);
-        final S3ClientSettings clientSettings = s3Service.settings(metadata2);
-        final S3ClientSettings otherClientSettings = s3Service.settings(metadata2);
+        final S3ClientSettings clientSettings = s3Service.settings(ProjectId.DEFAULT, metadata2);
+        final S3ClientSettings otherClientSettings = s3Service.settings(ProjectId.DEFAULT, metadata2);
         assertSame(clientSettings, otherClientSettings);
-        final AmazonS3Reference reference = s3Service.client(metadata1);
+        final AmazonS3Reference reference = s3Service.client(randomFrom(ProjectId.DEFAULT, null), metadata1);
         reference.close();
-        s3Service.onBlobStoreClose();
-        final AmazonS3Reference referenceReloaded = s3Service.client(metadata1);
+        s3Service.onBlobStoreClose(ProjectId.DEFAULT);
+        final AmazonS3Reference referenceReloaded = s3Service.client(randomFrom(ProjectId.DEFAULT, null), metadata1);
         assertNotSame(referenceReloaded, reference);
         referenceReloaded.close();
-        s3Service.onBlobStoreClose();
-        final S3ClientSettings clientSettingsReloaded = s3Service.settings(metadata1);
+        s3Service.onBlobStoreClose(ProjectId.DEFAULT);
+        final S3ClientSettings clientSettingsReloaded = s3Service.settings(ProjectId.DEFAULT, metadata1);
         assertNotSame(clientSettings, clientSettingsReloaded);
         s3Service.close();
     }

test/framework/src/main/java/org/elasticsearch/test/ESTestCase.java

@@ -2743,7 +2743,7 @@ public static void runInParallel(int numberOfTasks, IntConsumer taskFactory) {
                 future.run();
             } else {
                 threads[i] = new Thread(future);
-                threads[i].setName("runInParallel-T#" + i);
+                threads[i].setName("TEST-runInParallel-T#" + i);
                 threads[i].start();
             }
         }