Repository: rucio/rucio
Issue: adler32 function should open file in "with" env and propagate the exception correctly

Motivation
----------
The function to get the adler32 checksum of a given file:
https://github.com/rucio/rucio/blob/master/lib/rucio/common/utils.py#L172
doesn't close the file correctly if an exception is raised and also doesn't propagate the exception to the caller.

Modification
------------


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

diff --git a/lib/rucio/common/utils.py b/lib/rucio/common/utils.py
index 0e708d71c..fe7695560 100644
--- a/lib/rucio/common/utils.py
+++ b/lib/rucio/common/utils.py
@@ -24,6 +24,8 @@
 # - Brian Bockelman <bbockelm@cse.unl.edu>, 2018
 # - Tobias Wegner <twegner@cern.ch>, 2018
 # - Hannes Hansen <hannes.jakob.hansen@cern.ch>, 2018
+#
+# PY3K COMPATIBLE
 
 from __future__ import print_function
 
@@ -39,13 +41,13 @@ 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
+from six import string_types
 
 try:
     # Python 2
@@ -180,11 +182,11 @@ def adler32(file):
     adler = 1
 
     try:
-        openFile = open(file, 'rb')
-        for line in openFile:
-            adler = zlib.adler32(line, adler)
-    except Exception:
-        raise Exception('FATAL - could not get checksum of file %s' % file)
+        with open(file, 'rb') as openFile:
+            for line in openFile:
+                adler = zlib.adler32(line, adler)
+    except Exception as e:
+        raise Exception('FATAL - could not get Adler32 checksum of file %s - %s' % (file, e))
 
     # backflip on 32bit
     if adler < 0:
@@ -203,9 +205,9 @@ def md5(file):
     hash_md5 = hashlib.md5()
     try:
         with open(file, "rb") as f:
-            map(hash_md5.update, iter(lambda: f.read(4096), b""))
-    except Exception:
-        raise Exception('FATAL - could not get MD5 checksum of file %s' % file)
+            list(map(hash_md5.update, iter(lambda: f.read(4096), b"")))
+    except Exception as e:
+        raise Exception('FATAL - could not get MD5 checksum of file %s - %s' % (file, e))
 
     return hash_md5.hexdigest()
 
@@ -260,12 +262,8 @@ def render_json_list(l):
 def datetime_parser(dct):
     """ datetime parser
     """
-    try:
-        varType = basestring
-    except NameError:
-        varType = str
     for k, v in list(dct.items()):
-        if isinstance(v, varType) and re.search(" UTC", v):
+        if isinstance(v, string_types) and re.search(" UTC", v):
             try:
                 dct[k] = datetime.datetime.strptime(v, DATE_FORMAT)
             except Exception:
@@ -722,7 +720,7 @@ def add_url_query(url, query):
     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)
+    url_parts[4] = urlencode(mod_query)
     return urlparse.urlunparse(url_parts)
 
 
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/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():
diff --git a/lib/rucio/tests/test_utils.py b/lib/rucio/tests/test_utils.py
index d034ee1a5..efbd15b6c 100644
--- a/lib/rucio/tests/test_utils.py
+++ b/lib/rucio/tests/test_utils.py
@@ -7,14 +7,17 @@
 
  Authors:
  - Frank Berghaus, <frank.berghaus@cern.ch>, 2018
+ - Hannes Hansen, <hannes.jakob.hansen@cern.ch>, 2018
+
+ PY3K COMPATIBLE
 '''
 
 import unittest
 import tempfile
 
-from nose.tools import assert_equal, assert_is_instance, assert_is_not_none
+from nose.tools import assert_raises, assert_equal, assert_is_instance, assert_is_not_none
 from re import match
-from rucio.common.utils import md5
+from rucio.common.utils import md5, adler32
 
 
 class TestUtils(unittest.TestCase):
@@ -37,3 +40,18 @@ class TestUtils(unittest.TestCase):
         assert_is_not_none(match('[a-fA-F0-9]{32}', ret), msg="String returned by utils.md5 is not a md5 hex digest")
         assert_equal(ret, '31d50dd6285b9ff9f8611d0762265d04',
                      msg="Hex digest returned by utils.md5 is the MD5 checksum")
+
+        with assert_raises(Exception) as e:
+            md5('no_file')
+        assert_equal('FATAL - could not get MD5 checksum of file no_file - [Errno 2] No such file or directory: \\'no_file\\'', e.exception.message)
+
+    def test_utils_adler32(self):
+        """(COMMON/UTILS): test calculating Adler32 of a file"""
+        ret = adler32(self.temp_file_1.name)
+        assert_is_instance(ret, str)
+        assert_is_not_none(match('[a-fA-F0-9]', ret))
+        assert_equal(ret, '198d03ff')
+
+        with assert_raises(Exception) as e:
+            adler32('no_file')
+        assert_equal('FATAL - could not get Adler32 checksum of file no_file - [Errno 2] No such file or directory: \\'no_file\\'', e.exception.message)