Support per-project client for GCS repository (#131899)

Similar to #127631, this PR adds per project client support for GCS
repository.

Author: ywangd

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

@@ -23,6 +23,7 @@
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.BytesRefBuilder;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.BackoffPolicy;
 import org.elasticsearch.common.blobstore.BlobContainer;
@@ -235,8 +236,8 @@ public TestGoogleCloudStoragePlugin(Settings settings) {
         }
 
         @Override
-        protected GoogleCloudStorageService createStorageService(boolean isServerless) {
-            return new GoogleCloudStorageService() {
+        protected GoogleCloudStorageService createStorageService(ClusterService clusterService, ProjectResolver projectResolver) {
+            return new GoogleCloudStorageService(clusterService, projectResolver) {
                 @Override
                 StorageOptions createStorageOptions(
                     final GoogleCloudStorageClientSettings gcsClientSettings,
@@ -280,7 +281,7 @@ public Map<String, Repository.Factory> getRepositories(
                     projectId,
                     metadata,
                     registry,
-                    this.storageService,
+                    this.storageService.get(),
                     clusterService,
                     bigArrays,
                     recoverySettings,
@@ -289,10 +290,11 @@ public Map<String, Repository.Factory> getRepositories(
                     @Override
                     protected GoogleCloudStorageBlobStore createBlobStore() {
                         return new GoogleCloudStorageBlobStore(
+                            getProjectId(),
                             metadata.settings().get("bucket"),
                             "test",
                             metadata.name(),
-                            storageService,
+                            storageService.get(),
                             bigArrays,
                             randomIntBetween(1, 8) * 1024,
                             BackoffPolicy.noBackoff(),

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

@@ -23,6 +23,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.common.BackoffPolicy;
 import org.elasticsearch.common.blobstore.BlobContainer;
 import org.elasticsearch.common.blobstore.BlobPath;
@@ -40,6 +41,7 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.core.CheckedConsumer;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Streams;
 import org.elasticsearch.core.SuppressForbidden;
 import org.elasticsearch.core.TimeValue;
@@ -104,6 +106,8 @@ class GoogleCloudStorageBlobStore implements BlobStore {
         }
     }
 
+    @Nullable // for cluster level object store in MP
+    private final ProjectId projectId;
     private final String bucketName;
     private final String clientName;
     private final String repositoryName;
@@ -114,6 +118,7 @@ class GoogleCloudStorageBlobStore implements BlobStore {
     private final BackoffPolicy casBackoffPolicy;
 
     GoogleCloudStorageBlobStore(
+        ProjectId projectId,
         String bucketName,
         String clientName,
         String repositoryName,
@@ -123,6 +128,7 @@ class GoogleCloudStorageBlobStore implements BlobStore {
         BackoffPolicy casBackoffPolicy,
         GcsRepositoryStatsCollector statsCollector
     ) {
+        this.projectId = projectId;
         this.bucketName = bucketName;
         this.clientName = clientName;
         this.repositoryName = repositoryName;
@@ -134,7 +140,7 @@ class GoogleCloudStorageBlobStore implements BlobStore {
     }
 
     private MeteredStorage client() throws IOException {
-        return storageService.client(clientName, repositoryName, statsCollector);
+        return storageService.client(projectId, clientName, repositoryName, statsCollector);
     }
 
     @Override
@@ -144,7 +150,7 @@ public BlobContainer blobContainer(BlobPath path) {
 
     @Override
     public void close() {
-        storageService.closeRepositoryClients(repositoryName);
+        storageService.closeRepositoryClients(projectId, repositoryName);
     }
 
     /**

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

@@ -31,6 +31,7 @@
 import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Objects;
 
 import static org.elasticsearch.common.settings.Setting.timeSetting;
 
@@ -196,6 +197,25 @@ public Proxy getProxy() {
         return proxy;
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) return false;
+        GoogleCloudStorageClientSettings that = (GoogleCloudStorageClientSettings) o;
+        return Objects.equals(credential, that.credential)
+            && Objects.equals(endpoint, that.endpoint)
+            && Objects.equals(projectId, that.projectId)
+            && Objects.equals(connectTimeout, that.connectTimeout)
+            && Objects.equals(readTimeout, that.readTimeout)
+            && Objects.equals(applicationName, that.applicationName)
+            && Objects.equals(tokenUri, that.tokenUri)
+            && Objects.equals(proxy, that.proxy);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(credential, endpoint, projectId, connectTimeout, readTimeout, applicationName, tokenUri, proxy);
+    }
+
     public static Map<String, GoogleCloudStorageClientSettings> load(final Settings settings) {
         final Map<String, GoogleCloudStorageClientSettings> clients = new HashMap<>();
         for (final String clientName : settings.getGroups(PREFIX).keySet()) {

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

@@ -9,7 +9,8 @@
 
 package org.elasticsearch.repositories.gcs;
 
-import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.apache.lucene.util.SetOnce;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
@@ -24,26 +25,24 @@
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
 public class GoogleCloudStoragePlugin extends Plugin implements RepositoryPlugin, ReloadablePlugin {
 
+    private final Settings settings;
     // package-private for tests
-    final GoogleCloudStorageService storageService;
+    final SetOnce<GoogleCloudStorageService> storageService = new SetOnce<>();
 
-    @SuppressWarnings("this-escape")
     public GoogleCloudStoragePlugin(final Settings settings) {
-        var isServerless = DiscoveryNode.isStateless(settings);
-        this.storageService = createStorageService(isServerless);
-        // eagerly load client settings so that secure settings are readable (not closed)
-        reload(settings);
+        this.settings = settings;
     }
 
     // overridable for tests
-    protected GoogleCloudStorageService createStorageService(boolean isServerless) {
-        return new GoogleCloudStorageService(isServerless);
+    protected GoogleCloudStorageService createStorageService(ClusterService clusterService, ProjectResolver projectResolver) {
+        return new GoogleCloudStorageService(clusterService, projectResolver);
     }
 
     @Override
@@ -61,7 +60,7 @@ public Map<String, Repository.Factory> getRepositories(
                 projectId,
                 metadata,
                 namedXContentRegistry,
-                this.storageService,
+                this.storageService.get(),
                 clusterService,
                 bigArrays,
                 recoverySettings,
@@ -70,6 +69,15 @@ public Map<String, Repository.Factory> getRepositories(
         );
     }
 
+    @Override
+    public Collection<?> createComponents(PluginServices services) {
+        final ClusterService clusterService = services.clusterService();
+        storageService.set(createStorageService(clusterService, services.projectResolver()));
+        // eagerly load client settings so that secure settings are readable (not closed)
+        reload(settings);
+        return List.of(storageService.get());
+    }
+
     @Override
     public List<Setting<?>> getSettings() {
         return Arrays.asList(
@@ -94,6 +102,6 @@ public void reload(Settings settings) {
         // `GoogleCloudStorageClientSettings` instance) instead of the `Settings`
         // instance.
         final Map<String, GoogleCloudStorageClientSettings> clientsSettings = GoogleCloudStorageClientSettings.load(settings);
-        this.storageService.refreshAndClearCache(clientsSettings);
+        this.storageService.get().refreshAndClearCache(clientsSettings);
     }
 }

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

@@ -139,6 +139,7 @@ private static Map<String, String> buildLocation(RepositoryMetadata metadata) {
     @Override
     protected GoogleCloudStorageBlobStore createBlobStore() {
         return new GoogleCloudStorageBlobStore(
+            getProjectId(),
             bucket,
             clientName,
             metadata.name(),