Bump Rucio clients to 40.2.0

Update Galaxy's Rucio test and optional dependency declarations to require rucio-clients 40.2.0.

Adjust the Rucio object store for client 40 behavior, including checksum import changes, ingest checksum handling, Rucio config cache invalidation, and isolated download directories.

Run Rucio integration tests against the Savannah Rucio 40.2.0 container by default while keeping an environment override for the image.

doc/source/dev/writing_tests.md

@@ -1145,7 +1145,7 @@ class TestWithExternalService(integration_util.IntegrationTestCase):
 | ``keycloak/keycloak:26.2`` | OIDC authentication | ``oidc/test_auth_oidc.py`` |
 | ``mvdbeek/galaxy-integration-docker-images:slurm-22.01`` | Slurm scheduler | ``test_cli_runners.py`` |
 | ``mvdbeek/galaxy-integration-docker-images:openpbs-22.01`` | PBS scheduler | ``test_cli_runners.py`` |
-| ``savannah.ornl.gov/ndip/public-docker/rucio:1.29.8`` | Rucio data management | ``objectstore/`` tests |
+| ``savannah.ornl.gov/ndip/public-docker/rucio:40.2.0`` | Rucio data management | ``objectstore/`` tests |
 | ``onedata/onezone:21.02.5-dev`` | Onedata storage | ``objectstore/`` tests |
 
 **Environment variables for external services:**

lib/galaxy/dependencies/conditional-requirements.txt

@@ -79,4 +79,4 @@ htcondor
 boto3
 
 # For Rucio storage plugin
-rucio-clients>=33.6.0
+rucio-clients>=40.2.0

lib/galaxy/objectstore/rucio.py

@@ -2,12 +2,14 @@ import hashlib
 import logging
 import os
 import shutil
+import uuid
 
 try:
     import rucio.common
     from rucio.client import Client
     from rucio.client.downloadclient import DownloadClient
     from rucio.client.uploadclient import UploadClient
+    from rucio.common.config import clean_cached_config
 
     from .rucio_extra_clients import (
         DeleteClient,
@@ -92,11 +94,13 @@ def parse_config_xml(config_xml):
             rucio_upload_scheme = e_xml[0].get("scheme", None)
             rucio_scope = e_xml[0].get("scope", None)
             rucio_register_only = string_as_bool(e_xml[0].get("register_only", "False"))
+            rucio_register_with_checksum = string_as_bool(e_xml[0].get("rucio_register_with_checksum", "True"))
         else:
             rucio_upload_rse_name = None
             rucio_upload_scheme = None
             rucio_scope = None
             rucio_register_only = False
+            rucio_register_with_checksum = True
 
         e_xml = config_xml.findall("auth")
         if not e_xml:
@@ -117,6 +121,7 @@ def parse_config_xml(config_xml):
             "upload_scheme": rucio_upload_scheme,
             "scope": rucio_scope,
             "register_only": rucio_register_only,
+            "rucio_register_with_checksum": rucio_register_with_checksum,
             "download_schemes": rucio_download_schemes,
             "account": rucio_account,
             "auth_host": rucio_auth_host,
@@ -149,6 +154,9 @@ class RucioBroker:
         self.upload_rse_name = self.config["upload_rse_name"]
         self.scope = self.config["scope"]
         self.register_only = self.config["register_only"]
+        self.register_with_checksum = self.config.get(
+            "register_with_checksum", self.config.get("rucio_register_with_checksum", True)
+        )
         self.download_schemes = self.config["download_schemes"]
         if Client is None:
             raise Exception(NO_RUCIO_ERROR_MESSAGE)
@@ -173,30 +181,32 @@ username = {self.config["username"]}
         # We may have crossed a forkpool boundary. No harm setting the env var again.
         # Fixes rucio integration tests
         os.environ["RUCIO_CONFIG"] = self.rucio_config_path
+        clean_cached_config()
         client = Client(
             rucio_host=self.config["host"],
             auth_host=self.config["auth_host"],
             account=self.config["account"],
             auth_type=self.config["auth_type"],
+            logger=log,
             creds={"username": self.config["username"], "password": self.config["password"]},
         )
         return client
 
     def get_rucio_upload_client(self, auth_token=None):
         client = self.get_rucio_client()
-        uc = UploadClient(_client=client)
+        uc = UploadClient(_client=client, logger=log)
         uc.auth_token = auth_token
         return uc
 
     def get_rucio_download_client(self, auth_token=None):
         client = self.get_rucio_client()
-        dc = DownloadClient(client=client)
+        dc = DownloadClient(client=client, logger=log)
         dc.auth_token = auth_token
         return dc
 
     def get_rucio_ingest_client(self, auth_token=None):
         client = self.get_rucio_client()
-        ic = InPlaceIngestClient(_client=client)
+        ic = InPlaceIngestClient(client, self.register_with_checksum)
         ic.auth_token = auth_token
         return ic
 
@@ -232,7 +242,8 @@ username = {self.config["username"]}
 
     def download(self, key, dest_path, auth_token):
         key = _encode_key(key)
-        base_dir = os.path.dirname(dest_path)
+        base_dir = os.path.join(os.path.dirname(dest_path), uuid.uuid4().hex)
+        os.makedirs(base_dir, exist_ok=True)
         dids = [{"scope": self.scope, "name": key}]
         try:
             repl = next(self.get_rucio_client().list_replicas(dids))["rses"].keys()
@@ -257,8 +268,8 @@ username = {self.config["username"]}
                 }
             items = [item]
             download_client = self.get_rucio_download_client(auth_token=auth_token)
-            res = download_client.download_dids(items)
             try:
+                res = download_client.download_dids(items)
                 os.replace(res[0]["dest_file_paths"][0], dest_path)
             except Exception as e:
                 if os.path.exists(dest_path):
@@ -269,6 +280,8 @@ username = {self.config["username"]}
         except Exception as e:
             log.exception(f"Cannot download file: {str(e)}")
             return False
+        finally:
+            shutil.rmtree(base_dir, ignore_errors=True)
         return True
 
     def data_object_exists(self, key):

lib/galaxy/objectstore/rucio_extra_clients.py

@@ -1,9 +1,14 @@
 import copy
 import logging
+import os
 import time
 
 try:
     from rucio.client.uploadclient import UploadClient
+    from rucio.common.checksum import (
+        adler32,
+        md5,
+    )
     from rucio.common.exception import (
         InputValidationError,
         NoFilesUploaded,
@@ -57,6 +62,10 @@ class DeleteClient(UploadClient):
 
 
 class InPlaceIngestClient(UploadClient):
+    def __init__(self, client, register_with_checksum):
+        super().__init__(client)
+        self.register_with_checksum = register_with_checksum
+
     def ingest(self, items, summary_file_path=None, traces_copy_out=None, ignore_availability=False, activity=None):
         """
         :param items: List of dictionaries. Each dictionary describing a file to upload. Keys:
@@ -131,6 +140,7 @@ class InPlaceIngestClient(UploadClient):
             basename = file["basename"]
             logger(logging.INFO, "Preparing upload for file %s", basename)
 
+            no_register = False
             pfn = file.get("pfn")
 
             trace = copy.deepcopy(self.trace)
@@ -195,3 +205,36 @@ class InPlaceIngestClient(UploadClient):
         elif num_succeeded != len(files):
             raise NotAllFilesUploaded()
         return 0
+
+    def _collect_file_info(self, filepath, item):
+        """
+        Collects infos (e.g. size, checksums, etc.) about the file and
+        returns them as a dictionary
+        (This function is meant to be used as class internal only)
+
+        :param filepath: path where the file is stored
+        :param item: input options for the given file
+
+        :returns: a dictionary containing all collected info and the input options
+        """
+        new_item = copy.deepcopy(item)
+        new_item["path"] = filepath
+        new_item["dirname"] = os.path.dirname(filepath)
+        new_item["basename"] = os.path.basename(filepath)
+
+        new_item["bytes"] = os.stat(filepath).st_size
+        if self.register_with_checksum:
+            new_item["adler32"] = adler32(filepath)
+            new_item["md5"] = md5(filepath)
+        else:
+            new_item["adler32"] = "00000001"  # empty file
+            new_item["md5"] = "d41d8cd98f00b204e9800998ecf8427e"  # empty file
+
+        new_item["meta"] = {"guid": self._get_file_guid(new_item)}
+        new_item["state"] = "C"
+        if not new_item.get("did_scope"):
+            new_item["did_scope"] = self.default_file_scope
+        if not new_item.get("did_name"):
+            new_item["did_name"] = new_item["basename"]
+
+        return new_item

pyproject.toml

@@ -148,7 +148,7 @@ test = [
     "pytest-shard",
     "responses",
     "roc-validator!=0.7.2",  # https://github.com/crs4/rocrate-validator/pull/97
-    "rucio-clients>=37.2.0",  # https://github.com/rucio/rucio/pull/7665
+    "rucio-clients>=40.2.0",  # https://github.com/rucio/rucio/pull/7665
     "selenium",
     "seletools",
     "statsd",

test/integration/objectstore/_base.py

@@ -21,6 +21,9 @@ OBJECT_STORE_RUCIO_ACCOUNT = os.environ.get("GALAXY_INTEGRATION_OBJECT_STORE_RUC
 OBJECT_STORE_RUCIO_USERNAME = os.environ.get("GALAXY_INTEGRATION_OBJECT_STORE_RUCIO_USERNAME", "rucio")
 OBJECT_STORE_RUCIO_RSE_NAME = "TEST"
 OBJECT_STORE_RUCIO_ACCESS = os.environ.get("GALAXY_INTEGRATION_OBJECT_STORE_RUCIO_ACCESS", "rucio")
+OBJECT_STORE_RUCIO_IMAGE = os.environ.get(
+    "GALAXY_INTEGRATION_OBJECT_STORE_RUCIO_IMAGE", "savannah.ornl.gov/ndip/public-docker/rucio:40.2.0"
+)
 
 OBJECT_STORE_CONFIG = string.Template("""
 <object_store type="hierarchical" id="primary">
@@ -129,7 +132,7 @@ def wait_rucio_ready(container_name):
 
 def start_rucio(container_name):
     ports = [(OBJECT_STORE_PORT, 80)]
-    docker_run("savannah.ornl.gov/ndip/public-docker/rucio:1.29.8", container_name, ports=ports)
+    docker_run(OBJECT_STORE_RUCIO_IMAGE, container_name, ports=ports)
 
     wait_rucio_ready(container_name)