Make repositories project aware (#128285)

Pass project-id explicitly to repository factory and make it part of the
repository interface.

Relates: ES-11839

Author: ywangd

modules/repository-azure/src/main/java/org/elasticsearch/repositories/azure/AzureRepository.java

@@ -11,6 +11,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
@@ -110,6 +111,7 @@ public static final class Repository {
     private final RepositoriesMetrics repositoriesMetrics;
 
     public AzureRepository(
+        final ProjectId projectId,
         final RepositoryMetadata metadata,
         final NamedXContentRegistry namedXContentRegistry,
         final AzureStorageService storageService,
@@ -119,6 +121,7 @@ public AzureRepository(
         final RepositoriesMetrics repositoriesMetrics
     ) {
         super(
+            projectId,
             metadata,
             namedXContentRegistry,
             clusterService,

modules/repository-azure/src/main/java/org/elasticsearch/repositories/azure/AzureRepositoryPlugin.java

@@ -62,10 +62,11 @@ public Map<String, Repository.Factory> getRepositories(
         RecoverySettings recoverySettings,
         RepositoriesMetrics repositoriesMetrics
     ) {
-        return Collections.singletonMap(AzureRepository.TYPE, metadata -> {
+        return Collections.singletonMap(AzureRepository.TYPE, (projectId, metadata) -> {
             AzureStorageService storageService = azureStoreService.get();
             assert storageService != null;
             return new AzureRepository(
+                projectId,
                 metadata,
                 namedXContentRegistry,
                 storageService,

modules/repository-azure/src/test/java/org/elasticsearch/repositories/azure/AzureRepositorySettingsTests.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.repositories.azure;
 
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
@@ -23,6 +24,7 @@
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 
 import static org.elasticsearch.repositories.blobstore.BlobStoreRepository.READONLY_SETTING_KEY;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
 import static org.mockito.Mockito.mock;
@@ -35,7 +37,9 @@ private AzureRepository azureRepository(Settings settings) {
             .putList(Environment.PATH_DATA_SETTING.getKey(), tmpPaths())
             .put(settings)
             .build();
+        final ProjectId projectId = randomProjectIdOrDefault();
         final AzureRepository azureRepository = new AzureRepository(
+            projectId,
             new RepositoryMetadata("foo", "azure", internalSettings),
             NamedXContentRegistry.EMPTY,
             mock(AzureStorageService.class),
@@ -44,6 +48,7 @@ private AzureRepository azureRepository(Settings settings) {
             new RecoverySettings(settings, new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS)),
             RepositoriesMetrics.NOOP
         );
+        assertThat(azureRepository.getProjectId(), equalTo(projectId));
         assertThat(azureRepository.getBlobStore(), is(nullValue()));
         return azureRepository;
     }

modules/repository-gcs/src/internalClusterTest/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStoreRepositoryTests.java

@@ -276,7 +276,8 @@ public Map<String, Repository.Factory> getRepositories(
         ) {
             return Collections.singletonMap(
                 GoogleCloudStorageRepository.TYPE,
-                metadata -> new GoogleCloudStorageRepository(
+                (projectId, metadata) -> new GoogleCloudStorageRepository(
+                    projectId,
                     metadata,
                     registry,
                     this.storageService,

modules/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStoragePlugin.java

@@ -57,7 +57,8 @@ public Map<String, Repository.Factory> getRepositories(
     ) {
         return Collections.singletonMap(
             GoogleCloudStorageRepository.TYPE,
-            metadata -> new GoogleCloudStorageRepository(
+            (projectId, metadata) -> new GoogleCloudStorageRepository(
+                projectId,
                 metadata,
                 namedXContentRegistry,
                 this.storageService,

modules/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageRepository.java

@@ -11,6 +11,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.BackoffPolicy;
@@ -88,6 +89,7 @@ class GoogleCloudStorageRepository extends MeteredBlobStoreRepository {
     private final GcsRepositoryStatsCollector statsCollector;
 
     GoogleCloudStorageRepository(
+        final ProjectId projectId,
         final RepositoryMetadata metadata,
         final NamedXContentRegistry namedXContentRegistry,
         final GoogleCloudStorageService storageService,
@@ -97,6 +99,7 @@ class GoogleCloudStorageRepository extends MeteredBlobStoreRepository {
         final GcsRepositoryStatsCollector statsCollector
     ) {
         super(
+            projectId,
             metadata,
             namedXContentRegistry,
             clusterService,

modules/repository-gcs/src/test/java/org/elasticsearch/repositories/gcs/GoogleCloudStoragePluginTests.java

@@ -9,13 +9,22 @@
 
 package org.elasticsearch.repositories.gcs;
 
+import org.elasticsearch.cluster.metadata.RepositoryMetadata;
+import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.MockBigArrays;
+import org.elasticsearch.indices.recovery.RecoverySettings;
+import org.elasticsearch.repositories.blobstore.BlobStoreTestUtil;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.junit.Assert;
 
 import java.util.List;
 
+import static org.hamcrest.Matchers.equalTo;
+import static org.mockito.Mockito.mock;
+
 public class GoogleCloudStoragePluginTests extends ESTestCase {
 
     public void testExposedSettings() {
@@ -37,4 +46,26 @@ public void testExposedSettings() {
             settings.stream().map(Setting::getKey).toList()
         );
     }
+
+    public void testRepositoryProjectId() {
+        final var projectId = randomProjectIdOrDefault();
+        final var repository = new GoogleCloudStorageRepository(
+            projectId,
+            new RepositoryMetadata(
+                randomIdentifier(),
+                GoogleCloudStorageRepository.TYPE,
+                Settings.builder()
+                    .put(GoogleCloudStorageRepository.BUCKET.getKey(), randomIdentifier())
+                    .put(GoogleCloudStorageRepository.BASE_PATH.getKey(), randomIdentifier())
+                    .build()
+            ),
+            NamedXContentRegistry.EMPTY,
+            mock(GoogleCloudStorageService.class),
+            BlobStoreTestUtil.mockClusterService(),
+            MockBigArrays.NON_RECYCLING_INSTANCE,
+            new RecoverySettings(Settings.EMPTY, new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS)),
+            mock(GcsRepositoryStatsCollector.class)
+        );
+        assertThat(repository.getProjectId(), equalTo(projectId));
+    }
 }

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

@@ -147,6 +147,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(),
                 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/internalClusterTest/java/org/elasticsearch/repositories/s3/S3BlobStoreRepositoryTests.java

@@ -22,6 +22,7 @@
 import org.apache.logging.log4j.core.LogEvent;
 import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse;
 import org.elasticsearch.action.support.broadcast.BroadcastResponse;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
@@ -598,14 +599,24 @@ public List<Setting<?>> getSettings() {
 
         @Override
         protected S3Repository createRepository(
+            ProjectId projectId,
             RepositoryMetadata metadata,
             NamedXContentRegistry registry,
             ClusterService clusterService,
             BigArrays bigArrays,
             RecoverySettings recoverySettings,
             S3RepositoriesMetrics s3RepositoriesMetrics
         ) {
-            return new S3Repository(metadata, registry, getService(), clusterService, bigArrays, recoverySettings, s3RepositoriesMetrics) {
+            return new S3Repository(
+                projectId,
+                metadata,
+                registry,
+                getService(),
+                clusterService,
+                bigArrays,
+                recoverySettings,
+                s3RepositoriesMetrics
+            ) {
 
                 @Override
                 public BlobStore blobStore() {

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

@@ -27,6 +27,7 @@
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.common.BackoffPolicy;
 import org.elasticsearch.common.Strings;
@@ -38,6 +39,7 @@
 import org.elasticsearch.common.blobstore.OperationPurpose;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.repositories.RepositoriesMetrics;
 import org.elasticsearch.rest.RestStatus;
@@ -74,6 +76,8 @@ class S3BlobStore implements BlobStore {
 
     private static final Logger logger = LogManager.getLogger(S3BlobStore.class);
 
+    @Nullable // if the blobstore is at the cluster level
+    private final ProjectId projectId;
     private final S3Service service;
 
     private final BigArrays bigArrays;
@@ -106,6 +110,7 @@ class S3BlobStore implements BlobStore {
     private final boolean addPurposeCustomQueryParameter;
 
     S3BlobStore(
+        @Nullable ProjectId projectId,
         S3Service service,
         String bucket,
         boolean serverSideEncryption,
@@ -119,6 +124,7 @@ class S3BlobStore implements BlobStore {
         S3RepositoriesMetrics s3RepositoriesMetrics,
         BackoffPolicy retryThrottledDeleteBackoffPolicy
     ) {
+        this.projectId = projectId;
         this.service = service;
         this.bigArrays = bigArrays;
         this.bucket = bucket;
@@ -257,6 +263,7 @@ public String toString() {
     }
 
     public AmazonS3Reference clientReference() {
+        // TODO: use service.client(ProjectId, RepositoryMetadata), see https://github.com/elastic/elasticsearch/pull/127631
         return service.client(repositoryMetadata);
     }