Injector can crash when split_container option is enabled
Motivation
----------
The `split_container` option is enabled when requests on containers are done via R2D2. It creates rules on the constituent datasets of the container instead of a single rule on the container. 
- It can trigger `IntegrityError` in the case a rule for scope, name, account, rse_expression already exists. 
- Similarly the same error can be raised if 2 constituents contain the same datasets.

Modification
------------
To handle the first issue, the injector must check if a rule on scope, name, account, rse_expression already exists. To handle the second one, it should make a set from the list returned by `list_child_datasets`.


diff --git a/lib/rucio/core/replica.py b/lib/rucio/core/replica.py
index 89822b021..22a0fc27c 100644
--- a/lib/rucio/core/replica.py
+++ b/lib/rucio/core/replica.py
@@ -22,6 +22,7 @@
 # - Thomas Beermann <thomas.beermann@cern.ch>, 2014-2018
 # - Wen Guan <wguan.icedew@gmail.com>, 2014-2015
 # - Hannes Hansen <hannes.jakob.hansen@cern.ch>, 2018
+# - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2019
 #
 # PY3K COMPATIBLE
 
@@ -418,7 +419,7 @@ def get_pfn_to_rse(pfns, session=None):
     se_condition = []
     dict_rse = {}
     surls = clean_surls(pfns)
-    scheme = surls[0].split(':')[0]
+    scheme = surls[0].split(':')[0] if surls else None
     for surl in surls:
         if surl.split(':')[0] != scheme:
             raise exception.InvalidType('The PFNs specified must have the same protocol')
diff --git a/lib/rucio/core/rule.py b/lib/rucio/core/rule.py
index 478f01166..871201db5 100644
--- a/lib/rucio/core/rule.py
+++ b/lib/rucio/core/rule.py
@@ -605,6 +605,9 @@ def inject_rule(rule_id, session=None):
         logging.debug("Creating dataset rules for Split Container rule %s" % (str(rule.id)))
         # Get all child datasets and put rules on them
         dids = [{'scope': dataset['scope'], 'name': dataset['name']} for dataset in rucio.core.did.list_child_datasets(scope=rule.scope, name=rule.name, session=session)]
+        # Remove duplicates from the list of dictionaries
+        dids = [dict(t) for t in {tuple(d.items()) for d in dids}]
+        # Remove dids which already have a similar rule
         dids = [did for did in dids if session.query(models.ReplicationRule).filter_by(scope=did['scope'], name=did['name'], account=rule.account, rse_expression=rule.rse_expression).count() == 0]
         if rule.expires_at:
             lifetime = (rule.expires_at - datetime.utcnow()).days * 24 * 3600 + (rule.expires_at - datetime.utcnow()).seconds
diff --git a/lib/rucio/daemons/auditor/__init__.py b/lib/rucio/daemons/auditor/__init__.py
index 2a1d71a1a..993260e7f 100644
--- a/lib/rucio/daemons/auditor/__init__.py
+++ b/lib/rucio/daemons/auditor/__init__.py
@@ -17,7 +17,7 @@
 # - 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
+# - Dimitrios Christidis <dimitrios.christidis@cern.ch>, 2018-2019
 
 import Queue
 import bz2
@@ -35,9 +35,11 @@ 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.replica import declare_bad_file_replicas, list_replicas
 from rucio.core.rse import get_rse_usage
 from rucio.daemons.auditor.hdfs import ReplicaFromHDFS
 from rucio.daemons.auditor import srmdumps
+from rucio.db.sqla.constants import BadFilesStatus
 
 
 def consistency(rse, delta, configuration, cache_dir, results_dir):
@@ -118,8 +120,8 @@ 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.
+    may be deleted by the Dark Reaper.  LOST files are reported as
+    suspicious so that they may be further checked by the cloud squads.
 
     ``output`` should be an ``str`` with the absolute path to the file
     produced by ``consistency()``.  It must maintain its naming
@@ -133,18 +135,19 @@ def process_output(output, sanity_check=True, compress=True):
     """
     logger = logging.getLogger('auditor-worker')
     dark_replicas = []
+    lost_replicas = []
     try:
         with open(output) as f:
             for line in f:
                 label, path = line.rstrip().split(',', 1)
+                scope, name = guess_replica_info(path)
                 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
+                    lost_replicas.append({'scope': scope,
+                                          'name': name})
                 else:
                     raise ValueError('unexpected label')
     # Since the file is read immediately after its creation, any error
@@ -160,12 +163,30 @@ def process_output(output, sanity_check=True, compress=True):
     # 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')
+    found_error = False
+    if len(dark_replicas) > threshold * usage['files']:
+        logger.warning('Number of DARK files is exceeding threshold: "%s"',
+                       output)
+        found_error = True
+    if len(lost_replicas) > threshold * usage['files']:
+        logger.warning('Number of LOST files is exceeding threshold: "%s"',
+                       output)
+        found_error = True
+    if found_error and sanity_check:
+        raise AssertionError('sanity check failed')
+
+    # While converting LOST replicas to PFNs, entries that do not
+    # correspond to a replica registered in Rucio are silently dropped.
+    lost_pfns = [r['rses'][rse][0] for r in list_replicas(lost_replicas)
+                 if rse in r['rses']]
 
     add_quarantined_replicas(rse, dark_replicas)
     logger.debug('Processed %d DARK files from "%s"', len(dark_replicas),
                  output)
+    declare_bad_file_replicas(lost_pfns, reason='Reported by Auditor',
+                              issuer='root', status=BadFilesStatus.SUSPICIOUS)
+    logger.debug('Processed %d LOST files from "%s"', len(lost_replicas),
+                 output)
 
     if compress:
         destination = bz2_compress_file(output)
['lib/rucio/core/replica.py', 'lib/rucio/daemons/auditor/__init__.py', 'lib/rucio/core/rule.py']