Repository: rucio/rucio
Issue:  listing archived replicas returns wrong filemeta if RSE expression

Motivation
----------

If the client requests a specific RSE expression then the returned file data (checksum, bytes, ...) is from the archive, not the constituent.

Changed files:
lib/rucio/core/quarantined_replica.py, lib/rucio/daemons/auditor/__init__.py, lib/rucio/core/replica.py, lib/rucio/tests/test_auditor.py

diff --git a/lib/rucio/core/quarantined_replica.py b/lib/rucio/core/quarantined_replica.py
index 8afc49c2f..e416e1320 100644
--- a/lib/rucio/core/quarantined_replica.py
+++ b/lib/rucio/core/quarantined_replica.py
@@ -1,16 +1,24 @@
-"""
-  Copyright European Organization for Nuclear Research (CERN)
-
-  Licensed under the Apache License, Version 2.0 (the "License");
-  You may not use this file except in compliance with the License.
-  You may obtain a copy of the License at
-  http://www.apache.org/licenses/LICENSE-2.0
-
-  Authors:
-  - Vincent Garonne, <vincent.garonne@cern.ch>, 2016-2017
-
-  PY3K COMPATIBLE
-"""
+#!/usr/bin/env python
+# Copyright 2016-2018 CERN for the benefit of the ATLAS collaboration.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Authors:
+# - Vincent Garonne <vgaronne@gmail.com>, 2016-2017
+# - Hannes Hansen <hannes.jakob.hansen@cern.ch>, 2018
+# - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2018
+#
+# PY3K COMPATIBLE
 
 import datetime
 
@@ -28,37 +36,45 @@ def add_quarantined_replicas(rse, replicas, session=None):
     """
     Bulk add quarantined file replicas.
 
-    :param rse:     The rse name.
-    :param files:   The list of files.
-    :param session: The database session in use.
-
-    :returns: True is successful.
+    :param rse:      The rse name.
+    :param replicas: A list of dicts with the replica information.
+    :param session:  The database session in use.
     """
     rse_id = get_rse_id(rse, session=session)
 
-    file_clause = []
-    for replica in replicas:
-        file_clause.append(and_(models.RSEFileAssociation.scope == replica.get('scope', None),
-                                models.RSEFileAssociation.name == replica.get('name', None),
-                                models.RSEFileAssociation.rse_id == rse_id))
-
-    if file_clause:
+    for chunk in chunks(replicas, 100):
+        # Exlude files that have a registered replica.  This is a
+        # safeguard against potential issues in the Auditor.
+        file_clause = []
+        for replica in chunk:
+            file_clause.append(and_(models.RSEFileAssociation.scope == replica.get('scope', None),
+                                    models.RSEFileAssociation.name == replica.get('name', None),
+                                    models.RSEFileAssociation.rse_id == rse_id))
         file_query = session.query(models.RSEFileAssociation.scope,
                                    models.RSEFileAssociation.name,
                                    models.RSEFileAssociation.rse_id).\\
             with_hint(models.RSEFileAssociation, "index(REPLICAS REPLICAS_PK)", 'oracle').\\
             filter(or_(*file_clause))
-
         existing_replicas = [(scope, name, rseid) for scope, name, rseid in file_query]
-
-        replicas = [replica for replica in replicas if (replica.get('scope', None), replica.get('name', None), rse_id) not in existing_replicas]
-
-        for files in chunks(replicas, 1000):
-            session.bulk_insert_mappings(
-                models.QuarantinedReplica,
-                [{'rse_id': rse_id, 'path': file['path'],
-                  'scope': file.get('scope'), 'name': file.get('name'),
-                  'bytes': file.get('bytes')} for file in files])
+        chunk = [replica for replica in chunk if (replica.get('scope', None), replica.get('name', None), rse_id) not in existing_replicas]
+
+        # Exclude files that have already been added to the quarantined
+        # replica table.
+        quarantine_clause = []
+        for replica in chunk:
+            quarantine_clause.append(and_(models.QuarantinedReplica.path == replica['path'],
+                                          models.QuarantinedReplica.rse_id == rse_id))
+        quarantine_query = session.query(models.QuarantinedReplica.path,
+                                         models.QuarantinedReplica.rse_id).\\
+            filter(or_(*quarantine_clause))
+        quarantine_replicas = [(path, rseid) for path, rseid in quarantine_query]
+        chunk = [replica for replica in chunk if (replica['path'], rse_id) not in quarantine_replicas]
+
+        session.bulk_insert_mappings(
+            models.QuarantinedReplica,
+            [{'rse_id': rse_id, 'path': file['path'],
+              'scope': file.get('scope'), 'name': file.get('name'),
+              'bytes': file.get('bytes')} for file in chunk])
 
 
 @transactional_session
diff --git a/lib/rucio/core/replica.py b/lib/rucio/core/replica.py
index e20ba3488..1bdd141af 100644
--- a/lib/rucio/core/replica.py
+++ b/lib/rucio/core/replica.py
@@ -39,6 +39,7 @@ from sqlalchemy.exc import DatabaseError, IntegrityError
 from sqlalchemy.orm.exc import FlushError, NoResultFound
 from sqlalchemy.sql.expression import case, bindparam, select, text, false, true
 
+import rucio.core.did
 import rucio.core.lock
 
 from rucio.common import exception
@@ -783,14 +784,27 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns,
                 # 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:
 
+                    # retrieve the constituents metadata so we can downport it later
+                    # otherwise the zip meta will be used which won't match the actual file
+                    # full name used due to circular import dependency between modules
+                    constituent_meta = rucio.core.did.get_metadata(scope, name, session=session)
+
+                    for archive in archive_result:
                         # RSE expression might limit the archives we're allowed to use
                         # if it's empty due to non-matching RSE expression we must skip
                         if archive['pfns'] == {}:
                             continue
 
+                        # downport the constituents meta, we are not interested in the archives meta
+                        archive['scope'] = scope
+                        archive['name'] = name
+                        archive['adler32'] = constituent_meta['adler32']
+                        archive['md5'] = constituent_meta['md5']
+                        archive['bytes'] = constituent_meta['bytes']
+
                         for tmp_archive in archive['pfns']:
+                            # declare the constituent as being in a zip file
                             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:
@@ -1010,6 +1024,7 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns,
                 file['rses'][t_rse].append(t_pfn)
             else:
                 file['rses'][t_rse] = [t_pfn]
+
         yield file
         file = {}
 
diff --git a/lib/rucio/daemons/auditor/__init__.py b/lib/rucio/daemons/auditor/__init__.py
index c227f1a21..3c5c8192e 100644
--- a/lib/rucio/daemons/auditor/__init__.py
+++ b/lib/rucio/daemons/auditor/__init__.py
@@ -17,11 +17,13 @@
 # - Martin Barisits <martin.barisits@cern.ch>, 2017
 # - Mario Lassnig <mario.lassnig@cern.ch>, 2017
 # - Vincent Garonne <vgaronne@gmail.com>, 2018
+# - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2018
 
 import Queue
+import bz2
 import glob
 import logging
-import os.path
+import os
 import select
 import sys
 
@@ -32,23 +34,20 @@ from rucio.common.dumper import LogPipeHandler
 from rucio.common.dumper import mkdir
 from rucio.common.dumper import temp_file
 from rucio.common.dumper.consistency import Consistency
+from rucio.core.quarantined_replica import add_quarantined_replicas
+from rucio.core.rse import get_rse_usage
 from rucio.daemons.auditor.hdfs import ReplicaFromHDFS
 from rucio.daemons.auditor import srmdumps
 
 
-def total_seconds(td):
-    '''timedelta.total_seconds() for Python < 2.7'''
-    return (td.microseconds + (td.seconds + td.days * 24 * 3600) * (10 ** 6)) / float(10 ** 6)
-
-
 def consistency(rse, delta, configuration, cache_dir, results_dir):
     logger = logging.getLogger('auditor-worker')
     rsedump, rsedate = srmdumps.download_rse_dump(rse, configuration, destdir=cache_dir)
-    results_path = '{0}/{1}_{2}'.format(results_dir, rse, rsedate.strftime('%Y%m%d'))  # pylint: disable=no-member
+    results_path = os.path.join(results_dir, '{0}_{1}'.format(rse, rsedate.strftime('%Y%m%d')))  # pylint: disable=no-member
 
-    if os.path.exists(results_path):
+    if os.path.exists(results_path + '.bz2') or os.path.exists(results_path):
         logger.warn('Consistency check for "%s" (dump dated %s) already done, skipping check', rse, rsedate.strftime('%Y%m%d'))  # pylint: disable=no-member
-        return
+        return None
 
     rrdump_prev = ReplicaFromHDFS.download(rse, rsedate - delta, cache_dir=cache_dir)
     rrdump_next = ReplicaFromHDFS.download(rse, rsedate + delta, cache_dir=cache_dir)
@@ -66,6 +65,112 @@ def consistency(rse, delta, configuration, cache_dir, results_dir):
         for result in results:
             output.write('{0}\\n'.format(result.csv()))
 
+    return results_path
+
+
+def guess_replica_info(path):
+    """Try to extract the scope and name from a path
+
+    ``path`` should be an ``str`` with the relative path to the file on
+    the RSE.
+
+    Returns a ``tuple`` of which the first element is the scope of the
+    replica and the second element is its name.
+    """
+    items = path.split('/')
+    if len(items) == 1:
+        return None, path
+    elif len(items) > 2 and items[0] in ['group', 'user']:
+        return '.'.join(items[0:2]), items[-1]
+    else:
+        return items[0], items[-1]
+
+
+def bz2_compress_file(source, chunk_size=65000):
+    """Compress a file with bzip2
+
+    The destination is the path passed through ``source`` extended with
+    '.bz2'.  The original file is deleted.
+
+    Errors are deliberately not handled gracefully.  Any exceptions
+    should be propagated to the caller.
+
+    ``source`` should be an ``str`` with the absolute path to the file
+    to compress.
+
+    ``chunk_size`` should be an ``int`` with the size (in bytes) of the
+    chunks by which to read the file.
+
+    Returns an ``str`` with the destination path.
+    """
+    destination = '{}.bz2'.format(source)
+    with open(source) as plain, bz2.BZ2File(destination, 'w') as compressed:
+        while True:
+            chunk = plain.read(chunk_size)
+            if not chunk:
+                break
+            compressed.write(chunk)
+    os.remove(source)
+    return destination
+
+
+def process_output(output, sanity_check=True, compress=True):
+    """Perform post-consistency-check actions
+
+    DARK files are put in the quarantined-replica table so that they
+    may be deleted by the Dark Reaper.  LOST files are currently
+    ignored.
+
+    ``output`` should be an ``str`` with the absolute path to the file
+    produced by ``consistency()``.  It must maintain its naming
+    convention.
+
+    If ``sanity_check`` is ``True`` (default) and the number of entries
+    in the output file is deemed excessive, the actions are aborted.
+
+    If ``compress`` is ``True`` (default), the file is compressed with
+    bzip2 after the actions are successfully performed.
+    """
+    logger = logging.getLogger('auditor-worker')
+    dark_replicas = []
+    try:
+        with open(output) as f:
+            for line in f:
+                label, path = line.rstrip().split(',', 1)
+                if label == 'DARK':
+                    scope, name = guess_replica_info(path)
+                    dark_replicas.append({'path': path,
+                                          'scope': scope,
+                                          'name': name})
+                elif label == 'LOST':
+                    # TODO: Declare LOST files as suspicious.
+                    pass
+                else:
+                    raise ValueError('unexpected label')
+    # Since the file is read immediately after its creation, any error
+    # exposes a bug in the Auditor.
+    except Exception as error:
+        logger.critical('Error processing "%s"', output, exc_info=True)
+        raise error
+
+    rse = os.path.basename(output[:output.rfind('_')])
+    usage = get_rse_usage(rse, source='rucio')[0]
+    threshold = config.config_get('auditor', 'threshold', False, 0.2)
+
+    # Perform a basic sanity check by comparing the number of entries
+    # with the total number of files on the RSE.  If the percentage is
+    # significant, there is most likely an issue with the site dump.
+    if sanity_check and len(dark_replicas) > threshold * usage['files']:
+        raise AssertionError('number of DARK files is exceeding threshold')
+
+    add_quarantined_replicas(rse, dark_replicas)
+    logger.debug('Processed %d DARK files from "%s"', len(dark_replicas),
+                 output)
+
+    if compress:
+        destination = bz2_compress_file(output)
+        logger.debug('Compressed "%s"', destination)
+
 
 def check(queue, retry, terminate, logpipe, cache_dir, results_dir, keep_dumps, delta_in_days):
     logger = logging.getLogger('auditor-worker')
@@ -97,13 +202,16 @@ def check(queue, retry, terminate, logpipe, cache_dir, results_dir, keep_dumps,
         start = datetime.now()
         try:
             logger.debug('Checking "%s"', rse)
-            consistency(rse, delta, configuration, cache_dir, results_dir)
+            output = consistency(rse, delta, configuration, cache_dir,
+                                 results_dir)
+            if output:
+                process_output(output)
         except:
             success = False
         else:
             success = True
         finally:
-            elapsed = total_seconds(datetime.now() - start) / 60
+            elapsed = (datetime.now() - start).total_seconds() / 60
             if success:
                 logger.info('SUCCESS checking "%s" in %d minutes', rse, elapsed)
             else:
diff --git a/lib/rucio/tests/test_auditor.py b/lib/rucio/tests/test_auditor.py
index 7ad090dca..d5f62cbb7 100644
--- a/lib/rucio/tests/test_auditor.py
+++ b/lib/rucio/tests/test_auditor.py
@@ -19,15 +19,54 @@ from rucio.daemons import auditor
 from rucio.daemons.auditor import srmdumps
 from rucio.daemons.auditor import hdfs
 from rucio.tests.common import stubbed
+import bz2
 import collections
 import multiprocessing
+import os
 import tempfile
 
 
-def test_total_seconds():
-    """AUTHENTICATION (REST): Username and password (wrong credentials)."""
-    dif = timedelta(days=1, hours=1, minutes=1, seconds=1, microseconds=1)
-    ok_(abs(auditor.total_seconds(dif) - 90061) < 0.01)
+def test_auditor_guess_replica_info():
+    tests = {
+        'foo': (None, 'foo'),
+        'foo/bar': ('foo', 'bar'),
+        'foo/bar/baz': ('foo', 'baz'),
+        'user': (None, 'user'),
+        'user/foo': ('user', 'foo'),
+        'user/foo/bar': ('user.foo', 'bar'),
+        'group': (None, 'group'),
+        'group/foo': ('group', 'foo'),
+        'group/foo/bar': ('group.foo', 'bar'),
+    }
+    for input_, output in tests.items():
+        eq_(auditor.guess_replica_info(input_), output)
+
+
+class TestBz2CompressFile(object):
+
+    def setup(self):
+        self.source = tempfile.mktemp()
+        self.destination = self.source + '.bz2'
+
+    def teardown(self):
+        for path in [self.source, self.destination]:
+            try:
+                os.remove(path)
+            except OSError:
+                pass
+
+    def test_auditor_bz2_compress_file(self):
+        test_data = 'foo'
+        with open(self.source, 'w') as f:
+            f.write(test_data)
+
+        destination = auditor.bz2_compress_file(self.source)
+
+        eq_(destination, self.destination)
+        ok_(os.path.exists(self.destination))
+        ok_(not os.path.exists(self.source))
+        with bz2.BZ2File(self.destination) as f:
+            eq_(f.read(), test_data)
 
 
 def test_auditor_download_dumps_with_expected_dates():