Repository: rucio/rucio
Issue: Minos incorrectly handles pfns with different scheme for same RSE

Motivation
----------

Minos was recently failing with the following error:
```
2021-07-08 07:48:44,800	root	850	DEBUG	minos[1/23] Running on RSE IFIC-LCG2_LOCALGROUPDISK with 16 replicas
2021-07-08 07:48:44,800	root	850	DEBUG	minos[1/23] Running on 1 chunk out of 2
2021-07-08 07:48:44,803	root	850	CRITICAL	minos[1/23] Exception
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/rucio/daemons/badreplicas/minos.py", line 152, in minos
    unknown_replicas = declare_bad_file_replicas(pfns=chunk, reason=reason, issuer=account, status=state, session=session)
  File "/usr/local/lib/python3.6/site-packages/rucio/db/sqla/session.py", line 388, in new_funct
    result = function(*args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/rucio/core/replica.py", line 476, in declare_bad_file_replicas
    scheme, files_to_declare, unknown_replicas = get_pfn_to_rse(pfns, vo=issuer.vo, session=session)
  File "/usr/local/lib/python3.6/site-packages/retrying.py", line 49, in wrapped_f
    return Retrying(*dargs, **dkw).call(f, *args, **kw)
  File "/usr/local/lib/python3.6/site-packages/retrying.py", line 206, in call
    return attempt.get(self._wrap_exception)
  File "/usr/local/lib/python3.6/site-packages/retrying.py", line 247, in get
    six.reraise(self.value[0], self.value[1], self.value[2])
  File "/usr/local/lib/python3.6/site-packages/six.py", line 719, in reraise
    raise value
  File "/usr/local/lib/python3.6/site-packages/retrying.py", line 200, in call
    attempt = Attempt(fn(*args, **kwargs), attempt_number, False)
  File "/usr/local/lib/python3.6/site-packages/rucio/db/sqla/session.py", line 308, in new_funct
    return function(*args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/rucio/core/replica.py", line 503, in get_pfn_to_rse
    raise exception.InvalidType('The PFNs specified must have the same protocol')
rucio.common.exception.InvalidType: Provided type is considered invalid.
Details: The PFNs specified must have the same protocol
```

Checking the content of `chunk` at https://github.com/rucio/rucio/blob/4cfbd05b469962b74333237a75e7e8eed3561d95/lib/rucio/daemons/badreplicas/minos.py#L151 , it does contain pfns with two different schemes. 

The code at https://github.com/rucio/rucio/blob/4cfbd05b469962b74333237a75e7e8eed3561d95/lib/rucio/daemons/badreplicas/minos.py#L136 can actually group multiple scheme pfns together. Which is probably the reason of the bug. 

Modification
------------

Make minos handle correctly pfns with different schemes for the same RSE; and add a test to ensure non-regression. 



Changed files:
lib/rucio/daemons/badreplicas/minos.py

diff --git a/lib/rucio/daemons/badreplicas/minos.py b/lib/rucio/daemons/badreplicas/minos.py
index b8cf7908d..986f0795e 100644
--- a/lib/rucio/daemons/badreplicas/minos.py
+++ b/lib/rucio/daemons/badreplicas/minos.py
@@ -133,27 +133,30 @@ def minos(bulk=1000, once=False, sleep_time=60):
                         _, tmp_dict_rse, tmp_unknown_replicas = get_pfn_to_rse(schemes[scheme], vo=vo)
                         for rse_id in tmp_dict_rse:
                             if rse_id not in dict_rse:
-                                dict_rse[rse_id] = []
-                            dict_rse[rse_id].extend(tmp_dict_rse[rse_id])
+                                dict_rse[rse_id] = {}
+                            if scheme not in dict_rse[rse_id]:
+                                dict_rse[rse_id][scheme] = []
+                            dict_rse[rse_id][scheme].extend(tmp_dict_rse[rse_id])
                         unknown_replicas.extend(tmp_unknown_replicas.get('unknown', []))
                     # The replicas in unknown_replicas do not exist, so we flush them from bad_pfns
                     if unknown_replicas:
                         logger(logging.INFO, 'The following replicas are unknown and will be removed : %s' % str(unknown_replicas))
                         bulk_delete_bad_pfns(pfns=unknown_replicas, session=None)
 
-                    for rse_id in dict_rse:
-                        vo_str = '' if vo == 'def' else ' on VO ' + vo
-                        logger(logging.DEBUG, 'Running on RSE %s%s with %s replicas' % (get_rse_name(rse_id=rse_id), vo_str, len(dict_rse[rse_id])))
-                        nchunk = 0
-                        tot_chunk = int(math.ceil(len(dict_rse[rse_id]) / chunk_size))
-                        for chunk in chunks(dict_rse[rse_id], chunk_size):
-                            nchunk += 1
-                            logger(logging.DEBUG, 'Running on %s chunk out of %s' % (nchunk, tot_chunk))
-                            unknown_replicas = declare_bad_file_replicas(pfns=chunk, reason=reason, issuer=account, status=state, session=session)
-                            if unknown_replicas:
-                                logger(logging.DEBUG, 'Unknown replicas : %s' % (str(unknown_replicas)))
-                            bulk_delete_bad_pfns(pfns=chunk, session=session)
-                            session.commit()  # pylint: disable=no-member
+                    for rse_id, pfns_by_scheme in dict_rse.items():
+                        for scheme, pfns in pfns_by_scheme.items():
+                            vo_str = '' if vo == 'def' else ' on VO ' + vo
+                            logger(logging.DEBUG, 'Running on RSE %s%s with %s replicas' % (get_rse_name(rse_id=rse_id), vo_str, len(pfns)))
+                            nchunk = 0
+                            tot_chunk = int(math.ceil(len(pfns) / chunk_size))
+                            for chunk in chunks(pfns, chunk_size):
+                                nchunk += 1
+                                logger(logging.DEBUG, 'Running on %s chunk out of %s' % (nchunk, tot_chunk))
+                                unknown_replicas = declare_bad_file_replicas(pfns=chunk, reason=reason, issuer=account, status=state, session=session)
+                                if unknown_replicas:
+                                    logger(logging.DEBUG, 'Unknown replicas : %s' % (str(unknown_replicas)))
+                                bulk_delete_bad_pfns(pfns=chunk, session=session)
+                                session.commit()  # pylint: disable=no-member
                 except Exception:
                     session.rollback()  # pylint: disable=no-member
                     logger(logging.CRITICAL, 'Exception', exc_info=True)