Merge remote-tracking branch 'upstream/feature/repository-s3-sdk-v2-upgrade' into 2025/03/18/repository-s3-sdk-v2

Author: DaveCTurner

modules/repository-s3/qa/web-identity-token/src/test/java/org/elasticsearch/repositories/s3/CustomWebIdentityTokenCredentialsProviderTests.java

@@ -227,34 +227,4 @@ public void testPickUpNewWebIdentityTokenWhenItsChanged() throws Exception {
             httpServer.stop(0);
         }
     }
-
-    public void testSupportRegionalizedEndpoints() throws Exception {
-        Map<String, String> environmentVariables = Map.of(
-            "AWS_WEB_IDENTITY_TOKEN_FILE",
-            "/var/run/secrets/eks.amazonaws.com/serviceaccount/token",
-            "AWS_ROLE_ARN",
-            ROLE_ARN,
-            "AWS_STS_REGIONAL_ENDPOINTS",
-            "regional",
-            "AWS_REGION",
-            "us-west-2"
-        );
-        Map<String, String> systemProperties = Map.of();
-
-        var webIdentityTokenCredentialsProvider = new S3Service.CustomWebIdentityTokenCredentialsProvider(
-            getEnvironment(),
-            environmentVariables::get,
-            systemProperties::getOrDefault,
-            Clock.systemUTC(),
-            resourceWatcherService
-        );
-        // We can't verify that webIdentityTokenCredentialsProvider's STS client uses the "https://sts.us-west-2.amazonaws.com"
-        // endpoint in a unit test. The client depends on hardcoded RegionalEndpointsOptionResolver that in turn depends
-        // on the system environment that we can't change in the test. So we just verify we that we called `withRegion`
-        // on stsClientBuilder which should internally correctly configure the endpoint when the STS client is built.
-        // TODO NOMERGE: can't access region anymore, need to rethink this.
-        // assertEquals("us-west-2", webIdentityTokenCredentialsProvider.getStsRegion());
-
-        webIdentityTokenCredentialsProvider.close();
-    }
 }

modules/repository-s3/src/internalClusterTest/java/org/elasticsearch/repositories/s3/S3BlobStoreRepositoryTests.java

@@ -505,7 +505,6 @@ public void testMultipartUploadCleanup() {
                 .bucket(blobStore.bucket())
                 .key(blobStore.blobContainer(repository.basePath().add("test-multipart-upload")).path().buildAsString() + danglingBlobName)
                 .overrideConfiguration(
-                    // NOMERGE: check this conversion makes sense.
                     AwsRequestOverrideConfiguration.builder()
                         .putRawQueryParameter(S3BlobStore.CUSTOM_QUERY_PARAMETER_PURPOSE, OperationPurpose.SNAPSHOT_DATA.getKey())
                         .build()
@@ -518,7 +517,6 @@ public void testMultipartUploadCleanup() {
                 .bucket(blobStore.bucket())
                 .prefix(repository.basePath().buildAsString())
                 .overrideConfiguration(
-                    // NOMERGE: check this conversion makes sense.
                     AwsRequestOverrideConfiguration.builder()
                         .putRawQueryParameter(S3BlobStore.CUSTOM_QUERY_PARAMETER_PURPOSE, OperationPurpose.SNAPSHOT_DATA.getKey())
                         .build()

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

@@ -234,7 +234,6 @@ S3Client buildClient(final S3ClientSettings clientSettings, SdkHttpClient httpCl
     }
 
     protected S3ClientBuilder buildClientBuilder(S3ClientSettings clientSettings, SdkHttpClient httpClient) {
-        // TODO NOMERGE ensure this has all the same config features as the v1 SDK
         var s3clientBuilder = S3Client.builder();
         s3clientBuilder.httpClient(httpClient);
         s3clientBuilder.overrideConfiguration(buildConfiguration(clientSettings, isStateless));
@@ -297,6 +296,7 @@ DnsResolver getCustomDnsResolver() {
     static SdkHttpClient buildHttpClient(S3ClientSettings clientSettings, @Nullable /* to use default resolver */ DnsResolver dnsResolver) {
         ApacheHttpClient.Builder httpClientBuilder = ApacheHttpClient.builder();
 
+        // TODO NOMERGE: an IT test for maxConnections?
         httpClientBuilder.maxConnections(clientSettings.maxConnections);
         httpClientBuilder.socketTimeout(Duration.ofMillis(clientSettings.readTimeoutMillis));
 
@@ -313,7 +313,7 @@ static SdkHttpClient buildHttpClient(S3ClientSettings clientSettings, @Nullable
     }
 
     // TODO NOMERGE naming
-    static boolean RETRYABLE_403_RETRY_PREDICATE(Throwable e) {
+    static boolean RETRYABLE_403_PREDICATE(Throwable e) {
         if (e instanceof AwsServiceException ase) {
             return ase.statusCode() == RestStatus.FORBIDDEN.getStatus() && "InvalidAccessKeyId".equals(ase.awsErrorDetails().errorCode());
         }
@@ -330,7 +330,7 @@ static ClientOverrideConfiguration buildConfiguration(S3ClientSettings clientSet
             // Create a 403 error retryable policy. In serverless we sometimes get 403s during because of delays in propagating updated
             // credentials because IAM is not strongly consistent.
             // TODO NOMERGE this should be covered by some end-to-end test, and documented more accurately
-            retryStrategyBuilder.retryOnException(S3Service::RETRYABLE_403_RETRY_PREDICATE);
+            retryStrategyBuilder.retryOnException(S3Service::RETRYABLE_403_PREDICATE);
         }
         clientOverrideConfiguration.retryStrategy(retryStrategyBuilder.build());
         return clientOverrideConfiguration.build();
@@ -480,8 +480,6 @@ public CompletableFuture<? extends AwsCredentialsIdentity> resolveIdentity() {
      */
     static class CustomWebIdentityTokenCredentialsProvider implements AwsCredentialsProvider {
 
-        private static final String STS_HOSTNAME = "https://sts.amazonaws.com";
-
         static final String WEB_IDENTITY_TOKEN_FILE_LOCATION = "repository-s3/aws-web-identity-token-file";
 
         private StsWebIdentityTokenFileCredentialsProvider credentialsProvider;
@@ -537,6 +535,8 @@ static class CustomWebIdentityTokenCredentialsProvider implements AwsCredentials
             );
 
             {
+                // TODO NOMERGE: is there any testing we need to add for this? We used to have a unit test that verified the regional stuff,
+                // but we're using this endpoint override instead of region now.
                 final var securityTokenServiceClientBuilder = StsClient.builder();
                 final var endpointOverride = jvmEnvironment.getProperty("org.elasticsearch.repositories.s3.stsEndpointOverride", null);
                 if (endpointOverride != null) {
@@ -694,17 +694,9 @@ public CompletableFuture<? extends AwsCredentialsIdentity> resolveIdentity(Consu
             return SocketAccess.doPrivileged(() -> delegate.resolveIdentity(consumer).handle(this::resultHandler));
         }
 
-        // TODO NOMERGE: I changed this so I could test successfully that a log message occurs.
-        // resultHandler doesn't appear to be invoked. I'm not sure how the original code works (haven't looked yet).
         @Override
         public CompletableFuture<? extends AwsCredentialsIdentity> resolveIdentity() {
-            try {
-                return SocketAccess.doPrivileged(() -> delegate.resolveIdentity());
-            } catch (Exception e) {
-                logger.error(() -> "Unable to resolve identity from " + delegate, e);
-                throw e;
-            }
-            // return SocketAccess.doPrivileged(() -> delegate.resolveIdentity().handle(this::resultHandler));
+            return SocketAccess.doPrivileged(() -> delegate.resolveIdentity().handle(this::resultHandler));
         }
 
         @Override

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

@@ -17,16 +17,21 @@
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
 import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
 import software.amazon.awssdk.identity.spi.AwsCredentialsIdentity;
+import software.amazon.awssdk.regions.Region;
 
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.util.Supplier;
+import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.common.settings.MockSecureSettings;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.env.Environment;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.watcher.ResourceWatcherService;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 import org.mockito.stubbing.Answer;
 
+import java.io.IOException;
 import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -36,6 +41,7 @@
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.startsWith;
+import static org.mockito.Mockito.mock;
 
 public class AwsS3ServiceImplTests extends ESTestCase {
 
@@ -173,7 +179,7 @@ public void testAWSDefaultConfiguration() {
         );
     }
 
-    public void testAWSConfigurationWithAwsSettings() {
+    public void testAwsConfigurationWithAwsSettings() {
         final MockSecureSettings secureSettings = new MockSecureSettings();
         secureSettings.setString("s3.client.default.proxy.username", "aws_proxy_username");
         secureSettings.setString("s3.client.default.proxy.password", "aws_proxy_password");
@@ -193,7 +199,6 @@ public void testRepositoryMaxRetries() {
         launchAWSConfigurationTest(settings, null, -1, null, null, null, 5, 50000);
     }
 
-    // TODO NOMERGE unused params
     private void launchAWSConfigurationTest(
         Settings settings,
         String expectedProxyHost,
@@ -218,9 +223,6 @@ private void launchAWSConfigurationTest(
 
         final ClientOverrideConfiguration configuration = S3Service.buildConfiguration(clientSettings, false);
         assertThat(configuration.retryStrategy().get().maxAttempts(), is(expectedMaxRetries + 1));
-
-        // TODO NOMERGE: consider whether this needs to be tested elsewhere.
-        // assertThat(configuration.getSocketTimeout(), is(expectedReadTimeout)); // set on the httpClient
     }
 
     public void testEndpointSetting() {
@@ -234,6 +236,28 @@ private void assertEndpoint(Settings repositorySettings, Settings settings, Stri
         assertThat(clientSettings.endpoint, is(expectedEndpoint));
     }
 
+    public void testEndPointAndRegionOverrides() throws IOException {
+        try (
+            S3Service s3Service = new S3Service(
+                mock(Environment.class),
+                Settings.EMPTY,
+                mock(ResourceWatcherService.class),
+                () -> Region.of("es-test-region")
+            )
+        ) {
+            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));
+
+            assertEquals(endpointOverride, reference.client().serviceClientConfiguration().endpointOverride().get().toString());
+            assertEquals("es-test-region", reference.client().serviceClientConfiguration().region().toString());
+
+            reference.close();
+            s3Service.doClose();
+        }
+    }
+
     public void testLoggingCredentialsProviderCatchesErrorsOnResolveCredentials() {
         var mockProvider = Mockito.mock(AwsCredentialsProvider.class);
         String mockProviderErrorMessage = "mockProvider failed to generate credentials";
@@ -253,17 +277,20 @@ public void testLoggingCredentialsProviderCatchesErrorsOnResolveCredentials() {
     }
 
     public void testLoggingCredentialsProviderCatchesErrorsOnResolveIdentity() {
-        // Set up #resolveIdentity() to throw a fake exception.
+        // Set up #resolveIdentity() to return a future with an exception.
         var mockCredentialsProvider = Mockito.mock(AwsCredentialsProvider.class);
         String mockProviderErrorMessage = "mockProvider failed to generate credentials";
-        Mockito.when(mockCredentialsProvider.resolveIdentity()).thenThrow(new IllegalStateException(mockProviderErrorMessage));
-
+        Answer<CompletableFuture<? extends AwsCredentialsIdentity>> answer = invocation -> {
+            CompletableFuture<AwsCredentialsIdentity> future = new CompletableFuture<>();
+            future.completeExceptionally(new IllegalStateException(mockProviderErrorMessage));
+            return future;
+        };
+        Mockito.when(mockCredentialsProvider.resolveIdentity()).thenAnswer(answer);
         var mockLogger = Mockito.mock(Logger.class);
         var credentialsProvider = new S3Service.ErrorLoggingCredentialsProvider(mockCredentialsProvider, mockLogger);
 
         // The S3Service.ErrorLoggingCredentialsProvider should log the error.
-        var exception = expectThrows(IllegalStateException.class, credentialsProvider::resolveIdentity);
-        assertEquals(mockProviderErrorMessage, exception.getMessage());
+        credentialsProvider.resolveIdentity();
 
         var messageSupplierCaptor = ArgumentCaptor.forClass(Supplier.class);
         var throwableCaptor = ArgumentCaptor.forClass(Throwable.class);

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

@@ -209,22 +209,5 @@ public void testMaxConnectionsCanBeSet() {
 
         // the default appears in the docs so let's make sure it doesn't change:
         assertEquals(50, S3ClientSettings.Defaults.MAX_CONNECTIONS);
-
-        // TODO NOMERGE: max connection setting is no longer available. Need alternative testing.
-        /*
-        SdkHttpClient defaultHttpClient = S3Service.buildHttpClient(settings.get("default"));
-        assertThat(defaultHttpClient.getMaxConnections(), is(S3ClientSettings.Defaults.MAX_CONNECTIONS));
-        SdkHttpClient otherHttpClient = S3Service.buildHttpClient(settings.get("other"));
-        assertThat(otherHttpClient.getMaxConnections(), is(maxConnections));
-        */
-    }
-
-    /*
-    // TODO NOMERGE: retryability is no longer testable via unit test: policy is not accessible. Need alternative testing.
-    public void testStatelessDefaultRetryPolicy() {
-        final var s3ClientSettings = S3ClientSettings.load(Settings.EMPTY).get("default");
-        final var clientConfiguration = S3Service.buildConfiguration(s3ClientSettings, true);
-        assertThat(clientConfiguration.getRetryPolicy(), is(S3Service.RETRYABLE_403_RETRY_POLICY));
     }
-    */
 }

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

@@ -31,7 +31,6 @@
 import java.io.IOException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import static org.hamcrest.CoreMatchers.equalTo;
 import static org.mockito.Mockito.mock;
 
 public class S3ServiceTests extends ESTestCase {
@@ -44,19 +43,13 @@ public void testCachedClientsAreReleased() throws IOException {
             () -> Region.of("es-test-region")
         );
         s3Service.start();
-        final String endpointOverride = "http://first";
-        final Settings settings = Settings.builder().put("endpoint", endpointOverride).build();
+        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);
         assertSame(clientSettings, otherClientSettings);
         final AmazonS3Reference reference = s3Service.client(metadata1);
-
-        // TODO NOMERGE: move to its own test.
-        assertEquals(endpointOverride, reference.client().serviceClientConfiguration().endpointOverride().get().toString());
-        assertEquals("es-test-region", reference.client().serviceClientConfiguration().region().toString());
-
         reference.close();
         s3Service.doClose();
         final AmazonS3Reference referenceReloaded = s3Service.client(metadata1);
@@ -83,28 +76,16 @@ public void testRetryOn403RetryPolicy() {
 
         // The retryable 403 condition retries on 403 invalid access key id
         assertTrue(
-            S3Service.RETRYABLE_403_RETRY_PREDICATE(
+            S3Service.RETRYABLE_403_PREDICATE(
                 RetryPolicyContext.builder().retriesAttempted(between(0, 9)).exception(s3Exception).build().exception()
             )
         );
 
-        if (randomBoolean()) {
-            // Random for another error status that is not 403
-            var non403StatusCode = randomValueOtherThan(403, () -> between(0, 600));
-            var non403Exception = S3Exception.builder().statusCode(non403StatusCode).awsErrorDetails(awsErrorDetails).build();
-            var retryPolicyContext = RetryPolicyContext.builder().retriesAttempted(between(0, 9)).exception(non403Exception).build();
-            // Retryable 403 condition delegates to the AWS default retry condition. Its result must be consistent with the decision
-            // by the AWS default, e.g. some error status like 429 is retryable by default, the retryable 403 condition respects it.
-            boolean actual = S3Service.RETRYABLE_403_RETRY_PREDICATE(retryPolicyContext.exception());
-            boolean expected = RetryCondition.defaultRetryCondition().shouldRetry(retryPolicyContext);
-            assertThat(actual, equalTo(expected));
-        } else {
-            // Not retry for 403 with error code that is not invalid access key id
-            String errorCode = randomAlphaOfLength(10);
-            var exception = S3Exception.builder().awsErrorDetails(AwsErrorDetails.builder().errorCode(errorCode).build()).build();
-            var retryPolicyContext = RetryPolicyContext.builder().retriesAttempted(between(0, 9)).exception(exception).build();
-            assertFalse(S3Service.RETRYABLE_403_RETRY_PREDICATE(retryPolicyContext.exception()));
-        }
+        // Not retry for 403 with error code that is not invalid access key id
+        String errorCode = randomAlphaOfLength(10);
+        var exception = S3Exception.builder().awsErrorDetails(AwsErrorDetails.builder().errorCode(errorCode).build()).build();
+        var retryPolicyContext = RetryPolicyContext.builder().retriesAttempted(between(0, 9)).exception(exception).build();
+        assertFalse(S3Service.RETRYABLE_403_PREDICATE(retryPolicyContext.exception()));
     }
 
     @TestLogging(reason = "testing WARN log output", value = "org.elasticsearch.repositories.s3.S3Service:WARN")