feat: Support customizing S3 endpoints (apache#1913) (apache#93)

No functional change.

Introduce a dedicated interface for `StsClient` suppliers and implement it using a pool of cached clients.

All client are "thin" and share the same `SdkHttpClient`. The latter is closed when the server shuts down.

This is a step towards supporting non-AWS S3 storage (apache#1530).
For this reason the STS endpoint is present in new interfaces, but is not used yet.

Author: dimas-b

CHANGELOG.md

@@ -35,6 +35,8 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 
 ### New Features
 
+- Added Catalog configuration for S3 and STS endpoints. This also allows using non-AWS S3 implementations.
+
 ### Changes
 
 ### Deprecations

LICENSE

@@ -319,7 +319,12 @@ This product includes code from Project Nessie.
 * tools/config-docs/generator/src/test/java/tests/smallrye/SomeEnum.java
 * tools/config-docs/generator/src/test/java/tests/smallrye/VeryNested.java
 * tools/container-spec-helper/src/main/java/org/apache/polaris/containerspec/ContainerSpecHelper.java
+* tools/minio-testcontainer/src/main/java/org/apache/polaris/test/minio/Minio.java
+* tools/minio-testcontainer/src/main/java/org/apache/polaris/test/minio/MinioAccess.java
+* tools/minio-testcontainer/src/main/java/org/apache/polaris/test/minio/MinioContainer.java
+* tools/minio-testcontainer/src/main/java/org/apache/polaris/test/minio/MinioExtension.java
 * runtime/admin/src/main/java/org/apache/polaris/admintool/PolarisAdminTool.java
+* runtime/service/src/main/java/org/apache/polaris/service/storage/aws/StsClientsPool.java
 * helm/polaris/tests/logging_storage_test.yaml
 * helm/polaris/tests/quantity_test.yaml
 * helm/polaris/tests/service_monitor_test.yaml

bom/build.gradle.kts

@@ -30,6 +30,7 @@ dependencies {
     api(project(":polaris-api-management-service"))
 
     api(project(":polaris-container-spec-helper"))
+    api(project(":polaris-minio-testcontainer"))
     api(project(":polaris-immutables"))
     api(project(":polaris-misc-types"))
     api(project(":polaris-version"))

gradle/projects.main.properties

@@ -39,6 +39,7 @@ polaris-tests=integration-tests
 aggregated-license-report=aggregated-license-report
 polaris-immutables=tools/immutables
 polaris-container-spec-helper=tools/container-spec-helper
+polaris-minio-testcontainer=tools/minio-testcontainer
 polaris-version=tools/version
 polaris-misc-types=tools/misc-types
 

polaris-core/src/main/java/org/apache/polaris/core/entity/CatalogEntity.java

@@ -273,7 +273,9 @@ public Builder setStorageConfigurationInfo(
                     new ArrayList<>(allowedLocations),
                     awsConfigModel.getRoleArn(),
                     awsConfigModel.getExternalId(),
-                    awsConfigModel.getRegion());
+                    awsConfigModel.getRegion(),
+                    awsConfigModel.getEndpoint(),
+                    awsConfigModel.getStsEndpoint());
             awsConfig.validateArn(awsConfigModel.getRoleArn());
             config = awsConfig;
             break;

polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsCredentialsStorageIntegration.java

@@ -32,6 +32,7 @@
 import org.apache.polaris.core.storage.InMemoryStorageIntegration;
 import org.apache.polaris.core.storage.StorageAccessProperty;
 import org.apache.polaris.core.storage.StorageUtil;
+import org.apache.polaris.core.storage.aws.StsClientProvider.StsDestination;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.policybuilder.iam.IamConditionOperator;
 import software.amazon.awssdk.policybuilder.iam.IamEffect;
@@ -45,28 +46,22 @@
 /** Credential vendor that supports generating */
 public class AwsCredentialsStorageIntegration
     extends InMemoryStorageIntegration<AwsStorageConfigurationInfo> {
-  private final StsClient stsClient;
+  private final StsClientProvider stsClientProvider;
   private final Optional<AwsCredentialsProvider> credentialsProvider;
-  private final Map<StorageAccessProperty, String> awsCredentialsAdditionalProperties =
-      new HashMap<>();
 
-  public AwsCredentialsStorageIntegration(StsClient stsClient) {
-    this(stsClient, Optional.empty());
+  public AwsCredentialsStorageIntegration(StsClient fixedClient) {
+    this((destination) -> fixedClient);
   }
 
-  public AwsCredentialsStorageIntegration(
-      StsClient stsClient, Optional<AwsCredentialsProvider> credentialsProvider) {
-    this(stsClient, credentialsProvider, Map.of());
+  public AwsCredentialsStorageIntegration(StsClientProvider stsClientProvider) {
+    this(stsClientProvider, Optional.empty());
   }
 
   public AwsCredentialsStorageIntegration(
-      StsClient stsClient,
-      Optional<AwsCredentialsProvider> credentialsProvider,
-      Map<StorageAccessProperty, String> awsCredentialsAdditionalProperties) {
+      StsClientProvider stsClientProvider, Optional<AwsCredentialsProvider> credentialsProvider) {
     super(AwsCredentialsStorageIntegration.class.getName());
-    this.stsClient = stsClient;
+    this.stsClientProvider = stsClientProvider;
     this.credentialsProvider = credentialsProvider;
-    this.awsCredentialsAdditionalProperties.putAll(awsCredentialsAdditionalProperties);
   }
 
   /** {@inheritDoc} */
@@ -77,29 +72,36 @@ public EnumMap<StorageAccessProperty, String> getSubscopedCreds(
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations) {
+    int storageCredentialDurationSeconds =
+        callContext
+            .getPolarisCallContext()
+            .getConfigurationStore()
+            .getConfiguration(callContext.getRealmContext(), STORAGE_CREDENTIAL_DURATION_SECONDS);
+    AssumeRoleRequest.Builder request =
+        AssumeRoleRequest.builder()
+            .externalId(storageConfig.getExternalId())
+            .roleArn(storageConfig.getRoleARN())
+            .roleSessionName("PolarisAwsCredentialsStorageIntegration")
+            .policy(
+                policyString(
+                        storageConfig.getRoleARN(),
+                        allowListOperation,
+                        allowedReadLocations,
+                        allowedWriteLocations)
+                    .toJson())
+            .durationSeconds(storageCredentialDurationSeconds);
+    credentialsProvider.ifPresent(
+        cp -> request.overrideConfiguration(b -> b.credentialsProvider(cp)));
+
+    @SuppressWarnings("resource")
+    // Note: stsClientProvider returns "thin" clients that do not need closing
+    StsClient stsClient =
+        stsClientProvider.stsClient(
+            StsDestination.of(storageConfig.getStsEndpointUri(), storageConfig.getRegion()));
+
     EnumMap<StorageAccessProperty, String> credentialMap =
         new EnumMap<>(StorageAccessProperty.class);
     if (stsClient != null) {
-      int storageCredentialDurationSeconds =
-          callContext
-              .getPolarisCallContext()
-              .getConfigurationStore()
-              .getConfiguration(callContext.getRealmContext(), STORAGE_CREDENTIAL_DURATION_SECONDS);
-      AssumeRoleRequest.Builder request =
-          AssumeRoleRequest.builder()
-              .externalId(storageConfig.getExternalId())
-              .roleArn(storageConfig.getRoleARN())
-              .roleSessionName("PolarisAwsCredentialsStorageIntegration")
-              .policy(
-                  policyString(
-                          storageConfig.getRoleARN(),
-                          allowListOperation,
-                          allowedReadLocations,
-                          allowedWriteLocations)
-                      .toJson())
-              .durationSeconds(storageCredentialDurationSeconds);
-      credentialsProvider.ifPresent(
-          cp -> request.overrideConfiguration(b -> b.credentialsProvider(cp)));
       AssumeRoleResponse response = stsClient.assumeRole(request.build());
       credentialMap.put(StorageAccessProperty.AWS_KEY_ID, response.credentials().accessKeyId());
       credentialMap.put(
@@ -120,17 +122,18 @@ public EnumMap<StorageAccessProperty, String> getSubscopedCreds(
       credentialMap.put(StorageAccessProperty.CLIENT_REGION, storageConfig.getRegion());
     }
 
+    URI endpointUri = storageConfig.getEndpointUri();
+    if (endpointUri != null) {
+      credentialMap.put(StorageAccessProperty.AWS_ENDPOINT, endpointUri.toString());
+    }
+
     if (storageConfig.getAwsPartition().equals("aws-us-gov")
         && credentialMap.get(StorageAccessProperty.CLIENT_REGION) == null) {
       throw new IllegalArgumentException(
           String.format(
               "AWS region must be set when using partition %s", storageConfig.getAwsPartition()));
     }
 
-    if (!awsCredentialsAdditionalProperties.isEmpty()) {
-      awsCredentialsAdditionalProperties.forEach(credentialMap::putIfAbsent);
-    }
-
     return credentialMap;
   }
 

polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfo.java

@@ -24,6 +24,7 @@
 import com.google.common.base.MoreObjects;
 import jakarta.annotation.Nonnull;
 import jakarta.annotation.Nullable;
+import java.net.URI;
 import java.util.List;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -53,14 +54,38 @@ public class AwsStorageConfigurationInfo extends PolarisStorageConfigurationInfo
   @JsonProperty(value = "region")
   private @Nullable String region = null;
 
+  /** Endpoint URI for S3 API calls */
+  @JsonProperty(value = "endpoint")
+  private @Nullable String endpoint;
+
+  /** Endpoint URI for STS API calls */
+  @JsonProperty(value = "stsEndpoint")
+  private @Nullable String stsEndpoint;
+
   @JsonCreator
   public AwsStorageConfigurationInfo(
       @JsonProperty(value = "storageType", required = true) @Nonnull StorageType storageType,
       @JsonProperty(value = "allowedLocations", required = true) @Nonnull
           List<String> allowedLocations,
       @JsonProperty(value = "roleARN", required = true) @Nonnull String roleARN,
-      @JsonProperty(value = "region", required = false) @Nullable String region) {
-    this(storageType, allowedLocations, roleARN, null, region);
+      @JsonProperty(value = "externalId") @Nullable String externalId,
+      @JsonProperty(value = "region", required = false) @Nullable String region,
+      @JsonProperty(value = "endpoint") @Nullable String endpoint,
+      @JsonProperty(value = "stsEndpoint") @Nullable String stsEndpoint) {
+    super(storageType, allowedLocations);
+    this.roleARN = roleARN;
+    this.externalId = externalId;
+    this.region = region;
+    this.endpoint = endpoint;
+    this.stsEndpoint = stsEndpoint;
+  }
+
+  public AwsStorageConfigurationInfo(
+      @Nonnull StorageType storageType,
+      @Nonnull List<String> allowedLocations,
+      @Nonnull String roleARN,
+      @Nullable String region) {
+    this(storageType, allowedLocations, roleARN, null, region, null, null);
   }
 
   public AwsStorageConfigurationInfo(
@@ -69,10 +94,7 @@ public AwsStorageConfigurationInfo(
       @Nonnull String roleARN,
       @Nullable String externalId,
       @Nullable String region) {
-    super(storageType, allowedLocations);
-    this.roleARN = roleARN;
-    this.externalId = externalId;
-    this.region = region;
+    this(storageType, allowedLocations, roleARN, externalId, region, null, null);
   }
 
   @Override
@@ -121,6 +143,19 @@ public void setRegion(@Nullable String region) {
     this.region = region;
   }
 
+  @JsonIgnore
+  @Nullable
+  public URI getEndpointUri() {
+    return endpoint == null ? null : URI.create(endpoint);
+  }
+
+  /** Returns the STS endpoint if set, defaulting to {@link #getEndpointUri()} otherwise. */
+  @JsonIgnore
+  @Nullable
+  public URI getStsEndpointUri() {
+    return stsEndpoint == null ? getEndpointUri() : URI.create(stsEndpoint);
+  }
+
   @JsonIgnore
   public String getAwsAccountId() {
     return parseAwsAccountId(roleARN);

polaris-core/src/main/java/org/apache/polaris/core/storage/aws/StsClientProvider.java

@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.polaris.core.storage.aws;
+
+import jakarta.annotation.Nullable;
+import java.net.URI;
+import java.util.Optional;
+import org.apache.polaris.immutables.PolarisImmutable;
+import org.immutables.value.Value;
+import software.amazon.awssdk.awscore.client.builder.AwsClientBuilder;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.sts.StsBaseClientBuilder;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.endpoints.StsEndpointProvider;
+
+public interface StsClientProvider {
+
+  /**
+   * Returns an STS client for the given destination (endpoint + region). The returned client may
+   * not be a fresh instance for every call, however the client is reusable for multiple concurrent
+   * requests from multiple threads. If the endpoint or region parameters are not specified, AWS SDK
+   * defaults will be used.
+   *
+   * @param destination Endpoint and Region data for the client. Both values are optional.
+   */
+  StsClient stsClient(StsDestination destination);
+
+  @PolarisImmutable
+  interface StsDestination {
+    /** Corresponds to {@link StsBaseClientBuilder#endpointProvider(StsEndpointProvider)} */
+    @Value.Parameter(order = 1)
+    Optional<URI> endpoint();
+
+    /** Corresponds to {@link AwsClientBuilder#region(Region)} */
+    @Value.Parameter(order = 2)
+    Optional<String> region();
+
+    static StsDestination of(@Nullable URI endpoint, @Nullable String region) {
+      return ImmutableStsDestination.of(Optional.ofNullable(endpoint), Optional.ofNullable(region));
+    }
+  }
+}

polaris-core/src/test/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfoTest.java

@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.polaris.core.storage.aws;
+
+import static org.apache.polaris.core.storage.PolarisStorageConfigurationInfo.StorageType.S3;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.net.URI;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+public class AwsStorageConfigurationInfoTest {
+
+  private static AwsStorageConfigurationInfo config(String endpoint, String stsEndpoint) {
+    return new AwsStorageConfigurationInfo(
+        S3, List.of(), "role", null, null, endpoint, stsEndpoint);
+  }
+
+  @Test
+  public void testStsEndpoint() {
+    assertThat(config(null, null))
+        .extracting(
+            AwsStorageConfigurationInfo::getEndpointUri,
+            AwsStorageConfigurationInfo::getStsEndpointUri)
+        .containsExactly(null, null);
+    assertThat(config(null, "http://sts.example.com"))
+        .extracting(
+            AwsStorageConfigurationInfo::getEndpointUri,
+            AwsStorageConfigurationInfo::getStsEndpointUri)
+        .containsExactly(null, URI.create("http://sts.example.com"));
+    assertThat(config("http://s3.example.com", null))
+        .extracting(
+            AwsStorageConfigurationInfo::getEndpointUri,
+            AwsStorageConfigurationInfo::getStsEndpointUri)
+        .containsExactly(URI.create("http://s3.example.com"), URI.create("http://s3.example.com"));
+    assertThat(config("http://s3.example.com", "http://sts.example.com"))
+        .extracting(
+            AwsStorageConfigurationInfo::getEndpointUri,
+            AwsStorageConfigurationInfo::getStsEndpointUri)
+        .containsExactly(URI.create("http://s3.example.com"), URI.create("http://sts.example.com"));
+  }
+}

runtime/service/build.gradle.kts

@@ -84,6 +84,9 @@ dependencies {
   implementation("software.amazon.awssdk:sts")
   implementation("software.amazon.awssdk:iam-policy-builder")
   implementation("software.amazon.awssdk:s3")
+  implementation("software.amazon.awssdk:apache-client") {
+    exclude("commons-logging", "commons-logging")
+  }
   implementation(platform(libs.azuresdk.bom))
   implementation("com.azure:azure-core")
 
@@ -106,6 +109,8 @@ dependencies {
   testImplementation(project(":polaris-api-management-model"))
   testImplementation(testFixtures(project(":polaris-service-common")))
 
+  testImplementation(project(":polaris-minio-testcontainer"))
+
   testImplementation("org.apache.iceberg:iceberg-api:${libs.versions.iceberg.get()}:tests")
   testImplementation("org.apache.iceberg:iceberg-core:${libs.versions.iceberg.get()}:tests")