Repository: rucio/rucio
Issue: Suspicious file reporting broken in Kronos

Motivation
----------
The reporting of suspicious files in Kronos is broken


Changed files:
lib/rucio/api/replica.py, lib/rucio/daemons/tracer/kronos.py, lib/rucio/core/replica.py, lib/rucio/web/rest/webpy/v1/replica.py, lib/rucio/client/replicaclient.py

diff --git a/lib/rucio/api/replica.py b/lib/rucio/api/replica.py
index ef99a1d03..a8c6e057d 100644
--- a/lib/rucio/api/replica.py
+++ b/lib/rucio/api/replica.py
@@ -91,7 +91,8 @@ def get_did_from_pfns(pfns, rse):
 
 def list_replicas(dids, schemes=None, unavailable=False, request_id=None,
                   ignore_availability=True, all_states=False, rse_expression=None,
-                  client_location=None, domain=None, signature_lifetime=None, resolve_archives=True, issuer=None):
+                  client_location=None, domain=None, signature_lifetime=None,
+                  resolve_archives=True, resolve_parents=False, issuer=None):
     """
     List file replicas for a list of data identifiers.
 
@@ -105,6 +106,7 @@ def list_replicas(dids, schemes=None, unavailable=False, request_id=None,
     :param domain: The network domain for the call, either None, 'wan' or 'lan'. Compatibility fallback: None falls back to 'wan'.
     :param signature_lifetime: If supported, in seconds, restrict the lifetime of the signed PFN.
     :param resolve_archives: When set to True, find archives which contain the replicas.
+    :param resolve_parents: When set to True, find all parent datasets which contain the replicas.
     :param issuer: The issuer account.
     """
     validate_schema(name='r_dids', obj=dids)
@@ -120,7 +122,8 @@ def list_replicas(dids, schemes=None, unavailable=False, request_id=None,
                                  ignore_availability=ignore_availability,
                                  all_states=all_states, rse_expression=rse_expression,
                                  client_location=client_location, domain=domain,
-                                 sign_urls=sign_urls, signature_lifetime=signature_lifetime, resolve_archives=resolve_archives)
+                                 sign_urls=sign_urls, signature_lifetime=signature_lifetime,
+                                 resolve_archives=resolve_archives, resolve_parents=resolve_parents)
 
 
 def add_replicas(rse, files, issuer, ignore_availability=False):
diff --git a/lib/rucio/client/replicaclient.py b/lib/rucio/client/replicaclient.py
index 1f7eb5bb9..8f27ce4ff 100644
--- a/lib/rucio/client/replicaclient.py
+++ b/lib/rucio/client/replicaclient.py
@@ -95,7 +95,7 @@ class ReplicaClient(BaseClient):
     def list_replicas(self, dids, schemes=None, unavailable=False,
                       all_states=False, metalink=False, rse_expression=None,
                       client_location=None, sort=None, domain=None,
-                      resolve_archives=True):
+                      resolve_archives=True, resolve_parents=False):
         """
         List file replicas for a list of data identifiers (DIDs).
 
@@ -112,6 +112,7 @@ class ReplicaClient(BaseClient):
                                         ``dynamic`` - Rucio Dynamic Smart Sort (tm)
         :param domain: Define the domain. None is fallback to 'wan', otherwise 'wan, 'lan', or 'all'
         :param resolve_archives: When set to True, find archives which contain the replicas.
+        :param resolve_parents: When set to True, find all parent datasets which contain the replicas.
 
         :returns: A list of dictionaries with replica information.
 
@@ -136,6 +137,8 @@ class ReplicaClient(BaseClient):
 
         data['resolve_archives'] = resolve_archives
 
+        data['resolve_parents'] = resolve_parents
+
         url = build_url(choice(self.list_hosts),
                         path='/'.join([self.REPLICAS_BASEURL, 'list']))
 
diff --git a/lib/rucio/core/replica.py b/lib/rucio/core/replica.py
index fb5cd7cb7..62e07d78a 100644
--- a/lib/rucio/core/replica.py
+++ b/lib/rucio/core/replica.py
@@ -743,7 +743,8 @@ def _list_replicas_for_files(file_clause, state_clause, files, rse_clause, sessi
 
 def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns,
                    schemes, files, rse_clause, rse_expression, client_location, domain,
-                   sign_urls, signature_lifetime, constituents, session):
+                   sign_urls, signature_lifetime, constituents, resolve_parents,
+                   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)]
@@ -762,6 +763,7 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns,
                 pass  # do not hard fail if site cannot be resolved or is empty
 
     file, tmp_protocols, rse_info, pfns_cache = {}, {}, {}, {}
+
     for replicas in filter(None, files):
         for scope, name, bytes, md5, adler32, path, state, rse, rse_type, volatile in replicas:
 
@@ -770,6 +772,12 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns,
             # reset the domain selection to original user's choice (as this could get overwritten each iteration)
             domain = deepcopy(original_domain)
 
+            # resolve the parents only once per file, not per replica
+            # full name used due to circular import dependency between modules
+            if resolve_parents and 'parents' not in file:
+                file['parents'] = ['%s:%s' % (parent['scope'], parent['name'])
+                                   for parent in rucio.core.did.list_parent_dids(scope, name, session=session)]
+
             # if the file is a constituent, find the available archives and add them to the list of possible PFNs
             # taking into account the original rse_expression
             if '%s:%s' % (scope, name) in constituents:
@@ -814,12 +822,16 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns,
                                 tmp_archive = new_pfn
                             if 'xrdcl.unzip' not in tmp_archive:
                                 archive['pfns'][tmp_archive]['client_extract'] = True
-                        yield archive
-                    continue
+
+                            pfns.append((tmp_archive, 'zip',
+                                         archive['pfns'][tmp_archive]['priority'],
+                                         archive['pfns'][tmp_archive]['client_extract'],
+                                         archive['pfns'][tmp_archive]))
 
                 # 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 if r['pfns'] != {}]
+
                 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
@@ -989,10 +1001,11 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns,
                     file = {}
 
             if not ('scope' in file and 'name' in file):
-                file = {'scope': scope, 'name': name, 'bytes': bytes,
-                        'md5': md5, 'adler32': adler32,
-                        'pfns': {}, 'rses': defaultdict(list),
-                        'states': {rse: str(state)}}
+                file['scope'], file['name'] = scope, name
+                file['bytes'], file['md5'], file['adler32'] = bytes, md5, adler32
+                file['pfns'], file['rses'] = {}, defaultdict(list)
+                file['states'] = {rse: str(state)}
+
                 if rse:
                     # extract properly the pfn from the tuple
                     file['rses'][rse] = list(set([tmp_pfn[0] for tmp_pfn in pfns]))
@@ -1003,6 +1016,15 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns,
                                                     'domain': tmp_pfn[1],
                                                     'priority': tmp_pfn[2],
                                                     'client_extract': tmp_pfn[3]}
+                else:
+                    # extract properly the pfn from the tuple of the rse-expression restricted archive
+                    for tmp_pfn in pfns:
+                        file['pfns'][tmp_pfn[0]] = {'rse': tmp_pfn[4]['rse'],
+                                                    'type': tmp_pfn[4]['type'],
+                                                    'volatile': tmp_pfn[4]['volatile'],
+                                                    'domain': tmp_pfn[1],
+                                                    'priority': tmp_pfn[2],
+                                                    'client_extract': tmp_pfn[3]}
 
     # set the total order for the priority
     # --> exploit that L(AN) comes before W(AN) before Z(IP) alphabetically
@@ -1015,6 +1037,11 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns,
     if 'scope' in file and 'name' in file:
         file['rses'] = {}
 
+        # also do it for the last replica if necessary
+        if resolve_parents and 'parents' not in file:
+            file['parents'] = ['%s:%s' % (parent['scope'], parent['name'])
+                               for parent in rucio.core.did.list_parent_dids(file['scope'], file['name'], session=session)]
+
         # also sort the pfns inside the rse structure
         rse_pfns = []
         for t_rse, t_priority, t_pfn in [(file['pfns'][t_pfn]['rse'], file['pfns'][t_pfn]['priority'], t_pfn) for t_pfn in file['pfns']]:
@@ -1036,7 +1063,7 @@ def list_replicas(dids, schemes=None, unavailable=False, request_id=None,
                   ignore_availability=True, all_states=False, pfns=True,
                   rse_expression=None, client_location=None, domain=None,
                   sign_urls=False, signature_lifetime=None, resolve_archives=True,
-                  session=None):
+                  resolve_parents=False, session=None):
     """
     List file replicas for a list of data identifiers (DIDs).
 
@@ -1052,6 +1079,7 @@ def list_replicas(dids, schemes=None, unavailable=False, request_id=None,
     :param sign_urls: If set, will sign the PFNs if necessary.
     :param signature_lifetime: If supported, in seconds, restrict the lifetime of the signed PFN.
     :param resolve_archives: When set to true, find archives which contain the replicas.
+    :param resolve_parents: When set to true, find all parent datasets which contain the replicas.
     :param session: The database session in use.
     """
 
@@ -1067,7 +1095,7 @@ def list_replicas(dids, schemes=None, unavailable=False, request_id=None,
             rse_clause.append(models.RSEFileAssociation.rse_id == rse['id'])
     for f in _list_replicas(dataset_clause, file_clause, state_clause, pfns,
                             schemes, files, rse_clause, rse_expression, client_location, domain,
-                            sign_urls, signature_lifetime, constituents, session):
+                            sign_urls, signature_lifetime, constituents, resolve_parents, session):
         yield f
 
 
diff --git a/lib/rucio/daemons/tracer/kronos.py b/lib/rucio/daemons/tracer/kronos.py
index c5357ac4b..6fe45c751 100644
--- a/lib/rucio/daemons/tracer/kronos.py
+++ b/lib/rucio/daemons/tracer/kronos.py
@@ -143,11 +143,15 @@ class AMQConsumer(object):
                     for pattern in self.__bad_files_patterns:
                         if 'stateReason' in report and pattern.match(report['stateReason']):
                             reason = report['stateReason'][:255]
-                            try:
-                                declare_bad_file_replicas(report['url'], reason=reason, issuer='root', status=BadFilesStatus.SUSPICIOUS)
-                                logging.info('Declare suspicious file %s with reason %s' % (report['url'], reason))
-                            except Exception as error:
-                                logging.error('Failed to declare suspicious file' + str(error))
+                            if 'url' not in report or not report['url']:
+                                logging.error('Missing url in the following trace : ' + str(report))
+                            else:
+                                try:
+                                    surl = report['url']
+                                    declare_bad_file_replicas([surl, ], reason=reason, issuer='root', status=BadFilesStatus.SUSPICIOUS)
+                                    logging.info('Declare suspicious file %s with reason %s' % (report['url'], reason))
+                                except Exception as error:
+                                    logging.error('Failed to declare suspicious file' + str(error))
 
                 # check if scope in report. if not skip this one.
                 if 'scope' not in report:
diff --git a/lib/rucio/web/rest/webpy/v1/replica.py b/lib/rucio/web/rest/webpy/v1/replica.py
index b778dd944..776f65915 100755
--- a/lib/rucio/web/rest/webpy/v1/replica.py
+++ b/lib/rucio/web/rest/webpy/v1/replica.py
@@ -297,7 +297,7 @@ class ListReplicas(RucioController):
 
         dids, schemes, select, unavailable, limit = [], None, None, False, None
         ignore_availability, rse_expression, all_states, domain = False, None, False, None
-        signature_lifetime, resolve_archives = None, True
+        signature_lifetime, resolve_archives, resolve_parents = None, True, False
         client_location = {}
 
         json_data = data()
@@ -323,6 +323,8 @@ class ListReplicas(RucioController):
                 domain = params['domain']
             if 'resolve_archives' in params:
                 resolve_archives = params['resolve_archives']
+            if 'resolve_parents' in params:
+                resolve_parents = params['resolve_parents']
             if 'signature_lifetime' in params:
                 signature_lifetime = params['signature_lifetime']
             else:
@@ -358,6 +360,7 @@ class ListReplicas(RucioController):
                                        client_location=client_location,
                                        domain=domain, signature_lifetime=signature_lifetime,
                                        resolve_archives=resolve_archives,
+                                       resolve_parents=resolve_parents,
                                        issuer=ctx.env.get('issuer')):
                 if not metalink:
                     yield dumps(rfile, cls=APIEncoder) + '\\n'
@@ -373,6 +376,13 @@ class ListReplicas(RucioController):
                                                 rfile['pfns'][replica]['client_extract'])
 
                     yield ' <file name="' + rfile['name'] + '">\\n'
+
+                    if 'parents' in rfile and rfile['parents']:
+                        yield '  <parents>\\n'
+                        for parent in rfile['parents']:
+                            yield '   <did>' + parent + '</did>\\n'
+                        yield '  </parents>\\n'
+
                     yield '  <identity>' + rfile['scope'] + ':' + rfile['name'] + '</identity>\\n'
                     if rfile['adler32'] is not None:
                         yield '  <hash type="adler32">' + rfile['adler32'] + '</hash>\\n'
@@ -402,7 +412,7 @@ class ListReplicas(RucioController):
 
                     idx = 0
                     for replica in replicas:
-                        yield '   <url location="' + str(dictreplica[replica][2]) \\
+                        yield '  <url location="' + str(dictreplica[replica][2]) \\
                             + '" domain="' + str(dictreplica[replica][0]) \\
                             + '" priority="' + str(dictreplica[replica][1]) \\
                             + '" client_extract="' + str(dictreplica[replica][3]).lower() \\