Repository: rucio/rucio
Issue: __sort_link_ranking in conveyor is done in wrong order

Motivation
----------
https://github.com/rucio/rucio/blob/defa6efc011bb5c30ddf1177f7d6ea26bc8556b4/lib/rucio/daemons/conveyor/submitter.py#L361
This should not be done in reverse but in a normal sort.

Modification
------------
Remove `reverse=True`



Changed files:
lib/rucio/daemons/conveyor/submitter.py, lib/rucio/client/ruleclient.py, lib/rucio/rse/rsemanager.py, lib/rucio/client/uploadclient.py, lib/rucio/core/transfer.py, lib/rucio/client/baseclient.py, lib/rucio/core/monitor.py

diff --git a/lib/rucio/client/baseclient.py b/lib/rucio/client/baseclient.py
index 48036f69f..35528730c 100644
--- a/lib/rucio/client/baseclient.py
+++ b/lib/rucio/client/baseclient.py
@@ -530,7 +530,12 @@ class BaseClient(object):
                         print("Copy paste the code from the browser to the terminal and press enter:")
                         count = 0
                         while count < 3:
-                            fetchcode = raw_input()
+                            # Python3 default
+                            get_input = input
+                            # if Python version <= 2.7 use raw_input
+                            if sys.version_info[:2] <= (2, 7):
+                                get_input = raw_input
+                            fetchcode = get_input()
                             fetch_url = build_url(self.auth_host, path='auth/oidc_redirect', params=fetchcode)
                             result = self.session.get(fetch_url, headers=headers, verify=self.ca_cert)
                             if 'X-Rucio-Auth-Token' in result.headers and result.status_code == codes.ok:
diff --git a/lib/rucio/client/ruleclient.py b/lib/rucio/client/ruleclient.py
index 3967d8504..b08a53f8f 100644
--- a/lib/rucio/client/ruleclient.py
+++ b/lib/rucio/client/ruleclient.py
@@ -21,6 +21,11 @@
 #
 # PY3K COMPATIBLE
 
+try:
+    from urllib import quote_plus
+except ImportError:
+    from urllib.parse import quote_plus
+
 from json import dumps, loads
 from requests.status_codes import codes
 
@@ -205,7 +210,7 @@ class RuleClient(BaseClient):
         :param scope: The scope of the DID.
         :param name: The name of the DID.
         """
-        path = self.RULE_BASEURL + '/' + scope + '/' + name + '/history'
+        path = '/'.join([self.RULE_BASEURL, quote_plus(scope), quote_plus(name), 'history'])
         url = build_url(choice(self.list_hosts), path=path)
         r = self._send_request(url, type='GET')
         if r.status_code == codes.ok:
diff --git a/lib/rucio/client/uploadclient.py b/lib/rucio/client/uploadclient.py
index c44459220..35b1502a2 100644
--- a/lib/rucio/client/uploadclient.py
+++ b/lib/rucio/client/uploadclient.py
@@ -21,6 +21,7 @@
 # - Nicolo Magini <nicolo.magini@cern.ch>, 2018
 # - Hannes Hansen <hannes.jakob.hansen@cern.ch>, 2018
 # - Gabriele Fronze' <gfronze@cern.ch>, 2019
+# - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2020
 #
 # PY3K COMPATIBLE
 
@@ -166,12 +167,23 @@ class UploadClient:
                 logger.warning('Upload with given pfn implies that no_register is True, except non-deterministic RSEs')
                 no_register = True
 
+            # resolving local area networks
+            domain = 'wan'
+            rse_attributes = {}
+            try:
+                rse_attributes = self.client.list_rse_attributes(rse)
+            except:
+                logger.warning('Attributes of the RSE: %s not available.' % rse)
+            if (self.client_location and 'lan' in rse_settings['domain'] and 'site' in rse_attributes):
+                if self.client_location['site'] == rse_attributes['site']:
+                    domain = 'lan'
+
             if not no_register and not register_after_upload:
                 self._register_file(file, registered_dataset_dids)
             # if register_after_upload, file should be overwritten if it is not registered
             # otherwise if file already exists on RSE we're done
             if register_after_upload:
-                if rsemgr.exists(rse_settings, pfn if pfn else file_did, auth_token=self.auth_token):
+                if rsemgr.exists(rse_settings, pfn if pfn else file_did, domain=domain, auth_token=self.auth_token, logger=logger):
                     try:
                         self.client.get_did(file['did_scope'], file['did_name'])
                         logger.info('File already registered. Skipping upload.')
@@ -181,31 +193,20 @@ class UploadClient:
                         logger.info('File already exists on RSE. Previous left overs will be overwritten.')
                         delete_existing = True
             elif not is_deterministic and not no_register:
-                if rsemgr.exists(rse_settings, pfn, auth_token=self.auth_token):
+                if rsemgr.exists(rse_settings, pfn, domain=domain, auth_token=self.auth_token):
                     logger.info('File already exists on RSE with given pfn. Skipping upload. Existing replica has to be removed first.')
                     trace['stateReason'] = 'File already exists'
                     continue
-                elif rsemgr.exists(rse_settings, file_did, auth_token=self.auth_token):
+                elif rsemgr.exists(rse_settings, file_did, domain=domain, auth_token=self.auth_token):
                     logger.info('File already exists on RSE with different pfn. Skipping upload.')
                     trace['stateReason'] = 'File already exists'
                     continue
             else:
-                if rsemgr.exists(rse_settings, pfn if pfn else file_did, auth_token=self.auth_token):
+                if rsemgr.exists(rse_settings, pfn if pfn else file_did, domain=domain, auth_token=self.auth_token):
                     logger.info('File already exists on RSE. Skipping upload')
                     trace['stateReason'] = 'File already exists'
                     continue
 
-            # resolving local area networks
-            domain = 'wan'
-            rse_attributes = {}
-            try:
-                rse_attributes = self.client.list_rse_attributes(rse)
-            except:
-                logger.warning('Attributes of the RSE: %s not available.' % rse)
-            if (self.client_location and 'lan' in rse_settings['domain'] and 'site' in rse_attributes):
-                if self.client_location['site'] == rse_attributes['site']:
-                    domain = 'lan'
-
             # protocol handling and upload
             protocols = rsemgr.get_protocols_ordered(rse_settings=rse_settings, operation='write', scheme=force_scheme, domain=domain)
             protocols.reverse()
@@ -235,13 +236,15 @@ class UploadClient:
                 try:
                     state = rsemgr.upload(rse_settings=rse_settings,
                                           lfns=lfn,
+                                          domain=domain,
                                           source_dir=file['dirname'],
                                           force_scheme=cur_scheme,
                                           force_pfn=pfn,
                                           transfer_timeout=file.get('transfer_timeout'),
                                           delete_existing=delete_existing,
                                           sign_service=sign_service,
-                                          auth_token=self.auth_token)
+                                          auth_token=self.auth_token,
+                                          logger=logger)
                     success = state['success']
                     file['upload_result'] = state
                 except (ServiceUnavailable, ResourceTemporaryUnavailable) as error:
diff --git a/lib/rucio/core/monitor.py b/lib/rucio/core/monitor.py
index cb2dfa68e..8d20abc08 100644
--- a/lib/rucio/core/monitor.py
+++ b/lib/rucio/core/monitor.py
@@ -25,16 +25,16 @@ import time
 from prometheus_client import start_http_server
 from statsd import StatsClient
 
-from rucio.common.config import config_get
+from rucio.common.config import config_get, config_get_bool, config_get_int
 
 SERVER = config_get('monitor', 'carbon_server', raise_exception=False, default='localhost')
 PORT = config_get('monitor', 'carbon_port', raise_exception=False, default=8125)
 SCOPE = config_get('monitor', 'user_scope', raise_exception=False, default='rucio')
 CLIENT = StatsClient(host=SERVER, port=PORT, prefix=SCOPE)
 
-ENABLE_METRICS = config_get('monitor', 'enable_metrics', raise_exception=False, default=False)
+ENABLE_METRICS = config_get_bool('monitor', 'enable_metrics', raise_exception=False, default=False)
 if ENABLE_METRICS:
-    METRICS_PORT = config_get('monitor', 'metrics_port', raise_exception=False, default=8080)
+    METRICS_PORT = config_get_int('monitor', 'metrics_port', raise_exception=False, default=8080)
     start_http_server(METRICS_PORT)
 
 
diff --git a/lib/rucio/core/transfer.py b/lib/rucio/core/transfer.py
index 9d85967aa..dc2cb09f6 100644
--- a/lib/rucio/core/transfer.py
+++ b/lib/rucio/core/transfer.py
@@ -488,8 +488,7 @@ def get_hops(source_rse_id, dest_rse_id, include_multihop=False, multihop_rses=N
     if multihop_rses is None:
         multihop_rses = []
 
-    # TODO: Might be problematic to always load the distance_graph, since it might be expensive
-    # TODO: Have an rse_expression to specify the eligible hops
+    # TODO: Might be problematic to always load the distance_graph, since it might be expensiv
 
     # Load the graph from the distances table
     # distance_graph = __load_distance_graph(session=session)
@@ -523,6 +522,11 @@ def get_hops(source_rse_id, dest_rse_id, include_multihop=False, multihop_rses=N
     # 2. There is no connection or no scheme match --> Try a multi hop --> Dijkstra algorithm
     HOP_PENALTY = core_config_get('transfers', 'hop_penalty', default=10, session=session)  # Penalty to be applied to each further hop
 
+    # Check if the destination RSE is an island RSE:
+    if not __load_distance_edges_node(rse_id=dest_rse_id, session=session):
+        # The destination RSE is an island
+        raise NoDistance()
+
     visited_nodes = {source_rse_id: {'distance': 0,
                                      'path': []}}  # Dijkstra already visisted nodes
     # {rse_id: {'path': [{'source_rse_id':, 'dest_rse_id':, 'source_scheme', 'dest_scheme': }],
diff --git a/lib/rucio/daemons/conveyor/submitter.py b/lib/rucio/daemons/conveyor/submitter.py
index 9b9bb3ab0..1aef4f854 100644
--- a/lib/rucio/daemons/conveyor/submitter.py
+++ b/lib/rucio/daemons/conveyor/submitter.py
@@ -363,7 +363,7 @@ def __sort_link_ranking(sources):
             rank_sources[link_ranking] = []
         rank_sources[link_ranking].append(source)
     rank_keys = list(rank_sources.keys())
-    rank_keys.sort(reverse=True)
+    rank_keys.sort()
     for rank_key in rank_keys:
         sources_list = rank_sources[rank_key]
         random.shuffle(sources_list)
diff --git a/lib/rucio/rse/rsemanager.py b/lib/rucio/rse/rsemanager.py
index 8feadaf4e..3fe00d422 100644
--- a/lib/rucio/rse/rsemanager.py
+++ b/lib/rucio/rse/rsemanager.py
@@ -27,12 +27,14 @@
 # - Nicolo Magini <nicolo.magini@cern.ch>, 2018
 # - James Perry <j.perry@epcc.ed.ac.uk>, 2019
 # - Gabriele Fronze' <gfronze@cern.ch>, 2019
+# - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2020
 #
 # PY3K COMPATIBLE
 
 from __future__ import print_function
 
 import copy
+import logging
 import random
 from time import sleep
 
@@ -46,6 +48,8 @@ from rucio.common.config import config_get_int
 from rucio.common.constraints import STRING_TYPES
 from rucio.common.utils import make_valid_did, GLOBALLY_SUPPORTED_CHECKSUMS
 
+_logger = logging.getLogger(__name__)
+
 
 def get_rse_info(rse, session=None):
     """
@@ -227,7 +231,7 @@ def parse_pfns(rse_settings, pfns, operation='read', domain='wan', auth_token=No
     return create_protocol(rse_settings, operation, urlparse(pfns[0]).scheme, domain, auth_token=auth_token).parse_pfns(pfns)
 
 
-def exists(rse_settings, files, auth_token=None):
+def exists(rse_settings, files, domain='wan', auth_token=None, logger=_logger):
     """
         Checks if a file is present at the connected storage.
         Providing a list indicates the bulk mode.
@@ -236,7 +240,9 @@ def exists(rse_settings, files, auth_token=None):
         :param files: a single dict or a list with dicts containing 'scope' and 'name'
                       if LFNs are used and only 'name' if PFNs are used.
                       E.g. {'name': '2_rse_remote_get.raw', 'scope': 'user.jdoe'}, {'name': 'user/jdoe/5a/98/3_rse_remote_get.raw'}
+        :param domain: The network domain, either 'wan' (default) or 'lan'
         :param auth_token: Optionally passing JSON Web Token (OIDC) string for authentication
+        :param logger: An optional logging.Logger object
 
         :returns: True/False for a single file or a dict object with 'scope:name' for LFNs or 'name' for PFNs as keys and True or the exception as value for each file in bulk mode
 
@@ -245,7 +251,7 @@ def exists(rse_settings, files, auth_token=None):
     ret = {}
     gs = True  # gs represents the global status which inidcates if every operation workd in bulk mode
 
-    protocol = create_protocol(rse_settings, 'read', auth_token=auth_token)
+    protocol = create_protocol(rse_settings, 'read', domain=domain, auth_token=auth_token)
     protocol.connect()
 
     files = [files] if not type(files) is list else files
@@ -258,6 +264,7 @@ def exists(rse_settings, files, auth_token=None):
             pfn = list(protocol.lfns2pfns(f).values())[0]
             if isinstance(pfn, exception.RucioException):
                 raise pfn
+            logger.debug('Checking if %s exists', pfn)
             # deal with URL signing if required
             if rse_settings['sign_url'] is not None and pfn[:5] == 'https':
                 pfn = __get_signed_url(rse_settings['rse'], rse_settings['sign_url'], 'read', pfn)    # NOQA pylint: disable=undefined-variable
@@ -276,7 +283,7 @@ def exists(rse_settings, files, auth_token=None):
     return [gs, ret]
 
 
-def upload(rse_settings, lfns, source_dir=None, force_pfn=None, force_scheme=None, transfer_timeout=None, delete_existing=False, sign_service=None, auth_token=None):
+def upload(rse_settings, lfns, domain='wan', source_dir=None, force_pfn=None, force_scheme=None, transfer_timeout=None, delete_existing=False, sign_service=None, auth_token=None, logger=_logger):
     """
         Uploads a file to the connected storage.
         Providing a list indicates the bulk mode.
@@ -289,12 +296,14 @@ def upload(rse_settings, lfns, source_dir=None, force_pfn=None, force_scheme=Non
                             {'name': '2_rse_local_put.raw', 'scope': 'user.jdoe', 'filesize': 4711, 'adler32': 'RSSMICETHMISBA837464F'}
                             ]
                             If the 'filename' key is present, it will be used by Rucio as the actual name of the file on disk (separate from the Rucio 'name').
+        :param domain: The network domain, either 'wan' (default) or 'lan'
         :param source_dir:  path to the local directory including the source files
         :param force_pfn: use the given PFN -- can lead to dark data, use sparingly
         :param force_scheme: use the given protocol scheme, overriding the protocol priority in the RSE description
         :param transfer_timeout: set this timeout (in seconds) for the transfers, for protocols that support it
         :param sign_service: use the given service (e.g. gcs, s3, swift) to sign the URL
         :param auth_token: Optionally passing JSON Web Token (OIDC) string for authentication
+        :param logger: An optional logging.Logger object
 
         :returns: True/False for a single file or a dict object with 'scope:name' as keys and True or the exception as value for each file in bulk mode
 
@@ -306,9 +315,9 @@ def upload(rse_settings, lfns, source_dir=None, force_pfn=None, force_scheme=Non
     ret = {}
     gs = True  # gs represents the global status which indicates if every operation worked in bulk mode
 
-    protocol = create_protocol(rse_settings, 'write', scheme=force_scheme, auth_token=auth_token)
+    protocol = create_protocol(rse_settings, 'write', scheme=force_scheme, domain=domain, auth_token=auth_token)
     protocol.connect()
-    protocol_delete = create_protocol(rse_settings, 'delete', auth_token=auth_token)
+    protocol_delete = create_protocol(rse_settings, 'delete', domain=domain, auth_token=auth_token)
     protocol_delete.connect()
 
     lfns = [lfns] if not type(lfns) is list else lfns
@@ -348,6 +357,7 @@ def upload(rse_settings, lfns, source_dir=None, force_pfn=None, force_scheme=Non
             else:
                 if protocol.exists('%s.rucio.upload' % pfn):  # Check for left over of previous unsuccessful attempts
                     try:
+                        logger.debug('Deleting %s.rucio.upload', pfn)
                         protocol_delete.delete('%s.rucio.upload' % list(protocol_delete.lfns2pfns(make_valid_did(lfn)).values())[0])
                     except Exception as e:
                         ret['%s:%s' % (scope, name)] = exception.RSEOperationNotSupported('Unable to remove temporary file %s.rucio.upload: %s' % (pfn, str(e)))
@@ -357,6 +367,7 @@ def upload(rse_settings, lfns, source_dir=None, force_pfn=None, force_scheme=Non
                 if delete_existing:
                     if protocol.exists('%s' % pfn):  # Check for previous completed uploads that have to be removed before upload
                         try:
+                            logger.debug('Deleting %s', pfn)
                             protocol_delete.delete('%s' % list(protocol_delete.lfns2pfns(make_valid_did(lfn)).values())[0])
                         except Exception as e:
                             ret['%s:%s' % (scope, name)] = exception.RSEOperationNotSupported('Unable to remove file %s: %s' % (pfn, str(e)))
@@ -364,6 +375,7 @@ def upload(rse_settings, lfns, source_dir=None, force_pfn=None, force_scheme=Non
                             continue
 
                 try:  # Try uploading file
+                    logger.debug('Uploading to %s.rucio.upload', pfn)
                     protocol.put(base_name, '%s.rucio.upload' % pfn, source_dir, transfer_timeout=transfer_timeout)
                 except Exception as e:
                     gs = False
@@ -402,6 +414,7 @@ def upload(rse_settings, lfns, source_dir=None, force_pfn=None, force_scheme=Non
 
                 if valid:  # The upload finished successful and the file can be renamed
                     try:
+                        logger.debug('Renaming %s.rucio.upload to %s', pfn, pfn)
                         protocol.rename('%s.rucio.upload' % pfn, pfn)
                         ret['%s:%s' % (scope, name)] = True
                     except Exception as e:
@@ -418,6 +431,7 @@ def upload(rse_settings, lfns, source_dir=None, force_pfn=None, force_scheme=Non
                 gs = False
             else:
                 try:  # Try uploading file
+                    logger.debug('Uploading to %s', pfn)
                     protocol.put(base_name, pfn, source_dir, transfer_timeout=transfer_timeout)
                 except Exception as e:
                     gs = False
@@ -470,13 +484,14 @@ def upload(rse_settings, lfns, source_dir=None, force_pfn=None, force_scheme=Non
     return {0: gs, 1: ret, 'success': gs, 'pfn': pfn}
 
 
-def delete(rse_settings, lfns, auth_token=None):
+def delete(rse_settings, lfns, domain='wan', auth_token=None):
     """
         Delete a file from the connected storage.
         Providing a list indicates the bulk mode.
 
         :rse_settings:   RSE attributes
         :param lfns:        a single dict or a list with dicts containing 'scope' and 'name'. E.g. [{'name': '1_rse_remote_delete.raw', 'scope': 'user.jdoe'}, {'name': '2_rse_remote_delete.raw', 'scope': 'user.jdoe'}]
+        :param domain: The network domain, either 'wan' (default) or 'lan'
         :param auth_token: Optionally passing JSON Web Token (OIDC) string for authentication
 
         :returns: True/False for a single file or a dict object with 'scope:name' as keys and True or the exception as value for each file in bulk mode
@@ -489,7 +504,7 @@ def delete(rse_settings, lfns, auth_token=None):
     ret = {}
     gs = True  # gs represents the global status which inidcates if every operation workd in bulk mode
 
-    protocol = create_protocol(rse_settings, 'delete', auth_token=auth_token)
+    protocol = create_protocol(rse_settings, 'delete', domain, auth_token=auth_token)
     protocol.connect()
 
     lfns = [lfns] if not type(lfns) is list else lfns
@@ -512,7 +527,7 @@ def delete(rse_settings, lfns, auth_token=None):
     return [gs, ret]
 
 
-def rename(rse_settings, files, auth_token=None):
+def rename(rse_settings, files, domain='wan', auth_token=None):
     """
         Rename files stored on the connected storage.
         Providing a list indicates the bulk mode.
@@ -526,6 +541,7 @@ def rename(rse_settings, files, auth_token=None):
                       {'name': '3_rse_remote_rename.raw', 'scope': 'user.jdoe', 'new_name': '3_rse_new.raw', 'new_scope': 'user.jdoe'},
                       {'name': 'user/jdoe/d9/cb/9_rse_remote_rename.raw', 'new_name': 'user/jdoe/c6/4a/9_rse_new.raw'}
                       ]
+        :param domain: The network domain, either 'wan' (default) or 'lan'
         :param auth_token: Optionally passing JSON Web Token (OIDC) string for authentication
 
         :returns: True/False for a single file or a dict object with LFN (key) and True/False (value) in bulk mode
@@ -538,7 +554,7 @@ def rename(rse_settings, files, auth_token=None):
     ret = {}
     gs = True  # gs represents the global status which inidcates if every operation workd in bulk mode
 
-    protocol = create_protocol(rse_settings, 'write', auth_token=auth_token)
+    protocol = create_protocol(rse_settings, 'write', domain, auth_token=auth_token)
     protocol.connect()
 
     files = [files] if not type(files) is list else files
@@ -586,12 +602,13 @@ def rename(rse_settings, files, auth_token=None):
     return [gs, ret]
 
 
-def get_space_usage(rse_settings, scheme=None, auth_token=None):
+def get_space_usage(rse_settings, scheme=None, domain='wan', auth_token=None):
     """
         Get RSE space usage information.
 
         :rse_settings:   RSE attributes
         :param scheme: optional filter to select which protocol to be used.
+        :param domain: The network domain, either 'wan' (default) or 'lan'
         :param auth_token: Optionally passing JSON Web Token (OIDC) string for authentication
 
         :returns: a list with dict containing 'totalsize' and 'unusedsize'
@@ -601,7 +618,7 @@ def get_space_usage(rse_settings, scheme=None, auth_token=None):
     gs = True
     ret = {}
 
-    protocol = create_protocol(rse_settings, 'read', scheme, auth_token=auth_token)
+    protocol = create_protocol(rse_settings, 'read', scheme=scheme, domain=domain, auth_token=auth_token)
     protocol.connect()
 
     try: