Merge pull request #82 from lsst/tickets/DM-42704

DM-42704: Support multiple S3 endpoints

Author: dhirving

.github/workflows/lint.yaml

@@ -7,8 +7,6 @@ on:
   pull_request:
 
 jobs:
-  call-workflow:
-    uses: lsst/rubin_workflows/.github/workflows/lint.yaml@main
   ruff:
     runs-on: ubuntu-latest
     steps:

doc/changes/DM-42704.feature.rst

@@ -0,0 +1 @@
+``S3ResourcePath`` now supports using multiple S3 endpoints simultaneously.  This is configured using URIs in the form ``s3://profile@bucket/path`` and environment variables ``LSST_RESOURCES_S3_PROFILE_<profile>=https://<access key ID>:<secret key>@<s3 endpoint hostname>``.

python/lsst/resources/s3.py

@@ -192,13 +192,48 @@ def _transfer_config(self) -> TransferConfig:
     @property
     def client(self) -> boto3.client:
         """Client object to address remote resource."""
-        # Defer import for circular dependencies
-        return getS3Client()
+        return getS3Client(self._profile)
+
+    @property
+    def _profile(self) -> str | None:
+        """Profile name to use for looking up S3 credentials and endpoint."""
+        return self._uri.username
+
+    @property
+    def _bucket(self) -> str:
+        """S3 bucket where the files are stored."""
+        # Notionally the bucket is stored in the 'hostname' part of the URI.
+        # However, Ceph S3 uses a "multi-tenant" syntax for bucket names in the
+        # form 'tenant:bucket'.  The part after the colon is parsed as the port
+        # portion of the URI, and urllib throws an exception if you try to read
+        # a non-integer port value.  So manually split off this portion of the
+        # URI.
+        split = self._uri.netloc.split("@")
+        num_components = len(split)
+        if num_components == 2:
+            # There is a profile@ portion of the URL, so take the second half.
+            bucket = split[1]
+        elif num_components == 1:
+            # There is no profile@, so take the whole netloc.
+            bucket = split[0]
+        else:
+            raise ValueError(f"Unexpected extra '@' in S3 URI: '{str(self)}'")
+
+        if not bucket:
+            raise ValueError(f"S3 URI does not include bucket name: '{str(self)}'")
+
+        return bucket
 
     @classmethod
     def _mexists(cls, uris: Iterable[ResourcePath]) -> dict[ResourcePath, bool]:
-        # Force client to be created before creating threads.
-        getS3Client()
+        # Force client to be created for each profile before creating threads.
+        profiles = set[str | None]()
+        for path in uris:
+            if path.scheme == "s3":
+                path = cast(S3ResourcePath, path)
+                profiles.add(path._profile)
+        for profile in profiles:
+            getS3Client(profile)
 
         return super()._mexists(uris)
 
@@ -207,16 +242,16 @@ def exists(self) -> bool:
         """Check that the S3 resource exists."""
         if self.is_root:
             # Only check for the bucket since the path is irrelevant
-            return bucketExists(self.netloc)
-        exists, _ = s3CheckFileExists(self, client=self.client)
+            return bucketExists(self._bucket, self.client)
+        exists, _ = s3CheckFileExists(self, bucket=self._bucket, client=self.client)
         return exists
 
     @backoff.on_exception(backoff.expo, retryable_io_errors, max_time=max_retry_time)
     def size(self) -> int:
         """Return the size of the resource in bytes."""
         if self.dirLike:
             return 0
-        exists, sz = s3CheckFileExists(self, client=self.client)
+        exists, sz = s3CheckFileExists(self, bucket=self._bucket, client=self.client)
         if not exists:
             raise FileNotFoundError(f"Resource {self} does not exist")
         return sz
@@ -229,7 +264,7 @@ def remove(self) -> None:
         # for checking all the keys again, reponse is  HTTP 204 OK
         # response all the time
         try:
-            self.client.delete_object(Bucket=self.netloc, Key=self.relativeToPathRoot)
+            self.client.delete_object(Bucket=self._bucket, Key=self.relativeToPathRoot)
         except (self.client.exceptions.NoSuchKey, self.client.exceptions.NoSuchBucket) as err:
             raise FileNotFoundError("No such resource: {self}") from err
 
@@ -239,7 +274,7 @@ def read(self, size: int = -1) -> bytes:
         if size > 0:
             args["Range"] = f"bytes=0-{size-1}"
         try:
-            response = self.client.get_object(Bucket=self.netloc, Key=self.relativeToPathRoot, **args)
+            response = self.client.get_object(Bucket=self._bucket, Key=self.relativeToPathRoot, **args)
         except (self.client.exceptions.NoSuchKey, self.client.exceptions.NoSuchBucket) as err:
             raise FileNotFoundError(f"No such resource: {self}") from err
         except ClientError as err:
@@ -255,20 +290,20 @@ def write(self, data: bytes, overwrite: bool = True) -> None:
         if not overwrite and self.exists():
             raise FileExistsError(f"Remote resource {self} exists and overwrite has been disabled")
         with time_this(log, msg="Write to %s", args=(self,)):
-            self.client.put_object(Bucket=self.netloc, Key=self.relativeToPathRoot, Body=data)
+            self.client.put_object(Bucket=self._bucket, Key=self.relativeToPathRoot, Body=data)
 
     @backoff.on_exception(backoff.expo, all_retryable_errors, max_time=max_retry_time)
     def mkdir(self) -> None:
         """Write a directory key to S3."""
-        if not bucketExists(self.netloc):
-            raise ValueError(f"Bucket {self.netloc} does not exist for {self}!")
+        if not bucketExists(self._bucket, self.client):
+            raise ValueError(f"Bucket {self._bucket} does not exist for {self}!")
 
         if not self.dirLike:
             raise NotADirectoryError(f"Can not create a 'directory' for file-like URI {self}")
 
         # don't create S3 key when root is at the top-level of an Bucket
         if self.path != "/":
-            self.client.put_object(Bucket=self.netloc, Key=self.relativeToPathRoot)
+            self.client.put_object(Bucket=self._bucket, Key=self.relativeToPathRoot)
 
     @backoff.on_exception(backoff.expo, all_retryable_errors, max_time=max_retry_time)
     def _download_file(self, local_file: IO, progress: ProgressPercentage | None) -> None:
@@ -279,7 +314,7 @@ def _download_file(self, local_file: IO, progress: ProgressPercentage | None) ->
         """
         try:
             self.client.download_fileobj(
-                self.netloc,
+                self._bucket,
                 self.relativeToPathRoot,
                 local_file,
                 Callback=progress,
@@ -324,7 +359,7 @@ def _upload_file(self, local_file: ResourcePath, progress: ProgressPercentage |
         """
         try:
             self.client.upload_file(
-                local_file.ospath, self.netloc, self.relativeToPathRoot, Callback=progress
+                local_file.ospath, self._bucket, self.relativeToPathRoot, Callback=progress
             )
         except self.client.exceptions.NoSuchBucket as err:
             raise NotADirectoryError(f"Target does not exist: {err}") from err
@@ -333,13 +368,13 @@ def _upload_file(self, local_file: ResourcePath, progress: ProgressPercentage |
             raise
 
     @backoff.on_exception(backoff.expo, all_retryable_errors, max_time=max_retry_time)
-    def _copy_from(self, src: ResourcePath) -> None:
+    def _copy_from(self, src: S3ResourcePath) -> None:
         copy_source = {
-            "Bucket": src.netloc,
+            "Bucket": src._bucket,
             "Key": src.relativeToPathRoot,
         }
         try:
-            self.client.copy_object(CopySource=copy_source, Bucket=self.netloc, Key=self.relativeToPathRoot)
+            self.client.copy_object(CopySource=copy_source, Bucket=self._bucket, Key=self.relativeToPathRoot)
         except (self.client.exceptions.NoSuchKey, self.client.exceptions.NoSuchBucket) as err:
             raise FileNotFoundError("No such resource to transfer: {self}") from err
         except ClientError as err:
@@ -469,7 +504,7 @@ def walk(
         filenames = []
         files_there = False
 
-        for page in s3_paginator.paginate(Bucket=self.netloc, Prefix=prefix, Delimiter="/"):
+        for page in s3_paginator.paginate(Bucket=self._bucket, Prefix=prefix, Delimiter="/"):
             # All results are returned as full key names and we must
             # convert them back to the root form. The prefix is fixed
             # and delimited so that is a simple trim
@@ -507,7 +542,7 @@ def _openImpl(
         *,
         encoding: str | None = None,
     ) -> Iterator[ResourceHandleProtocol]:
-        with S3ResourceHandle(mode, log, self.client, self.netloc, self.relativeToPathRoot) as handle:
+        with S3ResourceHandle(mode, log, self.client, self._bucket, self.relativeToPathRoot) as handle:
             if "b" in mode:
                 yield handle
             else:
@@ -529,6 +564,6 @@ def generate_presigned_put_url(self, *, expiration_time_seconds: int) -> str:
     def _generate_presigned_url(self, method: str, expiration_time_seconds: int) -> str:
         return self.client.generate_presigned_url(
             method,
-            Params={"Bucket": self.netloc, "Key": self.relativeToPathRoot},
+            Params={"Bucket": self._bucket, "Key": self.relativeToPathRoot},
             ExpiresIn=expiration_time_seconds,
         )

python/lsst/resources/s3utils.py

@@ -30,17 +30,19 @@
 import functools
 import os
 import re
+import urllib.parse
 from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 from http.client import HTTPException, ImproperConnectionState
 from types import ModuleType
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any, NamedTuple, cast
 from unittest.mock import patch
 
 from botocore.exceptions import ClientError
 from botocore.handlers import validate_bucket_name
 from deprecated.sphinx import deprecated
 from urllib3.exceptions import HTTPError, RequestError
+from urllib3.util import Url, parse_url
 
 if TYPE_CHECKING:
     from unittest import TestCase
@@ -178,18 +180,35 @@ def clean_test_environment_for_s3() -> Iterator[None]:
         yield
 
 
-def getS3Client() -> boto3.client:
+def getS3Client(profile: str | None = None) -> boto3.client:
     """Create a S3 client with AWS (default) or the specified endpoint.
 
+    Parameters
+    ----------
+    profile : `str`, optional
+        The name of an S3 profile describing which S3 service to use.
+
     Returns
     -------
     s3client : `botocore.client.S3`
         A client of the S3 service.
 
     Notes
     -----
-    The endpoint URL is from the environment variable S3_ENDPOINT_URL.
-    If none is specified, the default AWS one is used.
+    If an explicit profile name is specified, its configuration will be read
+    from an environment variable named ``LSST_RESOURCES_S3_PROFILE_<profile>``
+    if it exists.  Note that the name of the profile is case sensitive.  This
+    configuration is specified in the format: ``https://<access key ID>:<secret
+    key>@<s3 endpoint hostname>``. If the access key ID or secret key values
+    contain slashes, the slashes must be URI-encoded (replace "/" with "%2F").
+
+    If profile is `None` or the profile environment variable was not set, the
+    configuration is read from the environment variable ``S3_ENDPOINT_URL``.
+    If it is not specified, the default AWS endpoint is used.
+
+    The access key ID and secret key are optional -- if not specified, they
+    will be looked up via the `AWS credentials file
+    <https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html>`_.
 
     If the environment variable LSST_DISABLE_BUCKET_VALIDATION exists
     and has a value that is not empty, "0", "f", "n", or "false"
@@ -202,26 +221,78 @@ def getS3Client() -> boto3.client:
     if botocore is None:
         raise ModuleNotFoundError("Could not find botocore. Are you sure it is installed?")
 
-    endpoint = os.environ.get("S3_ENDPOINT_URL", None)
+    endpoint = None
+    if profile is not None:
+        var_name = f"LSST_RESOURCES_S3_PROFILE_{profile}"
+        endpoint = os.environ.get(var_name, None)
+    if not endpoint:
+        endpoint = os.environ.get("S3_ENDPOINT_URL", None)
     if not endpoint:
         endpoint = None  # Handle ""
+
     disable_value = os.environ.get("LSST_DISABLE_BUCKET_VALIDATION", "0")
     skip_validation = not re.search(r"^(0|f|n|false)?$", disable_value, re.I)
 
-    return _get_s3_client(endpoint, skip_validation)
+    return _get_s3_client(endpoint, profile, skip_validation)
 
 
 @functools.lru_cache
-def _get_s3_client(endpoint: str, skip_validation: bool) -> boto3.client:
+def _get_s3_client(endpoint: str | None, profile: str | None, skip_validation: bool) -> boto3.client:
     # Helper function to cache the client for this endpoint
     config = botocore.config.Config(read_timeout=180, retries={"mode": "adaptive", "max_attempts": 10})
 
-    client = boto3.client("s3", endpoint_url=endpoint, config=config)
+    endpoint_config = _parse_endpoint_config(endpoint)
+
+    if endpoint_config.access_key_id is not None and endpoint_config.secret_access_key is not None:
+        # We already have the necessary configuration for the profile, so do
+        # not pass the profile to boto3.  boto3 will raise an exception if the
+        # profile is not defined in its configuration file, whether or not it
+        # needs to read the configuration from it.
+        profile = None
+    session = boto3.Session(profile_name=profile)
+
+    client = session.client(
+        "s3",
+        endpoint_url=endpoint_config.endpoint_url,
+        aws_access_key_id=endpoint_config.access_key_id,
+        aws_secret_access_key=endpoint_config.secret_access_key,
+        config=config,
+    )
     if skip_validation:
         client.meta.events.unregister("before-parameter-build.s3", validate_bucket_name)
     return client
 
 
+class _EndpointConfig(NamedTuple):
+    endpoint_url: str | None = None
+    access_key_id: str | None = None
+    secret_access_key: str | None = None
+
+
+def _parse_endpoint_config(endpoint: str | None) -> _EndpointConfig:
+    if not endpoint:
+        return _EndpointConfig()
+
+    parsed = parse_url(endpoint)
+
+    # Strip the username/password portion of the URL from the result.
+    endpoint_url = Url(host=parsed.host, path=parsed.path, port=parsed.port, scheme=parsed.scheme).url
+
+    access_key_id = None
+    secret_access_key = None
+    if parsed.auth:
+        split = parsed.auth.split(":")
+        if len(split) != 2:
+            raise ValueError("S3 access key and secret not in expected format.")
+        access_key_id, secret_access_key = split
+        access_key_id = urllib.parse.unquote(access_key_id)
+        secret_access_key = urllib.parse.unquote(secret_access_key)
+
+    return _EndpointConfig(
+        endpoint_url=endpoint_url, access_key_id=access_key_id, secret_access_key=secret_access_key
+    )
+
+
 def s3CheckFileExists(
     path: Location | ResourcePath | str,
     bucket: str | None = None,
@@ -268,7 +339,8 @@ def s3CheckFileExists(
             bucket = uri.netloc
             filepath = uri.relativeToPathRoot
     elif isinstance(path, ResourcePath | Location):
-        bucket = path.netloc
+        if bucket is None:
+            bucket = path.netloc
         filepath = path.relativeToPathRoot
     else:
         raise TypeError(f"Unsupported path type: {path!r}.")