Transfers: Refactor lfn2pfn path translation to a separate class. ruc…

…io#570

lib/rucio/common/config.py

@@ -61,6 +61,24 @@ def config_get_items(section):
     return __CONFIG.items(section)
 
 
+def config_remove_option(section, option):
+    """Remove the specified option from a given section.
+
+       If the option existed in the configuration, return True.
+
+       If the section does not exist, throws NoSectionError.
+    """
+    return __CONFIG.remove_option(section, option)
+
+
+def config_set(section, option, value):
+    """Set a configuration option in a given section.
+
+       If the section does not exist, throws a NoSectionError.
+    """
+    return __CONFIG.set(section, option, value)
+
+
 def get_config_dir():
     """Return the rucio configuration directory"""
     configdirs = ['/opt/rucio/etc/', ]

lib/rucio/rse/protocols/protocol.py

@@ -13,9 +13,11 @@
 # - Mario Lassnig, <mario.lassnig@cern.ch>, 2017
 
 import hashlib
+import importlib
 
 from exceptions import NotImplementedError
 from urlparse import urlparse
+from ConfigParser import NoOptionError, NoSectionError
 
 from rucio.common import config, exception
 from rucio.rse import rsemanager
@@ -26,58 +28,101 @@
 if getattr(rsemanager, 'SERVER_MODE', None):
     from rucio.core import replica
 
-# Provide the ability to register custom lfn2pfn algorithms for this server isntance
-# Algorithms take in three inputs:
-# - scope
-# - name
-# - RSE
-# And should return part of a PFN that will be appended to the rest of the URL plus
-# the prefix.
-__LFN2PFN_ALGORITHMS = {}
-def register_lfn2pfn(lfn2pfn_callable, name=None):
+
+class RSEDeterministicTranslation(object):
     """
-    Provided a callable function, register it as one of the valid LFN2PFN algorithms.
+    Execute the logic for translating a LFN to a path.
     """
-    global __LFN2PFN_ALGORITHMS
-    if name is None:
-        name = lfn2pfn_callable.__name__
-    __LFN2PFN_ALGORITHMS[name] = lfn2pfn_callable
 
+    _LFN2PFN_ALGORITHMS = {}
+    _DEFAULT_LFN2PFN = "hash"
 
-def lfn2pfn_hash(scope, name, rse):
-    """
-    Given a LFN, turn it into a sub-directory structure using a hash function.
-    """
-    hstr = hashlib.md5('%s:%s' % (scope, name)).hexdigest()
-    if scope.startswith('user') or scope.startswith('group'):
-        scope = scope.replace('.', '/')
-    return '%s/%s/%s/%s' % (scope, hstr[0:2], hstr[2:4], name)
+    def __init__(self, rse=None, rse_attributes=None, protocol_attributes=None):
+        """
+        Initialize a translator object from the RSE, its attributes, and the protocol-specific
+        attributes.
+        """
+        self.rse = rse
+        self.rse_attributes = rse_attributes if rse_attributes else {}
+        self.protocol_attributes = protocol_attributes if protocol_attributes else {}
 
+    @staticmethod
+    def register(lfn2pfn_callable, name=None):
+        """
+        Provided a callable function, register it as one of the valid LFN2PFN algorithms.
+
+        The callable will receive five arguments:
+         - scope: Scope of the LFN.
+         - name: LFN's path name
+         - rse: RSE name the translation is being done for.
+         - rse_attributes: Attributes of the RSE.
+         - protocol_attributes: Attributes of the RSE's protocol
+        The return value should be the last part of the PFN - it will be appended to the
+        rest of the URL.
+        """
+        if name is None:
+            name = lfn2pfn_callable.__name__
+        RSEDeterministicTranslation._LFN2PFN_ALGORITHMS[name] = lfn2pfn_callable
 
-def lfn2pfn_identity(scope, name, rse):
-    """
-    Given a LFN, use it as a subdirectory.
-    """
-    if scope.startswith('user') or scope.startswith('group'):
-        scope = scope.replace('.', '/')
-    return '%s/%s' % (scope, name)
-
-
-__LFN2PFN_ALGORITHMS['hash'] = lfn2pfn_hash
-__LFN2PFN_ALGORITHMS['identity'] = lfn2pfn_identity
-if config.config_has_section('policy'):
-    POLICY_MODULE = None
-    try:
-        POLICY_MODULE = config.config_get('policy', 'lfn2pfn_module')
-    except (NoOptionError, NoSectionError) as error:
-        pass
-    if POLICY_MODULE:
-        importlib.import_module(POLICY_MODULE)
-    DEFAULT_LFN2PFN = "hash"
-    try:
-        DEFAULT_LFN2PFN = config.config_get('policy', 'lfn2pfn_algorithm_default')
-    except (NoOptionError, NoSectionError) as error:
-        pass
+    @staticmethod
+    def __hash(scope, name, rse, _, __):
+        """
+        Given a LFN, turn it into a sub-directory structure using a hash function.
+        """
+        hstr = hashlib.md5('%s:%s' % (scope, name)).hexdigest()
+        if scope.startswith('user') or scope.startswith('group'):
+            scope = scope.replace('.', '/')
+        return '%s/%s/%s/%s' % (scope, hstr[0:2], hstr[2:4], name)
+
+    @staticmethod
+    def __identity(scope, name, rse, _, __):
+        """
+        Given a LFN, convert it directly to a path using the mapping:
+
+            scope:path -> scope/path
+
+        """
+        if scope.startswith('user') or scope.startswith('group'):
+            scope = scope.replace('.', '/')
+        return '%s/%s' % (scope, name)
+
+    @classmethod
+    def _module_init_(cls):
+        """
+        Initialize the class object on first module load.
+        """
+        cls.register(cls.__hash, "hash")
+        cls.register(cls.__identity, "identity")
+        if config.config_has_section('policy'):
+            policy_module = None
+        try:
+            policy_module = config.config_get('policy', 'lfn2pfn_module')
+        except (NoOptionError, NoSectionError):
+            pass
+        if policy_module:
+            importlib.import_module(policy_module)
+        cls._DEFAULT_LFN2PFN = "hash"
+        try:
+            cls._DEFAULT_LFN2PFN = config.config_get('policy', 'lfn2pfn_algorithm_default')
+        except (NoOptionError, NoSectionError):
+            pass
+
+    def path(self, scope, name):
+        """ Transforms the logical file name into a PFN's path.
+
+            :param lfn: filename
+            :param scope: scope
+
+            :returns: RSE specific URI of the physical file
+        """
+        algorithm = self.rse_attributes.get('lfn2pfn_algorithm', 'default')
+        if algorithm == 'default':
+            algorithm = RSEDeterministicTranslation._DEFAULT_LFN2PFN
+        algorithm_callable = RSEDeterministicTranslation._LFN2PFN_ALGORITHMS[algorithm]
+        return algorithm_callable(scope, name, self.rse, self.rse_attributes, self.protocol_attributes)
+
+
+RSEDeterministicTranslation._module_init_()
 
 
 class RSEProtocol(object):
@@ -89,16 +134,17 @@ def __init__(self, protocol_attr, rse_settings):
             :param props: Properties of the reuested protocol
         """
         self.attributes = protocol_attr
+        self.translator = None
         self.renaming = True
         self.overwrite = False
         self.rse = rse_settings
-        if not self.rse['deterministic']:
+        if self.rse['deterministic']:
+            self.translator = RSEDeterministicTranslation(self.rse['rse'], self.rse_settings, self.attributes)
+        else:
             if getattr(rsemanager, 'CLIENT_MODE', None):
                 setattr(self, 'lfns2pfns', self.__lfns2pfns_client)
             if getattr(rsemanager, 'SERVER_MODE', None):
                 setattr(self, '_get_path', self._get_path_nondeterministic_server)
-        else:
-            self.attributes['lfn2pfn_algorithm'] = self.rse.get('lfn2pfn_algorithm', None)
 
     def lfns2pfns(self, lfns):
         """
@@ -164,18 +210,14 @@ def __lfns2pfns_client(self, lfns):
     def _get_path(self, scope, name):
         """ Transforms the logical file name into a PFN.
             Suitable for sites implementing the RUCIO naming convention.
+            This implementation is only invoked if the RSE is deterministic.
 
             :param lfn: filename
             :param scope: scope
 
             :returns: RSE specific URI of the physical file
         """
-        algorithm = self.attributes.get('lfn2pfn_algorithm', 'default')
-        if algorithm == 'default':
-            algorithm = DEFAULT_LFN2PFN
-        algorithm_callable = __LFN2PFN_ALGORITHMS[algorithm]
-        # TODO: include RSE.
-        return algorithm_callable(scope, name, None)
+        return self.translator.path(scope, name)
 
     def _get_path_nondeterministic_server(self, scope, name):
         """ Provides the path of a replica for non-deterministic sites. Will be assigned to get path by the __init__ method if neccessary. """