Support per-project s3 clients (elastic#127631)

With project secrets in place, we can now create and manage per-project
repository clients in addition to the cluster level repository clients.
This PR adds a manager class for that. Note that the logic is not yet
fully wired because it needs per-project repository/objec_store which
will be added in separate PRs (as part of MP snapshots and MP
objecStoreService). As such the tests are currently unit tests.

Relates: elastic#126584 Resolves:
ES-11713

Author: ywangd

modules/repository-s3/qa/insecure-credentials/src/test/java/org/elasticsearch/repositories/s3/RepositoryCredentialsTests.java

@@ -17,6 +17,8 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.cluster.project.ProjectResolver;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.MockSecureSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsFilter;
@@ -256,8 +258,13 @@ public ProxyS3RepositoryPlugin(Settings settings) {
         }
 
         @Override
-        S3Service s3Service(Environment environment, Settings nodeSettings, ResourceWatcherService resourceWatcherService) {
-            return new ProxyS3Service(environment, nodeSettings, resourceWatcherService);
+        S3Service s3Service(
+            Environment environment,
+            ClusterService clusterService,
+            ProjectResolver projectResolver,
+            ResourceWatcherService resourceWatcherService
+        ) {
+            return new ProxyS3Service(environment, clusterService, projectResolver, resourceWatcherService);
         }
 
         /**
@@ -293,8 +300,13 @@ public static final class ProxyS3Service extends S3Service {
 
             private static final Logger logger = LogManager.getLogger(ProxyS3Service.class);
 
-            ProxyS3Service(Environment environment, Settings nodeSettings, ResourceWatcherService resourceWatcherService) {
-                super(environment, nodeSettings, resourceWatcherService, () -> null);
+            ProxyS3Service(
+                Environment environment,
+                ClusterService clusterService,
+                ProjectResolver projectResolver,
+                ResourceWatcherService resourceWatcherService
+            ) {
+                super(environment, clusterService, projectResolver, resourceWatcherService, () -> null);
             }
 
             @Override

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

@@ -15,6 +15,7 @@
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
@@ -80,13 +81,20 @@ protected S3Repository createRepository(
 
     @Override
     public Collection<?> createComponents(PluginServices services) {
-        service.set(s3Service(services.environment(), services.clusterService().getSettings(), services.resourceWatcherService()));
+        service.set(
+            s3Service(services.environment(), services.clusterService(), services.projectResolver(), services.resourceWatcherService())
+        );
         this.service.get().refreshAndClearCache(S3ClientSettings.load(settings));
         return List.of(service.get());
     }
 
-    S3Service s3Service(Environment environment, Settings nodeSettings, ResourceWatcherService resourceWatcherService) {
-        return new S3Service(environment, nodeSettings, resourceWatcherService, S3RepositoryPlugin::getDefaultRegion);
+    S3Service s3Service(
+        Environment environment,
+        ClusterService clusterService,
+        ProjectResolver projectResolver,
+        ResourceWatcherService resourceWatcherService
+    ) {
+        return new S3Service(environment, clusterService, projectResolver, resourceWatcherService, S3RepositoryPlugin::getDefaultRegion);
     }
 
     private static Region getDefaultRegion() {

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

@@ -37,17 +37,19 @@
 import org.apache.http.conn.DnsResolver;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.lucene.store.AlreadyClosedException;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.cluster.coordination.stateless.StoreHeartbeatService;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.project.ProjectResolver;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.util.Maps;
 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;
@@ -72,7 +74,6 @@
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
-import static java.util.Collections.emptyMap;
 import static software.amazon.awssdk.core.SdkSystemSetting.AWS_ROLE_ARN;
 import static software.amazon.awssdk.core.SdkSystemSetting.AWS_ROLE_SESSION_NAME;
 import static software.amazon.awssdk.core.SdkSystemSetting.AWS_WEB_IDENTITY_TOKEN_FILE;
@@ -93,21 +94,6 @@ class S3Service extends AbstractLifecycleComponent {
         TimeValue.timeValueHours(24),
         Setting.Property.NodeScope
     );
-    private volatile Map<S3ClientSettings, AmazonS3Reference> clientsCache = emptyMap();
-
-    /**
-     * Client settings calculated from static configuration and settings in the keystore.
-     */
-    private volatile Map<String, S3ClientSettings> staticClientSettings = Map.of(
-        "default",
-        S3ClientSettings.getClientSettings(Settings.EMPTY, "default")
-    );
-
-    /**
-     * Client settings derived from those in {@link #staticClientSettings} by combining them with settings
-     * in the {@link RepositoryMetadata}.
-     */
-    private volatile Map<Settings, S3ClientSettings> derivedClientSettings = emptyMap();
 
     private final Runnable defaultRegionSetter;
     private volatile Region defaultRegion;
@@ -124,13 +110,16 @@ class S3Service extends AbstractLifecycleComponent {
     final TimeValue compareAndExchangeTimeToLive;
     final TimeValue compareAndExchangeAntiContentionDelay;
     final boolean isStateless;
+    private final S3ClientsManager s3ClientsManager;
 
     S3Service(
         Environment environment,
-        Settings nodeSettings,
+        ClusterService clusterService,
+        ProjectResolver projectResolver,
         ResourceWatcherService resourceWatcherService,
         Supplier<Region> defaultRegionSupplier
     ) {
+        final Settings nodeSettings = clusterService.getSettings();
         webIdentityTokenCredentialsProvider = new CustomWebIdentityTokenCredentialsProvider(
             environment,
             System::getenv,
@@ -142,6 +131,20 @@ class S3Service extends AbstractLifecycleComponent {
         compareAndExchangeAntiContentionDelay = REPOSITORY_S3_CAS_ANTI_CONTENTION_DELAY_SETTING.get(nodeSettings);
         isStateless = DiscoveryNode.isStateless(nodeSettings);
         defaultRegionSetter = new RunOnce(() -> defaultRegion = defaultRegionSupplier.get());
+        s3ClientsManager = new S3ClientsManager(
+            nodeSettings,
+            this::buildClientReference,
+            clusterService.threadPool().generic(),
+            projectResolver.supportsMultipleProjects()
+        );
+        if (projectResolver.supportsMultipleProjects()) {
+            clusterService.addHighPriorityApplier(s3ClientsManager);
+        }
+    }
+
+    // visible to tests
+    S3ClientsManager getS3ClientsManager() {
+        return s3ClientsManager;
     }
 
     /**
@@ -151,86 +154,55 @@ class S3Service extends AbstractLifecycleComponent {
      * of being returned to the cache.
      */
     public synchronized void refreshAndClearCache(Map<String, S3ClientSettings> clientsSettings) {
-        // shutdown all unused clients
-        // others will shutdown on their respective release
-        releaseCachedClients();
-        this.staticClientSettings = Maps.ofEntries(clientsSettings.entrySet());
-        derivedClientSettings = emptyMap();
-        assert this.staticClientSettings.containsKey("default") : "always at least have 'default'";
-        /* clients are built lazily by {@link #client} */
+        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) {
-        final S3ClientSettings clientSettings = settings(repositoryMetadata);
-        {
-            final AmazonS3Reference clientReference = clientsCache.get(clientSettings);
-            if (clientReference != null && clientReference.tryIncRef()) {
-                return clientReference;
-            }
-        }
-        synchronized (this) {
-            final AmazonS3Reference existing = clientsCache.get(clientSettings);
-            if (existing != null && existing.tryIncRef()) {
-                return existing;
-            }
-
-            if (lifecycle.started() == false) {
-                // doClose() calls releaseCachedClients() which is also synchronized (this) so if we're STARTED here then the client we
-                // create will definitely not leak on close.
-                throw new AlreadyClosedException("S3Service is in state [" + lifecycle + "]");
-            }
+        return client(ProjectId.DEFAULT, repositoryMetadata);
+    }
 
-            final SdkHttpClient httpClient = buildHttpClient(clientSettings, getCustomDnsResolver());
-            Releasable toRelease = httpClient::close;
-            try {
-                final AmazonS3Reference clientReference = new AmazonS3Reference(buildClient(clientSettings, httpClient), httpClient);
-                clientReference.mustIncRef();
-                clientsCache = Maps.copyMapWithAddedEntry(clientsCache, clientSettings, clientReference);
-                toRelease = null;
-                return clientReference;
-            } finally {
-                Releasables.close(toRelease);
-            }
-        }
+    /**
+     * 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.
+     * @param projectId The project associated with the client, or null if the client is cluster level
+     */
+    public AmazonS3Reference client(@Nullable ProjectId projectId, RepositoryMetadata repositoryMetadata) {
+        return s3ClientsManager.client(effectiveProjectId(projectId), repositoryMetadata);
     }
 
     /**
-     * Either fetches {@link S3ClientSettings} for a given {@link RepositoryMetadata} from cached settings or creates them
-     * by overriding static client settings from {@link #staticClientSettings} with settings found in the repository metadata.
-     * @param repositoryMetadata Repository Metadata
-     * @return S3ClientSettings
+     * We use the default project-id for cluster level clients.
      */
-    S3ClientSettings settings(RepositoryMetadata repositoryMetadata) {
-        final Settings settings = repositoryMetadata.settings();
-        {
-            final S3ClientSettings existing = derivedClientSettings.get(settings);
-            if (existing != null) {
-                return existing;
-            }
-        }
-        final String clientName = S3Repository.CLIENT_NAME.get(settings);
-        final S3ClientSettings staticSettings = staticClientSettings.get(clientName);
-        if (staticSettings != null) {
-            synchronized (this) {
-                final S3ClientSettings existing = derivedClientSettings.get(settings);
-                if (existing != null) {
-                    return existing;
-                }
-                final S3ClientSettings newSettings = staticSettings.refine(settings);
-                derivedClientSettings = Maps.copyMapWithAddedOrReplacedEntry(derivedClientSettings, settings, newSettings);
-                return newSettings;
-            }
+    ProjectId effectiveProjectId(@Nullable ProjectId projectId) {
+        return projectId == null ? ProjectId.DEFAULT : projectId;
+    }
+
+    // TODO: consider moving client building into S3ClientsManager
+    private AmazonS3Reference buildClientReference(final S3ClientSettings clientSettings) {
+        final SdkHttpClient httpClient = buildHttpClient(clientSettings, getCustomDnsResolver());
+        Releasable toRelease = httpClient::close;
+        try {
+            final AmazonS3Reference clientReference = new AmazonS3Reference(buildClient(clientSettings, httpClient), httpClient);
+            clientReference.mustIncRef();
+            toRelease = null;
+            return clientReference;
+        } finally {
+            Releasables.close(toRelease);
         }
-        throw new IllegalArgumentException(
-            "Unknown s3 client name ["
-                + clientName
-                + "]. Existing client configs: "
-                + Strings.collectionToDelimitedString(staticClientSettings.keySet(), ",")
-        );
+    }
+
+    @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);
     }
 
     // proxy for testing
@@ -448,18 +420,17 @@ static AwsCredentialsProvider buildCredentials(
         }
     }
 
-    private synchronized void releaseCachedClients() {
-        // the clients will shutdown when they will not be used anymore
-        for (final AmazonS3Reference clientReference : clientsCache.values()) {
-            clientReference.decRef();
-        }
-        // clear previously cached clients, they will be build lazily
-        clientsCache = emptyMap();
-        derivedClientSettings = emptyMap();
+    @FixForMultiProject(description = "can be removed once blobstore is project aware")
+    public void onBlobStoreClose() {
+        onBlobStoreClose(ProjectId.DEFAULT);
     }
 
-    public void onBlobStoreClose() {
-        releaseCachedClients();
+    /**
+     * Release clients for the specified project.
+     * @param projectId The project associated with the client, or null if the client is cluster level
+     */
+    public void onBlobStoreClose(@Nullable ProjectId projectId) {
+        s3ClientsManager.releaseCachedClients(effectiveProjectId(projectId));
     }
 
     @Override
@@ -472,7 +443,7 @@ protected void doStop() {}
 
     @Override
     public void doClose() throws IOException {
-        releaseCachedClients();
+        s3ClientsManager.close();
         webIdentityTokenCredentialsProvider.close();
     }
 

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

@@ -22,9 +22,12 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.util.Supplier;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
+import org.elasticsearch.cluster.project.TestProjectResolvers;
 import org.elasticsearch.common.settings.MockSecureSettings;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.DeterministicTaskQueue;
 import org.elasticsearch.env.Environment;
+import org.elasticsearch.test.ClusterServiceUtils;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.watcher.ResourceWatcherService;
 import org.mockito.ArgumentCaptor;
@@ -234,7 +237,8 @@ public void testEndPointAndRegionOverrides() throws IOException {
         try (
             S3Service s3Service = new S3Service(
                 mock(Environment.class),
-                Settings.EMPTY,
+                ClusterServiceUtils.createClusterService(new DeterministicTaskQueue().getThreadPool()),
+                TestProjectResolvers.DEFAULT_PROJECT_ONLY,
                 mock(ResourceWatcherService.class),
                 () -> Region.of("es-test-region")
             )
@@ -248,7 +252,6 @@ public void testEndPointAndRegionOverrides() throws IOException {
             assertEquals("es-test-region", reference.client().serviceClientConfiguration().region().toString());
 
             reference.close();
-            s3Service.doClose();
         }
     }
 

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

@@ -22,6 +22,7 @@
 import org.apache.logging.log4j.Level;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
+import org.elasticsearch.cluster.project.TestProjectResolvers;
 import org.elasticsearch.common.BackoffPolicy;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.blobstore.BlobContainer;
@@ -54,6 +55,7 @@
 import org.elasticsearch.telemetry.InstrumentType;
 import org.elasticsearch.telemetry.Measurement;
 import org.elasticsearch.telemetry.RecordingMeterRegistry;
+import org.elasticsearch.test.ClusterServiceUtils;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.MockLog;
 import org.elasticsearch.watcher.ResourceWatcherService;
@@ -126,7 +128,13 @@ public class S3BlobContainerRetriesTests extends AbstractBlobContainerRetriesTes
     @Before
     public void setUp() throws Exception {
         shouldErrorOnDns = false;
-        service = new S3Service(Mockito.mock(Environment.class), Settings.EMPTY, Mockito.mock(ResourceWatcherService.class), () -> null) {
+        service = new S3Service(
+            Mockito.mock(Environment.class),
+            ClusterServiceUtils.createClusterService(new DeterministicTaskQueue().getThreadPool()),
+            TestProjectResolvers.DEFAULT_PROJECT_ONLY,
+            Mockito.mock(ResourceWatcherService.class),
+            () -> null
+        ) {
             private InetAddress[] resolveHost(String host) throws UnknownHostException {
                 assertEquals("127.0.0.1", host);
                 if (shouldErrorOnDns && randomBoolean() && randomBoolean()) {
@@ -1308,7 +1316,11 @@ public void testRetryOn403InStateless() {
 
         service = new S3Service(
             Mockito.mock(Environment.class),
-            Settings.builder().put(STATELESS_ENABLED_SETTING_NAME, "true").build(),
+            ClusterServiceUtils.createClusterService(
+                new DeterministicTaskQueue().getThreadPool(),
+                Settings.builder().put(STATELESS_ENABLED_SETTING_NAME, "true").build()
+            ),
+            TestProjectResolvers.DEFAULT_PROJECT_ONLY,
             Mockito.mock(ResourceWatcherService.class),
             () -> null
         );