Feature / S3 storage custom endpoints (#244)

* Allow setting endpoint for S3 storage in the data service

* Make region and endpoint optional in platform configuration for S3 endpoints

* Add "endpoint" option for S3 storage in the runtime

tracdap-plugins/aws-storage/src/main/java/org/finos/tracdap/plugins/aws/storage/S3ObjectStorage.java

@@ -39,6 +39,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.net.URI;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Properties;
@@ -53,9 +54,10 @@
 
 public class S3ObjectStorage implements IFileStorage {
 
-    public static final String REGION_PROPERTY = "region";
     public static final String BUCKET_PROPERTY = "bucket";
     public static final String PREFIX_PROPERTY = "prefix";
+    public static final String REGION_PROPERTY = "region";
+    public static final String ENDPOINT_PROPERTY = "endpoint";
 
     public static final String CREDENTIALS_PROPERTY = "credentials";
     public static final String CREDENTIALS_DEFAULT = "default";
@@ -66,10 +68,11 @@ public class S3ObjectStorage implements IFileStorage {
     private final Logger log = LoggerFactory.getLogger(getClass());
 
     private final String storageKey;
-    private final Region region;
+    private final AwsCredentialsProvider credentials;
     private final String bucket;
     private final StoragePath prefix;
-    private final AwsCredentialsProvider credentials;
+    private final Region region;
+    private final URI endpoint;
 
     private final StorageErrors errors;
 
@@ -80,13 +83,15 @@ public S3ObjectStorage(Properties properties) {
 
         this.storageKey = properties.getProperty(IStorageManager.PROP_STORAGE_KEY);
 
-        var region = properties.getProperty(REGION_PROPERTY);
         var bucket = properties.getProperty(BUCKET_PROPERTY);
         var prefix = properties.getProperty(PREFIX_PROPERTY);
+        var region = properties.getProperty(REGION_PROPERTY);
+        var endpoint = properties.getProperty(ENDPOINT_PROPERTY);
 
-        this.region = Region.of(region);
         this.bucket = bucket;
-        this.prefix = prefix != null ? StoragePath.forPath(prefix) : StoragePath.root();
+        this.prefix = prefix != null && !prefix.isBlank() ? StoragePath.forPath(prefix) : StoragePath.root();
+        this.region = region != null && !region.isBlank() ? Region.of(region) : null;
+        this.endpoint = endpoint != null && !endpoint.isBlank() ? URI.create(endpoint) : null;
 
         this.credentials = setupCredentials(properties);
 
@@ -151,14 +156,20 @@ public void start(EventLoopGroup eventLoopGroup) {
                 .advancedOption(SdkAdvancedAsyncClientOption.FUTURE_COMPLETION_EXECUTOR, Runnable::run)
                 .build();
 
-        this.client = S3AsyncClient.builder()
-                .region(region)
-                .credentialsProvider(credentials)
+        var clientBuilder = S3AsyncClient.builder()
                 .httpClientBuilder(httpClient)
                 .asyncConfiguration(async)
-                .build();
+                .credentialsProvider(credentials);
+
+        if (region != null)
+            clientBuilder.region(region);
+
+        if (endpoint != null)
+            clientBuilder.endpointOverride(endpoint);
+
+        this.client = clientBuilder.build();
 
-        log.info("Created S3 storage, region = [{}], bucket = [{}], prefix = [{}]", region, bucket, prefix);
+        log.info("Created S3 storage, bucket = [{}], prefix = [{}]", bucket, prefix);
     }
 
     @Override

tracdap-runtime/python/src/tracdap/rt/_plugins/aws_storage.py

@@ -41,9 +41,10 @@ class S3ObjectStorage(IFileStorage):
 
     # https://arrow.apache.org/docs/python/filesystems.html
 
-    REGION_PROPERTY = "region"
     BUCKET_PROPERTY = "bucket"
     PREFIX_PROPERTY = "prefix"
+    REGION_PROPERTY = "region"
+    ENDPOINT_PROPERTY = "endpoint"
 
     CREDENTIALS_PROPERTY = "credentials"
     CREDENTIALS_DEFAULT = "default"
@@ -59,16 +60,24 @@ def __init__(self, config: _cfg.PluginConfig, options: dict = None):
         self._config = config
         self._options = options
 
-        self._region = config.properties[self.REGION_PROPERTY]
         self._bucket = config.properties[self.BUCKET_PROPERTY]
         self._prefix = config.properties[self.PREFIX_PROPERTY] if self.PREFIX_PROPERTY in config.properties else ""
+        self._region = config.properties[self.REGION_PROPERTY] if self.REGION_PROPERTY in config.properties else None
+        self._endpoint = config.properties[self.ENDPOINT_PROPERTY] if self.ENDPOINT_PROPERTY in config.properties else None
 
         credentials_params = self.setup_credentials(config.properties)
 
-        self.__client = boto3.client(
-            service_name="s3",
-            region_name=self._region,
-            **credentials_params)
+        client_args = {
+            "service_name": "s3",
+            **credentials_params}
+
+        if self._region is not None:
+            client_args["region_name"] = self._region
+
+        if self._endpoint is not None:
+            client_args["endpoint_url"] = self._endpoint
+
+        self.__client = boto3.client(**client_args)
 
     def setup_credentials(self, properties: dict):