Support per-project client for Azure repository (#132701)

This PR adds per project client support for Azure repository similar to
#131899 (GCS) and #127631 (s3).

Resolves: ES-11714

Author: ywangd

modules/repository-azure/src/internalClusterTest/java/org/elasticsearch/repositories/azure/AzureBlobStoreRepositoryTests.java

@@ -17,6 +17,8 @@
 import com.sun.net.httpserver.HttpHandler;
 
 import org.elasticsearch.action.support.broadcast.BroadcastResponse;
+import org.elasticsearch.cluster.project.ProjectResolver;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Randomness;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.blobstore.BlobContainer;
@@ -172,8 +174,13 @@ public TestAzureRepositoryPlugin(Settings settings) {
         }
 
         @Override
-        AzureStorageService createAzureStorageService(Settings settingsToUse, AzureClientProvider azureClientProvider) {
-            return new AzureStorageService(settingsToUse, azureClientProvider) {
+        AzureStorageService createAzureStorageService(
+            Settings settingsToUse,
+            AzureClientProvider azureClientProvider,
+            ClusterService clusterService,
+            ProjectResolver projectResolver
+        ) {
+            return new AzureStorageService(settingsToUse, azureClientProvider, clusterService, projectResolver) {
                 @Override
                 RequestRetryOptions getRetryOptions(LocationMode locationMode, AzureStorageSettings azureStorageSettings) {
                     return new RequestRetryOptions(

modules/repository-azure/src/internalClusterTest/java/org/elasticsearch/repositories/azure/AzureStorageCleanupThirdPartyTests.java

@@ -21,6 +21,9 @@
 import org.elasticsearch.action.ActionRunnable;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.project.ProjectResolver;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.blobstore.BlobContainer;
@@ -76,9 +79,14 @@ public TestAzureRepositoryPlugin(Settings settings) {
         }
 
         @Override
-        AzureStorageService createAzureStorageService(Settings settings, AzureClientProvider azureClientProvider) {
+        AzureStorageService createAzureStorageService(
+            Settings settings,
+            AzureClientProvider azureClientProvider,
+            ClusterService clusterService,
+            ProjectResolver projectResolver
+        ) {
             final long blockSize = ByteSizeValue.ofKb(64L).getBytes() * randomIntBetween(1, 15);
-            return new AzureStorageService(settings, azureClientProvider) {
+            return new AzureStorageService(settings, azureClientProvider, clusterService, projectResolver) {
                 @Override
                 long getUploadBlockSize() {
                     return blockSize;
@@ -163,7 +171,7 @@ private void ensureSasTokenPermissions() {
         repository.threadPool().generic().execute(ActionRunnable.wrap(future, l -> {
             final AzureBlobStore blobStore = (AzureBlobStore) repository.blobStore();
             final AzureBlobServiceClient azureBlobServiceClient = blobStore.getService()
-                .client("default", LocationMode.PRIMARY_ONLY, randomFrom(OperationPurpose.values()));
+                .client(ProjectId.DEFAULT, "default", LocationMode.PRIMARY_ONLY, randomFrom(OperationPurpose.values()));
             final BlobServiceClient client = azureBlobServiceClient.getSyncClient();
             try {
                 final BlobContainerClient blobContainer = client.getBlobContainerClient(blobStore.toString());

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

@@ -49,6 +49,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.core.util.Throwables;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.blobstore.BlobContainer;
@@ -117,6 +118,8 @@ public class AzureBlobStore implements BlobStore {
     private static final long DEFAULT_READ_CHUNK_SIZE = ByteSizeValue.of(32, ByteSizeUnit.MB).getBytes();
     private static final int DEFAULT_UPLOAD_BUFFERS_SIZE = (int) ByteSizeValue.of(64, ByteSizeUnit.KB).getBytes();
 
+    @Nullable // for cluster level object store in MP
+    private final ProjectId projectId;
     private final AzureStorageService service;
     private final BigArrays bigArrays;
     private final RepositoryMetadata repositoryMetadata;
@@ -133,11 +136,13 @@ public class AzureBlobStore implements BlobStore {
     private final AzureClientProvider.RequestMetricsHandler requestMetricsHandler;
 
     public AzureBlobStore(
+        @Nullable ProjectId projectId,
         RepositoryMetadata metadata,
         AzureStorageService service,
         BigArrays bigArrays,
         RepositoriesMetrics repositoriesMetrics
     ) {
+        this.projectId = projectId;
         this.container = Repository.CONTAINER_SETTING.get(metadata.settings());
         this.clientName = Repository.CLIENT_NAME.get(metadata.settings());
         this.service = service;
@@ -355,7 +360,7 @@ public InputStream getInputStream(OperationPurpose purpose, String blob, long po
             totalSize = position + length;
         }
         BlobAsyncClient blobAsyncClient = asyncClient.getBlobContainerAsyncClient(container).getBlobAsyncClient(blob);
-        int maxReadRetries = service.getMaxReadRetries(clientName);
+        int maxReadRetries = service.getMaxReadRetries(projectId, clientName);
         try {
             return new AzureInputStream(
                 blobAsyncClient,
@@ -941,7 +946,7 @@ private BlobServiceAsyncClient asyncClient(OperationPurpose purpose) {
     }
 
     private AzureBlobServiceClient getAzureBlobServiceClientClient(OperationPurpose purpose) {
-        return service.client(clientName, locationMode, purpose, requestMetricsHandler);
+        return service.client(projectId, clientName, locationMode, purpose, requestMetricsHandler);
     }
 
     @Override

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

@@ -178,7 +178,7 @@ protected BlobStore getBlobStore() {
 
     @Override
     protected AzureBlobStore createBlobStore() {
-        final AzureBlobStore blobStore = new AzureBlobStore(metadata, storageService, bigArrays, repositoriesMetrics);
+        final AzureBlobStore blobStore = new AzureBlobStore(getProjectId(), metadata, storageService, bigArrays, repositoriesMetrics);
 
         logger.debug(
             () -> format(
@@ -204,6 +204,6 @@ public boolean isReadOnly() {
 
     @Override
     protected Set<String> getExtraUsageFeatures() {
-        return storageService.getExtraUsageFeatures(Repository.CLIENT_NAME.get(getMetadata().settings()));
+        return storageService.getExtraUsageFeatures(getProjectId(), Repository.CLIENT_NAME.get(getMetadata().settings()));
     }
 }

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

@@ -11,6 +11,7 @@
 
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
@@ -84,13 +85,20 @@ public Map<String, Repository.Factory> getRepositories(
     @Override
     public Collection<?> createComponents(PluginServices services) {
         AzureClientProvider azureClientProvider = AzureClientProvider.create(services.threadPool(), settings);
-        azureStoreService.set(createAzureStorageService(settings, azureClientProvider));
+        azureStoreService.set(
+            createAzureStorageService(settings, azureClientProvider, services.clusterService(), services.projectResolver())
+        );
         assert assertRepositoryAzureMaxThreads(settings, services.threadPool());
         return List.of(azureClientProvider);
     }
 
-    AzureStorageService createAzureStorageService(Settings settingsToUse, AzureClientProvider azureClientProvider) {
-        return new AzureStorageService(settingsToUse, azureClientProvider);
+    AzureStorageService createAzureStorageService(
+        Settings settingsToUse,
+        AzureClientProvider azureClientProvider,
+        ClusterService clusterService,
+        ProjectResolver projectResolver
+    ) {
+        return new AzureStorageService(settingsToUse, azureClientProvider, clusterService, projectResolver);
     }
 
     @Override
@@ -140,7 +148,7 @@ public void reload(Settings settingsToLoad) {
         final Map<String, AzureStorageSettings> clientsSettings = AzureStorageSettings.load(settingsToLoad);
         AzureStorageService storageService = azureStoreService.get();
         assert storageService != null;
-        storageService.refreshSettings(clientsSettings);
+        storageService.refreshClusterClientSettings(clientsSettings);
     }
 
     private static boolean assertRepositoryAzureMaxThreads(Settings settings, ThreadPool threadPool) {