Repository: rucio/rucio
Issue: Resurrect method doesn't remove expired_at for deleted_dids

Motivation
----------
The resurrect method doesn't remove the expired_at when a DID is resurrected which triggers the immediate deletion of the resurrected DID


Changed files:
lib/rucio/tests/test_archive.py, lib/rucio/core/did.py, lib/rucio/daemons/bb8/nuclei_background_rebalance.py, lib/rucio/core/replica.py, lib/rucio/web/rest/webpy/v1/replica.py, lib/rucio/daemons/bb8/t2_background_rebalance.py, lib/rucio/common/utils.py, lib/rucio/tests/test_replica_sorting.py

diff --git a/lib/rucio/common/utils.py b/lib/rucio/common/utils.py
index 8853835f4..4f05f0ff1 100644
--- a/lib/rucio/common/utils.py
+++ b/lib/rucio/common/utils.py
@@ -15,7 +15,7 @@
 # Authors:
 # - Vincent Garonne <vgaronne@gmail.com>, 2012-2018
 # - Thomas Beermann <thomas.beermann@cern.ch>, 2012-2018
-# - Mario Lassnig <mario.lassnig@cern.ch>, 2012-2017
+# - Mario Lassnig <mario.lassnig@cern.ch>, 2012-2018
 # - Cedric Serfon <cedric.serfon@cern.ch>, 2013-2017
 # - Ralph Vigne <ralph.vigne@cern.ch>, 2013
 # - Joaquin Bogado <jbogado@linti.unlp.edu.ar>, 2015-2018
@@ -39,30 +39,38 @@ import re
 import requests
 import socket
 import subprocess
+import urllib
 import zlib
 
 from getpass import getuser
+from logging import getLogger, Formatter
+from logging.handlers import RotatingFileHandler
+from uuid import uuid4 as uuid
+
 try:
     # Python 2
     from itertools import izip_longest
 except ImportError:
     # Python 3
     from itertools import zip_longest as izip_longest
-from logging import getLogger, Formatter
-from logging.handlers import RotatingFileHandler
 try:
     # Python 2
     from urllib import urlencode, quote
 except ImportError:
     # Python 3
     from urllib.parse import urlencode, quote
-from uuid import uuid4 as uuid
 try:
     # Python 2
     from StringIO import StringIO
 except ImportError:
     # Python 3
     from io import StringIO
+try:
+    # Python 2
+    import urlparse
+except ImportError:
+    # Python 3
+    import urllib.parse as urlparse
 
 from rucio.common.config import config_get
 from rucio.common.exception import MissingModuleException
@@ -704,3 +712,19 @@ def send_trace(trace, trace_endpoint, user_agent, retries=5, logger=None, log_pr
         except Exception as error:
             logger.debug('%s%s' % (log_prefix, error))
     return 1
+
+
+def add_url_query(url, query):
+    """
+    Add a new dictionary to URL parameters
+
+    :param url: The existing URL
+    :param query: A dictionary containing key/value pairs to be added to the URL
+    :return: The expanded URL with the new query parameters
+    """
+
+    url_parts = list(urlparse.urlparse(url))
+    mod_query = dict(urlparse.parse_qsl(url_parts[4]))
+    mod_query.update(query)
+    url_parts[4] = urllib.urlencode(mod_query)
+    return urlparse.urlunparse(url_parts)
diff --git a/lib/rucio/core/did.py b/lib/rucio/core/did.py
index eb13cae01..a38e4a63b 100644
--- a/lib/rucio/core/did.py
+++ b/lib/rucio/core/did.py
@@ -312,11 +312,9 @@ def __add_files_to_archive(scope, name, files, account, ignore_duplicate=False,
             session.query(models.DataIdentifier).\\
                 filter(models.DataIdentifier.did_type == DIDType.FILE).\\
                 filter(or_(*existing_files_condition)).update({'constituent': True})
-
-        print contents
         contents and session.bulk_insert_mappings(models.ConstituentAssociation, contents)
         session.flush()
-    except IntegrityError, error:
+    except IntegrityError as error:
         raise exception.RucioException(error.args)
 
 
@@ -1593,6 +1591,8 @@ def resurrect(dids, session=None):
         #    raise exception.UnsupportedOperation("File '%(scope)s:%(name)s' cannot be resurrected" % did)
 
         kargs = del_did.to_dict()
+        if kargs['expired_at']:
+            kargs['expired_at'] = None
         kargs.pop("_sa_instance_state", None)
 
         session.query(models.DeletedDataIdentifier).\\
diff --git a/lib/rucio/core/replica.py b/lib/rucio/core/replica.py
index 48f867db6..bfb46e813 100644
--- a/lib/rucio/core/replica.py
+++ b/lib/rucio/core/replica.py
@@ -38,7 +38,7 @@ from sqlalchemy.sql.expression import case, bindparam, select, text, false, true
 import rucio.core.lock
 
 from rucio.common import exception
-from rucio.common.utils import chunks, clean_surls, str_to_date
+from rucio.common.utils import chunks, clean_surls, str_to_date, add_url_query
 from rucio.core.credential import get_signed_url
 from rucio.core.rse import get_rse, get_rse_id, get_rse_name, get_rse_attribute, get_rses_with_attribute_value
 from rucio.core.rse_counter import decrease, increase
@@ -549,7 +549,7 @@ def get_did_from_pfns(pfns, rse=None, session=None):
 
 def _resolve_dids(dids, unavailable, ignore_availability, all_states, session):
     """
-    resolve list of dids into a list of conditions.
+    Resolve list of DIDs into a list of conditions.
 
     :param dids: The list of data identifiers (DIDs).
     :param unavailable: Also include unavailable replicas in the list.
@@ -557,7 +557,7 @@ def _resolve_dids(dids, unavailable, ignore_availability, all_states, session):
     :param all_states: Return all replicas whatever state they are in. Adds an extra 'states' entry in the result dictionary.
     :param session: The database session in use.
     """
-    did_clause, dataset_clause, file_clause, files = [], [], [], []
+    did_clause, dataset_clause, file_clause, files, constituents = [], [], [], [], {}
     for did in [dict(tupleized) for tupleized in set(tuple(item.items()) for item in dids)]:
         if 'type' in did and did['type'] in (DIDType.FILE, DIDType.FILE.value) or 'did_type' in did and did['did_type'] in (DIDType.FILE, DIDType.FILE.value):  # pylint: disable=no-member
             files.append({'scope': did['scope'], 'name': did['name']})
@@ -569,9 +569,22 @@ def _resolve_dids(dids, unavailable, ignore_availability, all_states, session):
                                    models.DataIdentifier.name == did['name']))
 
     if did_clause:
-        for scope, name, did_type in session.query(models.DataIdentifier.scope,
-                                                   models.DataIdentifier.name,
-                                                   models.DataIdentifier.did_type).with_hint(models.DataIdentifier, "INDEX(DIDS DIDS_PK)", 'oracle').filter(or_(*did_clause)):
+        for scope, name, did_type, constituent in session.query(models.DataIdentifier.scope,
+                                                                models.DataIdentifier.name,
+                                                                models.DataIdentifier.did_type,
+                                                                models.DataIdentifier.constituent)\\
+                                                         .with_hint(models.DataIdentifier, "INDEX(DIDS DIDS_PK)", 'oracle')\\
+                                                         .filter(or_(*did_clause)):
+            if constituent:
+                # file is a constituent, resolve to any parent archive
+                archive = session.query(models.ConstituentAssociation.scope,
+                                        models.ConstituentAssociation.name)\\
+                                 .filter(models.ConstituentAssociation.child_scope == scope,
+                                         models.ConstituentAssociation.child_name == name)\\
+                                 .with_hint(models.ConstituentAssociation, "INDEX(ARCHIVE_CONTENTS ARCH_CONTENTS_PK)", 'oracle')\\
+                                 .all()
+                constituents['%s:%s' % (scope, name)] = [{'scope': tmp[0], 'name': tmp[1]} for tmp in archive]
+
             if did_type == DIDType.FILE:
                 files.append({'scope': scope, 'name': name})
                 file_clause.append(and_(models.RSEFileAssociation.scope == scope,
@@ -599,7 +612,6 @@ def _resolve_dids(dids, unavailable, ignore_availability, all_states, session):
 
     state_clause = None
     if not all_states:
-        # models.RSE.volatile == false()
         if not unavailable:
             state_clause = and_(models.RSEFileAssociation.state == ReplicaState.AVAILABLE)
 
@@ -608,7 +620,7 @@ def _resolve_dids(dids, unavailable, ignore_availability, all_states, session):
                                models.RSEFileAssociation.state == ReplicaState.UNAVAILABLE,
                                models.RSEFileAssociation.state == ReplicaState.COPYING)
 
-    return file_clause, dataset_clause, state_clause, files
+    return file_clause, dataset_clause, state_clause, files, constituents
 
 
 def _list_replicas_for_datasets(dataset_clause, state_clause, rse_clause, session):
@@ -722,11 +734,10 @@ def _list_replicas_for_files(file_clause, state_clause, files, rse_clause, sessi
             yield scope, name, bytes, md5, adler32, None, None, None, None, None
             {'scope': scope, 'name': name} in files and files.remove({'scope': scope, 'name': name})
 
-        # if files:
-        #    raise exception.DataIdentifierNotFound("Files not found %s", str(files))
-
 
-def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns, schemes, files, rse_clause, client_location, domain, sign_urls, signature_lifetime, session):
+def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns,
+                   schemes, files, rse_clause, client_location, domain,
+                   sign_urls, signature_lifetime, constituents, session):
 
     files = [dataset_clause and _list_replicas_for_datasets(dataset_clause, state_clause, rse_clause, session),
              file_clause and _list_replicas_for_files(file_clause, state_clause, files, rse_clause, session)]
@@ -748,12 +759,58 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns, schemes
     for replicas in filter(None, files):
         for scope, name, bytes, md5, adler32, path, state, rse, rse_type, volatile in replicas:
 
+            pfns = []
+
             # reset the domain selection to original user's choice (as this could get overwritten each iteration)
             domain = deepcopy(original_domain)
 
-            pfns = []
-            if show_pfns and rse:
+            # if the file is a constituent, find the available archives and add them to the list of possible PFNs
+            if '%s:%s' % (scope, name) in constituents:
+                archive_result = list_replicas(dids=constituents['%s:%s' % (scope, name)],
+                                               schemes=schemes, client_location=client_location,
+                                               domain=domain, sign_urls=sign_urls,
+                                               signature_lifetime=signature_lifetime,
+                                               session=session)
+
+                # is it the only instance (i.e., no RSE for the replica)?
+                # then yield and continue as the replica does not exist and we don't want to return
+                # an additional empty pfn to the client
+                if rse is None:
+                    for archive in archive_result:
+                        for tmp_archive in archive['pfns']:
+                            archive['pfns'][tmp_archive]['domain'] = 'zip'
+                            # at this point we don't know the protocol of the parent, so we have to peek
+                            if tmp_archive.startswith('root://') and 'xrdcl.unzip' not in tmp_archive:
+                                # use direct addressable path for root
+                                new_pfn = add_url_query(tmp_archive, {'xrdcl.unzip': name})
+                                archive['pfns'][new_pfn] = archive['pfns'].pop(tmp_archive)
+                                tmp_archive = new_pfn
+                            if 'xrdcl.unzip' not in tmp_archive:
+                                archive['pfns'][tmp_archive]['client_extract'] = True
+                        yield archive
+                    continue
+
+                # now, repeat the procedure slightly different in case if there are replicas available
+                # and make the archive just an additional available PFN
+                available_archives = [r['pfns'] for r in archive_result]
+                for archive in available_archives:
+                    for archive_pfn in archive:
+                        # at this point we don't know the protocol of the parent, so we have to peek
+                        if archive_pfn.startswith('root://'):
+                            # use direct addressable path for root
+                            pfn = add_url_query(archive_pfn, {'xrdcl.unzip': name})
+                            client_extract = False
+                        else:
+                            # otherwise just use the zip and tell the client to extract
+                            pfn = archive_pfn
+                            client_extract = True
+
+                        # use zip domain for sorting, and pass through all the information
+                        # we need it later to reconstruct the final PFNS
+                        # ('pfn', 'domain', 'priority', 'client_extract', archive-passthrough)
+                        pfns.append((pfn, 'zip', archive[archive_pfn]['priority'], client_extract, archive))
 
+            if show_pfns and rse:
                 if rse not in rse_info:
                     rse_info[rse] = rsemgr.get_rse_info(rse, session=session)
 
@@ -852,12 +909,11 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns, schemes
                             if sign and isinstance(sign, list) and sign[0]:
                                 pfn = get_signed_url(service='gcs', operation='read', url=pfn, lifetime=signature_lifetime)
 
-                        # TODO: this is not nice, but since pfns don't have the concept of 'domain' or 'priority'
-                        #       we can work around by encapsulating it in a tuple. a proper refactor requires
-                        #       far-reaching changes in the rsemgr
-                        pfns.append((pfn, tmp_protocol[0], tmp_protocol[2]))
+                        # PFNs don't have concepts, therefore quickly encapsulate in a tuple
+                        # ('pfn', 'domain', 'priority', 'client_extract')
+                        pfns.append((pfn, tmp_protocol[0], tmp_protocol[2], False))
                     except:
-                        # temporary protection
+                        # never end up here
                         print format_exc()
 
                     if protocol.attributes['scheme'] == 'srm':
@@ -868,7 +924,8 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns, schemes
 
             if 'scope' in file and 'name' in file:
                 if file['scope'] == scope and file['name'] == name:
-                    file['rses'][rse] += list(set([tmp_pfn[0] for tmp_pfn in pfns]))  # extract properly the pfn from the (pfn, domain, priority)
+                    # extract properly the pfn from the tuple
+                    file['rses'][rse] += list(set([tmp_pfn[0] for tmp_pfn in pfns]))
                     file['states'][rse] = str(state)
                     for tmp_pfn in pfns:
                         file['pfns'][tmp_pfn[0]] = {'rse': rse,
@@ -876,7 +933,16 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns, schemes
                                                     'volatile': volatile,
                                                     'domain': tmp_pfn[1],
                                                     'priority': tmp_pfn[2]}
+                        # instruct the client that the archive needs to be manually extracted
+                        if tmp_pfn[3]:
+                            file['pfns'][tmp_pfn[0]]['client_extract'] = True
                 else:
+                    # quick exit, but don't forget to set the total order for the priority
+                    # --> exploit that L(AN) comes before W(AN) before Z(IP) alphabetically
+                    # and use 1-indexing to be compatible with metalink
+                    tmp = sorted([(file['pfns'][p]['domain'], file['pfns'][p]['priority'], p) for p in file['pfns']])
+                    for i in xrange(0, len(tmp)):
+                        file['pfns'][tmp[i][2]]['priority'] = i + 1
                     yield file
                     file = {}
 
@@ -886,13 +952,25 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns, schemes
                         'pfns': {}, 'rses': defaultdict(list),
                         'states': {rse: str(state)}}
                 if rse:
-                    file['rses'][rse] = list(set([tmp_pfn[0] for tmp_pfn in pfns]))  # extract properly the pfn from the (pfn, domain, priority)
+                    # extract properly the pfn from the tuple
+                    file['rses'][rse] = list(set([tmp_pfn[0] for tmp_pfn in pfns]))
                     for tmp_pfn in pfns:
                         file['pfns'][tmp_pfn[0]] = {'rse': rse,
                                                     'type': str(rse_type),
                                                     'volatile': volatile,
                                                     'domain': tmp_pfn[1],
                                                     'priority': tmp_pfn[2]}
+                        # instruct the client that the archive needs to be manually extracted
+                        if tmp_pfn[3]:
+                            file['pfns'][tmp_pfn[0]]['client_extract'] = True
+
+    # set the total order for the priority
+    # --> exploit that L(AN) comes before W(AN) before Z(IP) alphabetically
+    # and use 1-indexing to be compatible with metalink
+    if 'pfns' in file:
+        tmp = sorted([(file['pfns'][p]['domain'], file['pfns'][p]['priority'], p) for p in file['pfns']])
+        for i in xrange(0, len(tmp)):
+            file['pfns'][tmp[i][2]]['priority'] = i + 1
 
     if 'scope' in file and 'name' in file:
         yield file
@@ -921,16 +999,18 @@ def list_replicas(dids, schemes=None, unavailable=False, request_id=None,
     :param session: The database session in use.
     """
 
-    file_clause, dataset_clause, state_clause, files = _resolve_dids(dids=dids, unavailable=unavailable,
-                                                                     ignore_availability=ignore_availability,
-                                                                     all_states=all_states, session=session)
+    file_clause, dataset_clause, state_clause, files, constituents = _resolve_dids(dids=dids, unavailable=unavailable,
+                                                                                   ignore_availability=ignore_availability,
+                                                                                   all_states=all_states, session=session)
+
     rse_clause = []
     if rse_expression:
         for rse in parse_expression(expression=rse_expression, session=session):
             rse_clause.append(models.RSEFileAssociation.rse_id == rse['id'])
-
-    for file in _list_replicas(dataset_clause, file_clause, state_clause, pfns, schemes, files, rse_clause, client_location, domain, sign_urls, signature_lifetime, session):
-        yield file
+    for f in _list_replicas(dataset_clause, file_clause, state_clause, pfns,
+                            schemes, files, rse_clause, client_location, domain,
+                            sign_urls, signature_lifetime, constituents, session):
+        yield f
 
 
 @transactional_session
diff --git a/lib/rucio/daemons/bb8/nuclei_background_rebalance.py b/lib/rucio/daemons/bb8/nuclei_background_rebalance.py
index c11a3bd98..74561dfd8 100644
--- a/lib/rucio/daemons/bb8/nuclei_background_rebalance.py
+++ b/lib/rucio/daemons/bb8/nuclei_background_rebalance.py
@@ -6,11 +6,11 @@
 #
 # Authors:
 # - Martin Barisits, <martin.barisits@cern.ch>, 2016-2017
+# - Tomas Javurek, <tomas.javurek@cern.ch>, 2017
 
 """
-This script is to be used to background rebalance ATLAS Nuclei datadisks
+This script is to be used to background rebalance ATLAS t2 datadisks
 """
-
 from sqlalchemy import or_
 
 from rucio.core.rse_expression_parser import parse_expression
@@ -23,24 +23,27 @@ from rucio.db.sqla.constants import RuleState
 
 tolerance = 0.1
 max_total_rebalance_volume = 200 * 1E12
-max_rse_rebalance_volume = 20 * 1E12
-
+max_rse_rebalance_volume = 40 * 1E12
+min_total = 50 * 1E12
 total_rebalance_volume = 0
 
 # Calculate the current ratios
-rses = parse_expression('(datapolicynucleus=1|tier=1)&type=DATADISK')
+rses = parse_expression("(datapolicynucleus=1|tier=1)&type=DATADISK\\\\bb8-enabled=0")
 total_primary = 0
 total_secondary = 0
+total_total = 0
 global_ratio = float(0)
 for rse in rses:
-    rse['primary'] = get_rse_usage(rse=None, rse_id=rse['id'], source='rucio')[0]['used']
+    rse['primary'] = get_rse_usage(rse=None, rse_id=rse['id'], source='rucio')[0]['used'] - get_rse_usage(rse=None, rse_id=rse['id'], source='expired')[0]['used']
     rse['secondary'] = get_rse_usage(rse=None, rse_id=rse['id'], source='expired')[0]['used']
-    rse['ratio'] = float(rse['primary']) / float(rse['secondary'])
+    rse['total'] = get_rse_usage(rse=None, rse_id=rse['id'], source='storage')[0]['total'] - get_rse_usage(rse=None, rse_id=rse['id'], source='min_free_space')[0]['used']
+    rse['ratio'] = float(rse['primary']) / float(rse['total'])
     total_primary += rse['primary']
     total_secondary += rse['secondary']
+    total_total += float(rse['total'])
     rse['receive_volume'] = 0  # Already rebalanced volume in this run
-global_ratio = float(total_primary) / float(total_secondary)
 
+global_ratio = float(total_primary) / float(total_total)
 print 'Global ratio: %f' % (global_ratio)
 for rse in sorted(rses, key=lambda k: k['ratio']):
     print '  %s (%f)' % (rse['rse'], rse['ratio'])
@@ -50,7 +53,9 @@ rses_under_ratio = sorted([rse for rse in rses if rse['ratio'] < global_ratio -
 
 session = get_session()
 active_rses = session.query(models.ReplicationRule.rse_expression).filter(or_(models.ReplicationRule.state == RuleState.REPLICATING, models.ReplicationRule.state == RuleState.STUCK),
-                                                                          models.ReplicationRule.comments == 'Nuclei Background rebalancing').group_by(models.ReplicationRule.rse_expression).all()
+                                                                          models.ReplicationRule.comments == 'T2 Background rebalancing').group_by(models.ReplicationRule.rse_expression).all()
+
+# Excluding RSEs
 print 'Excluding RSEs as destination which have active Background Rebalancing rules:'
 for rse in active_rses:
     print '  %s' % (rse[0])
@@ -59,26 +64,52 @@ for rse in active_rses:
             rses_under_ratio.remove(des)
             break
 
+print 'Excluding RSEs as destination which are too small by size:'
+for des in rses_under_ratio:
+    if des['total'] < min_total:
+        print '  %s' % (des['rse'])
+        rses_under_ratio.remove(des)
+
+print 'Excluding RSEs as sources which are too small by size:'
+for src in rses_over_ratio:
+    if src['total'] < min_total:
+        print '  %s' % (src['rse'])
+        rses_over_ratio.remove(src)
+
+print 'Excluding RSEs as desetinations which are blacklisted:'
+for des in rses_under_ratio:
+    if des['availability'] != 7:
+        print '  %s' % (des['rse'])
+        rses_under_ratio.remove(des)
+
+print 'Excluding RSEs as sources which are blacklisted:'
+for src in rses_over_ratio:
+    if src['availability'] != 7:
+        print '  %s' % (src['rse'])
+        rses_over_ratio.remove(src)
+
 # Loop over RSEs over the ratio
 for source_rse in rses_over_ratio:
-    if source_rse['ratio'] > global_ratio + global_ratio * tolerance:
-        available_source_rebalance_volume = int((source_rse['primary'] - global_ratio * source_rse['secondary']) / (global_ratio + 1))
-        if available_source_rebalance_volume > max_rse_rebalance_volume:
-            available_source_rebalance_volume = max_rse_rebalance_volume
-        if available_source_rebalance_volume > max_total_rebalance_volume - total_rebalance_volume:
-            available_source_rebalance_volume = max_total_rebalance_volume - total_rebalance_volume
-        # Select a target:
-        for destination_rse in rses_under_ratio:
-            if available_source_rebalance_volume > 0:
-                if destination_rse['receive_volume'] >= max_rse_rebalance_volume:
-                    continue
-                available_target_rebalance_volume = max_rse_rebalance_volume - destination_rse['receive_volume']
-                if available_target_rebalance_volume >= available_source_rebalance_volume:
-                    available_target_rebalance_volume = available_source_rebalance_volume
-
-                print 'Rebalance %dTB from %s(%f) to %s(%f)' % (available_target_rebalance_volume / 1E12, source_rse['rse'], source_rse['ratio'], destination_rse['rse'], destination_rse['ratio'])
-                rebalance_rse(source_rse['rse'], max_bytes=available_target_rebalance_volume, dry_run=False, comment='Nuclei Background rebalancing', force_expression=destination_rse['rse'])
-
-                destination_rse['receive_volume'] += available_target_rebalance_volume
-                total_rebalance_volume += available_target_rebalance_volume
-                available_source_rebalance_volume -= available_target_rebalance_volume
+
+    # The volume that would be rebalanced, not real availability of the data:
+    available_source_rebalance_volume = int((source_rse['primary'] - global_ratio * source_rse['secondary']) / (global_ratio + 1))
+    if available_source_rebalance_volume > max_rse_rebalance_volume:
+        available_source_rebalance_volume = max_rse_rebalance_volume
+    if available_source_rebalance_volume > max_total_rebalance_volume - total_rebalance_volume:
+        available_source_rebalance_volume = max_total_rebalance_volume - total_rebalance_volume
+
+    # Select a target:
+    for destination_rse in rses_under_ratio:
+        if available_source_rebalance_volume > 0:
+            if destination_rse['receive_volume'] >= max_rse_rebalance_volume:
+                continue
+            available_target_rebalance_volume = max_rse_rebalance_volume - destination_rse['receive_volume']
+            if available_target_rebalance_volume >= available_source_rebalance_volume:
+                available_target_rebalance_volume = available_source_rebalance_volume
+
+            print 'Rebalance %dTB from %s(%f) to %s(%f)' % (available_target_rebalance_volume / 1E12, source_rse['rse'], source_rse['ratio'], destination_rse['rse'], destination_rse['ratio'])
+            rebalance_rse(source_rse['rse'], max_bytes=available_target_rebalance_volume, dry_run=False, comment='Nuclei Background rebalancing', force_expression=destination_rse['rse'])
+
+            destination_rse['receive_volume'] += available_target_rebalance_volume
+            total_rebalance_volume += available_target_rebalance_volume
+            available_source_rebalance_volume -= available_target_rebalance_volume
diff --git a/lib/rucio/daemons/bb8/t2_background_rebalance.py b/lib/rucio/daemons/bb8/t2_background_rebalance.py
index 4e264cbde..cb4dadc5d 100644
--- a/lib/rucio/daemons/bb8/t2_background_rebalance.py
+++ b/lib/rucio/daemons/bb8/t2_background_rebalance.py
@@ -90,24 +90,26 @@ for src in rses_over_ratio:
 
 # Loop over RSEs over the ratio
 for source_rse in rses_over_ratio:
-    if source_rse['ratio'] > global_ratio + global_ratio * tolerance:
-        available_source_rebalance_volume = int((source_rse['primary'] - global_ratio * source_rse['secondary']) / (global_ratio + 1))
-        if available_source_rebalance_volume > max_rse_rebalance_volume:
-            available_source_rebalance_volume = max_rse_rebalance_volume
-        if available_source_rebalance_volume > max_total_rebalance_volume - total_rebalance_volume:
-            available_source_rebalance_volume = max_total_rebalance_volume - total_rebalance_volume
-        # Select a target:
-        for destination_rse in rses_under_ratio:
-            if available_source_rebalance_volume > 0:
-                if destination_rse['receive_volume'] >= max_rse_rebalance_volume:
-                    continue
-                available_target_rebalance_volume = max_rse_rebalance_volume - destination_rse['receive_volume']
-                if available_target_rebalance_volume >= available_source_rebalance_volume:
-                    available_target_rebalance_volume = available_source_rebalance_volume
-
-                print 'Rebalance %dTB from %s(%f) to %s(%f)' % (available_target_rebalance_volume / 1E12, source_rse['rse'], source_rse['ratio'], destination_rse['rse'], destination_rse['ratio'])
-                rebalance_rse(source_rse['rse'], max_bytes=available_target_rebalance_volume, dry_run=False, comment='T2 Background rebalancing', force_expression=destination_rse['rse'])
-
-                destination_rse['receive_volume'] += available_target_rebalance_volume
-                total_rebalance_volume += available_target_rebalance_volume
-                available_source_rebalance_volume -= available_target_rebalance_volume
+
+    # The volume that would be rebalanced, not real availability of the data:
+    available_source_rebalance_volume = int((source_rse['primary'] - global_ratio * source_rse['secondary']) / (global_ratio + 1))
+    if available_source_rebalance_volume > max_rse_rebalance_volume:
+        available_source_rebalance_volume = max_rse_rebalance_volume
+    if available_source_rebalance_volume > max_total_rebalance_volume - total_rebalance_volume:
+        available_source_rebalance_volume = max_total_rebalance_volume - total_rebalance_volume
+
+    # Select a target:
+    for destination_rse in rses_under_ratio:
+        if available_source_rebalance_volume > 0:
+            if destination_rse['receive_volume'] >= max_rse_rebalance_volume:
+                continue
+            available_target_rebalance_volume = max_rse_rebalance_volume - destination_rse['receive_volume']
+            if available_target_rebalance_volume >= available_source_rebalance_volume:
+                available_target_rebalance_volume = available_source_rebalance_volume
+
+            print 'Rebalance %dTB from %s(%f) to %s(%f)' % (available_target_rebalance_volume / 1E12, source_rse['rse'], source_rse['ratio'], destination_rse['rse'], destination_rse['ratio'])
+            rebalance_rse(source_rse['rse'], max_bytes=available_target_rebalance_volume, dry_run=False, comment='T2 Background rebalancing', force_expression=destination_rse['rse'])
+
+            destination_rse['receive_volume'] += available_target_rebalance_volume
+            total_rebalance_volume += available_target_rebalance_volume
+            available_source_rebalance_volume -= available_target_rebalance_volume
diff --git a/lib/rucio/tests/test_archive.py b/lib/rucio/tests/test_archive.py
index f3792250d..cbcd2401f 100644
--- a/lib/rucio/tests/test_archive.py
+++ b/lib/rucio/tests/test_archive.py
@@ -16,23 +16,26 @@
 # - Vincent Garonne <vgaronne@gmail.com>, 2017-2018
 # - Martin Barisits <martin.barisits@cern.ch>, 2017
 # - Joaquin Bogado <jbogado@linti.unlp.edu.ar>, 2018
+# - Mario Lassnig <mario.lassnig@cern.ch>, 2018
 
-from nose.tools import assert_equal
+from nose.tools import assert_equal, assert_in, assert_not_in
 
 from rucio.client.didclient import DIDClient
 from rucio.client.replicaclient import ReplicaClient
-
 from rucio.common.utils import generate_uuid
+from rucio.core.replica import add_replicas
+from rucio.core.rse import add_rse, del_rse, add_protocol
+from rucio.tests.common import rse_name_generator
 
 
-class TestDIDClients(object):
+class TestArchive(object):
 
     def __init__(self):
-        self.did_client = DIDClient()
-        self.replica_client = ReplicaClient()
+        self.dc = DIDClient()
+        self.rc = ReplicaClient()
 
     def test_add_and_list_archive(self):
-        """  ARCHIVE (CLIENT): Add files to archive and list the content."""
+        """  ARCHIVE (CLIENT): Add files to archive and list the content """
         scope, rse = 'mock', 'MOCK'
         archive_files = ['file_' + generate_uuid() + '.zip' for _ in range(2)]
         files = []
@@ -44,14 +47,62 @@ class TestDIDClients(object):
                           'meta': {'guid': str(generate_uuid())}})
         for archive_file in archive_files:
 
-            self.replica_client.add_replicas(rse=rse, files=[{'scope': scope,
-                                                              'name': archive_file,
-                                                              'bytes': 1,
-                                                              'adler32': '0cc737eb'}])
+            self.rc.add_replicas(rse=rse, files=[{'scope': scope,
+                                                  'name': archive_file,
+                                                  'bytes': 1,
+                                                  'adler32': '0cc737eb'}])
 
-            self.did_client.add_files_to_archive(scope=scope, name=archive_file, files=files)
+            self.dc.add_files_to_archive(scope=scope, name=archive_file, files=files)
 
-            content = [fil for fil in self.did_client.list_archive_content(scope=scope,
-                                                                           name=archive_file)]
+            content = [f for f in self.dc.list_archive_content(scope=scope, name=archive_file)]
 
             assert_equal(len(content), 10)
+
+    def test_list_archive_contents_transparently(self):
+        """ ARCHIVE (CORE): Transparent archive listing """
+
+        scope = 'mock'
+        rse = 'APERTURE_%s' % rse_name_generator()
+        add_rse(rse)
+
+        add_protocol(rse, {'scheme': 'root',
+                           'hostname': 'root.aperture.com',
+                           'port': 1409,
+                           'prefix': '//test/chamber/',
+                           'impl': 'rucio.rse.protocols.xrootd.Default',
+                           'domains': {
+                               'lan': {'read': 1, 'write': 1, 'delete': 1},
+                               'wan': {'read': 1, 'write': 1, 'delete': 1}}})
+
+        # register archive
+        archive = {'scope': scope, 'name': 'weighted.storage.cube.zip', 'type': 'FILE',
+                   'bytes': 2596, 'adler32': 'beefdead'}
+        add_replicas(rse=rse, files=[archive], account='root')
+
+        # archived files with replicas
+        files_with_replicas = [{'scope': scope, 'name': 'witrep-%i-%s' % (i, str(generate_uuid())), 'type': 'FILE',
+                                'bytes': 1234, 'adler32': 'deadbeef'} for i in xrange(2)]
+        add_replicas(rse=rse, files=files_with_replicas, account='root')
+        self.dc.add_files_to_archive(scope=scope, name=archive['name'], files=files_with_replicas)
+
+        res = [r['pfns'] for r in self.rc.list_replicas(dids=[{'scope': scope, 'name': f['name']} for f in files_with_replicas])]
+        assert_equal(len(res), 2)
+        assert_equal(len(res[0]), 2)
+        assert_equal(len(res[1]), 2)
+        for r in res:
+            for p in r:
+                if r[p]['domain'] == 'zip':
+                    assert_in('weighted.storage.cube.zip?xrdcl.unzip=witrep-', p)
+                else:
+                    assert_not_in('weighted.storage.cube.zip?xrdcl.unzip=witrep-', p)
+
+        # archived files without replicas
+        files = [{'scope': scope, 'name': 'norep-%i-%s' % (i, str(generate_uuid())), 'type': 'FILE',
+                  'bytes': 1234, 'adler32': 'deadbeef'} for i in xrange(2)]
+        self.dc.add_files_to_archive(scope=scope, name=archive['name'], files=files)
+        res = [r['pfns'] for r in self.rc.list_replicas(dids=[{'scope': scope, 'name': f['name']} for f in files])]
+        assert_equal(len(res), 2)
+        for r in res:
+            assert_in('weighted.storage.cube.zip?xrdcl.unzip=norep-', r.keys()[0])
+
+        del_rse(rse)
diff --git a/lib/rucio/tests/test_replica_sorting.py b/lib/rucio/tests/test_replica_sorting.py
index 94ac36ed6..7c49d941a 100644
--- a/lib/rucio/tests/test_replica_sorting.py
+++ b/lib/rucio/tests/test_replica_sorting.py
@@ -99,16 +99,16 @@ class TestReplicaSorting(object):
                                                      client_location={'site': 'APERTURE'})]
         pfns = [r['pfns'] for r in replicas][0]
         assert_equal(len(pfns.keys()), 5)
+        assert_equal(pfns['root://root.aperture.com:1409//test/chamber/mock/58/b5/element_0']['domain'], 'lan')
+        assert_equal(pfns['root://root.aperture.com:1409//test/chamber/mock/58/b5/element_0']['priority'], 1)
         assert_equal(pfns['davs://davs.aperture.com:443/test/chamber/mock/58/b5/element_0']['domain'], 'lan')
         assert_equal(pfns['davs://davs.aperture.com:443/test/chamber/mock/58/b5/element_0']['priority'], 2)
-        assert_equal(pfns['davs://davs.blackmesa.com:443/lambda/complex/mock/58/b5/element_0']['domain'], 'wan')
-        assert_equal(pfns['davs://davs.blackmesa.com:443/lambda/complex/mock/58/b5/element_0']['priority'], 2)
         assert_equal(pfns['gsiftp://gsiftp.blackmesa.com:8446/lambda/complex/mock/58/b5/element_0']['domain'], 'wan')
-        assert_equal(pfns['gsiftp://gsiftp.blackmesa.com:8446/lambda/complex/mock/58/b5/element_0']['priority'], 1)
-        assert_equal(pfns['root://root.aperture.com:1409//test/chamber/mock/58/b5/element_0']['domain'], 'lan')
-        assert_equal(pfns['root://root.aperture.com:1409//test/chamber/mock/58/b5/element_0']['priority'], 1)
+        assert_equal(pfns['gsiftp://gsiftp.blackmesa.com:8446/lambda/complex/mock/58/b5/element_0']['priority'], 3)
+        assert_equal(pfns['davs://davs.blackmesa.com:443/lambda/complex/mock/58/b5/element_0']['domain'], 'wan')
+        assert_equal(pfns['davs://davs.blackmesa.com:443/lambda/complex/mock/58/b5/element_0']['priority'], 4)
         assert_equal(pfns['root://root.blackmesa.com:1409//lambda/complex/mock/58/b5/element_0']['domain'], 'wan')
-        assert_equal(pfns['root://root.blackmesa.com:1409//lambda/complex/mock/58/b5/element_0']['priority'], 3)
+        assert_equal(pfns['root://root.blackmesa.com:1409//lambda/complex/mock/58/b5/element_0']['priority'], 5)
 
         replicas = [r for r in self.rc.list_replicas(dids=[{'scope': 'mock',
                                                             'name': f['name'],
@@ -117,16 +117,16 @@ class TestReplicaSorting(object):
                                                      client_location={'site': 'BLACKMESA'})]
         pfns = [r['pfns'] for r in replicas][0]
         assert_equal(len(pfns.keys()), 5)
-        assert_equal(pfns['davs://davs.aperture.com:443/test/chamber/mock/58/b5/element_0']['domain'], 'wan')
-        assert_equal(pfns['davs://davs.aperture.com:443/test/chamber/mock/58/b5/element_0']['priority'], 2)
-        assert_equal(pfns['gsiftp://gsiftp.aperture.com:8446/test/chamber/mock/58/b5/element_0']['domain'], 'wan')
-        assert_equal(pfns['gsiftp://gsiftp.aperture.com:8446/test/chamber/mock/58/b5/element_0']['priority'], 3)
+        assert_equal(pfns['root://root.blackmesa.com:1409//lambda/complex/mock/58/b5/element_0']['domain'], 'lan')
+        assert_equal(pfns['root://root.blackmesa.com:1409//lambda/complex/mock/58/b5/element_0']['priority'], 1)
         assert_equal(pfns['gsiftp://gsiftp.blackmesa.com:8446/lambda/complex/mock/58/b5/element_0']['domain'], 'lan')
         assert_equal(pfns['gsiftp://gsiftp.blackmesa.com:8446/lambda/complex/mock/58/b5/element_0']['priority'], 2)
         assert_equal(pfns['root://root.aperture.com:1409//test/chamber/mock/58/b5/element_0']['domain'], 'wan')
-        assert_equal(pfns['root://root.aperture.com:1409//test/chamber/mock/58/b5/element_0']['priority'], 1)
-        assert_equal(pfns['root://root.blackmesa.com:1409//lambda/complex/mock/58/b5/element_0']['domain'], 'lan')
-        assert_equal(pfns['root://root.blackmesa.com:1409//lambda/complex/mock/58/b5/element_0']['priority'], 1)
+        assert_equal(pfns['root://root.aperture.com:1409//test/chamber/mock/58/b5/element_0']['priority'], 3)
+        assert_equal(pfns['davs://davs.aperture.com:443/test/chamber/mock/58/b5/element_0']['domain'], 'wan')
+        assert_equal(pfns['davs://davs.aperture.com:443/test/chamber/mock/58/b5/element_0']['priority'], 4)
+        assert_equal(pfns['gsiftp://gsiftp.aperture.com:8446/test/chamber/mock/58/b5/element_0']['domain'], 'wan')
+        assert_equal(pfns['gsiftp://gsiftp.aperture.com:8446/test/chamber/mock/58/b5/element_0']['priority'], 5)
 
         replicas = [r for r in self.rc.list_replicas(dids=[{'scope': 'mock',
                                                             'name': f['name'],
@@ -135,18 +135,19 @@ class TestReplicaSorting(object):
                                                      client_location={'site': 'XEN'})]
         pfns = [r['pfns'] for r in replicas][0]
         assert_equal(len(pfns.keys()), 6)
+        # TODO: intractable until RSE sorting is enabled
+        assert_equal(pfns['gsiftp://gsiftp.blackmesa.com:8446/lambda/complex/mock/58/b5/element_0']['domain'], 'wan')
+        assert_in(pfns['gsiftp://gsiftp.blackmesa.com:8446/lambda/complex/mock/58/b5/element_0']['priority'], [1, 2])
+        assert_equal(pfns['root://root.aperture.com:1409//test/chamber/mock/58/b5/element_0']['domain'], 'wan')
+        assert_in(pfns['root://root.aperture.com:1409//test/chamber/mock/58/b5/element_0']['priority'], [1, 2])
         assert_equal(pfns['davs://davs.aperture.com:443/test/chamber/mock/58/b5/element_0']['domain'], 'wan')
-        assert_equal(pfns['davs://davs.aperture.com:443/test/chamber/mock/58/b5/element_0']['priority'], 2)
+        assert_in(pfns['davs://davs.aperture.com:443/test/chamber/mock/58/b5/element_0']['priority'], [3, 4])
         assert_equal(pfns['davs://davs.blackmesa.com:443/lambda/complex/mock/58/b5/element_0']['domain'], 'wan')
-        assert_equal(pfns['davs://davs.blackmesa.com:443/lambda/complex/mock/58/b5/element_0']['priority'], 2)
+        assert_in(pfns['davs://davs.blackmesa.com:443/lambda/complex/mock/58/b5/element_0']['priority'], [3, 4])
         assert_equal(pfns['gsiftp://gsiftp.aperture.com:8446/test/chamber/mock/58/b5/element_0']['domain'], 'wan')
-        assert_equal(pfns['gsiftp://gsiftp.aperture.com:8446/test/chamber/mock/58/b5/element_0']['priority'], 3)
-        assert_equal(pfns['gsiftp://gsiftp.blackmesa.com:8446/lambda/complex/mock/58/b5/element_0']['domain'], 'wan')
-        assert_equal(pfns['gsiftp://gsiftp.blackmesa.com:8446/lambda/complex/mock/58/b5/element_0']['priority'], 1)
-        assert_equal(pfns['root://root.aperture.com:1409//test/chamber/mock/58/b5/element_0']['domain'], 'wan')
-        assert_equal(pfns['root://root.aperture.com:1409//test/chamber/mock/58/b5/element_0']['priority'], 1)
+        assert_in(pfns['gsiftp://gsiftp.aperture.com:8446/test/chamber/mock/58/b5/element_0']['priority'], [5, 6])
         assert_equal(pfns['root://root.blackmesa.com:1409//lambda/complex/mock/58/b5/element_0']['domain'], 'wan')
-        assert_equal(pfns['root://root.blackmesa.com:1409//lambda/complex/mock/58/b5/element_0']['priority'], 3)
+        assert_in(pfns['root://root.blackmesa.com:1409//lambda/complex/mock/58/b5/element_0']['priority'], [5, 6])
 
         ml = self.rc.list_replicas(dids=[{'scope': 'mock',
                                           'name': f['name'],
@@ -174,20 +175,20 @@ class TestReplicaSorting(object):
         assert_in('domain="wan" priority="5">gsiftp://gsiftp.aperture.com:8446/test/chamber/mock/58/b5/element_0', ml)
         assert_not_in('priority="6"', ml)
 
-        ml = self.rc.list_replicas(dids=[{'scope': 'mock',
-                                          'name': f['name'],
-                                          'type': 'FILE'} for f in self.files],
-                                   schemes=['root', 'gsiftp', 'davs'],
-                                   metalink=True,
-                                   client_location={'site': 'XEN'})
-
-        assert_in('domain="wan" priority="1">root://root.aperture.com:1409//test/chamber/mock/58/b5/element_0', ml)
-        assert_in('domain="wan" priority="2">gsiftp://gsiftp.blackmesa.com:8446/lambda/complex/mock/58/b5/element_0', ml)
-        assert_in('domain="wan" priority="3">davs://davs.aperture.com:443/test/chamber/mock/58/b5/element_0', ml)
-        assert_in('domain="wan" priority="4">davs://davs.blackmesa.com:443/lambda/complex/mock/58/b5/element_0', ml)
-        assert_in('domain="wan" priority="5">gsiftp://gsiftp.aperture.com:8446/test/chamber/mock/58/b5/element_0', ml)
-        assert_in('domain="wan" priority="6">root://root.blackmesa.com:1409//lambda/complex/mock/58/b5/element_0', ml)
-        assert_not_in('priority="7"', ml)
+        # TODO: intractable until RSE sorting is enabled
+        # ml = self.rc.list_replicas(dids=[{'scope': 'mock',
+        #                                   'name': f['name'],
+        #                                   'type': 'FILE'} for f in self.files],
+        #                            schemes=['root', 'gsiftp', 'davs'],
+        #                            metalink=True,
+        #                            client_location={'site': 'XEN'})
+        # assert_in('domain="wan" priority="1">root://root.aperture.com:1409//test/chamber/mock/58/b5/element_0', ml)
+        # assert_in('domain="wan" priority="2">gsiftp://gsiftp.blackmesa.com:8446/lambda/complex/mock/58/b5/element_0', ml)
+        # assert_in('domain="wan" priority="3">davs://davs.aperture.com:443/test/chamber/mock/58/b5/element_0', ml)
+        # assert_in('domain="wan" priority="4">davs://davs.blackmesa.com:443/lambda/complex/mock/58/b5/element_0', ml)
+        # assert_in('domain="wan" priority="5">gsiftp://gsiftp.aperture.com:8446/test/chamber/mock/58/b5/element_0', ml)
+        # assert_in('domain="wan" priority="6">root://root.blackmesa.com:1409//lambda/complex/mock/58/b5/element_0', ml)
+        # assert_not_in('priority="7"', ml)
 
         delete_replicas(rse=self.rse1, files=self.files)
         delete_replicas(rse=self.rse2, files=self.files)
diff --git a/lib/rucio/web/rest/webpy/v1/replica.py b/lib/rucio/web/rest/webpy/v1/replica.py
index ced35b6c3..316acb964 100755
--- a/lib/rucio/web/rest/webpy/v1/replica.py
+++ b/lib/rucio/web/rest/webpy/v1/replica.py
@@ -315,7 +315,7 @@ class ListReplicas(RucioController):
             if 'signature_lifetime' in params:
                 signature_lifetime = params['signature_lifetime']
             else:
-                # hardcoded default of 1h if config is not parseable
+                # hardcoded default of 10 minutes if config is not parseable
                 signature_lifetime = config_get('credentials', 'signature_lifetime', raise_exception=False, default=600)
 
         except ValueError:
@@ -348,7 +348,6 @@ class ListReplicas(RucioController):
                                        client_location=client_location,
                                        domain=domain, signature_lifetime=signature_lifetime,
                                        issuer=ctx.env.get('issuer')):
-
                 if not metalink:
                     yield dumps(rfile, cls=APIEncoder) + '\\n'
                 else:
@@ -360,15 +359,6 @@ class ListReplicas(RucioController):
                             replicas.append(replica)
                             dictreplica[replica] = (rfile['pfns'][replica]['domain'], rfile['pfns'][replica]['priority'], rse)
 
-                    # we need to sort first by domain
-                    # --> exploit that L(AN) comes before W(AN) in the alphabet :-D
-                    tmp = sorted([(dictreplica[r] + (r,)) for r in dictreplica])
-
-                    # then rewrite priorities in ascending order starting from 1 across both domains to be correct metalink
-                    dictreplica = {}
-                    for i in xrange(len(tmp)):
-                        dictreplica[tmp[i][3]] = (tmp[i][0], i + 1, tmp[i][2])
-
                     yield ' <file name="' + rfile['name'] + '">\\n'
                     yield '  <identity>' + rfile['scope'] + ':' + rfile['name'] + '</identity>\\n'
                     if rfile['adler32'] is not None: