Repository: rucio/rucio
Issue: minos-expiration doesn't handle properly ReplicaNotFound

Motivation
----------
minos-expiration doesn't handle properly ReplicaNotFound






Changed files:
lib/rucio/daemons/badreplicas/minos_temporary_expiration.py, lib/rucio/rse/rsemanager.py, lib/rucio/daemons/hermes/hermes.py, lib/rucio/client/downloadclient.py

diff --git a/lib/rucio/client/downloadclient.py b/lib/rucio/client/downloadclient.py
index 7b916bfff..1eeab073a 100644
--- a/lib/rucio/client/downloadclient.py
+++ b/lib/rucio/client/downloadclient.py
@@ -175,7 +175,8 @@ class DownloadClient:
             rse                 - rse name (e.g. 'CERN-PROD_DATADISK'). RSE Expressions are not allowed
             base_dir            - Optional: Base directory where the downloaded files will be stored. (Default: '.')
             no_subdir           - Optional: If true, files are written directly into base_dir and existing files are overwritten. (Default: False)
-            ignore_checksum     - Optional: If true, the checksum validation is skipped (for pfn downloads the checksum must be given explicitly). (Default: True)
+            adler32             - Optional: The adler32 checmsum to compare the downloaded files adler32 checksum with
+            md5                 - Optional: The md5 checksum to compare the downloaded files md5 checksum with
             transfer_timeout    - Optional: Timeout time for the download protocols. (Default: None)
         :param num_threads: Suggestion of number of threads to use for the download. It will be lowered if it's too high.
         :param trace_custom_fields: Custom key value pairs to send with the traces
@@ -221,7 +222,7 @@ class DownloadClient:
             item['dest_file_paths'] = [dest_file_path]
             item['temp_file_path'] = '%s.part' % dest_file_path
             options = item.setdefault('merged_options', {})
-            options.setdefault('ignore_checksum', item.pop('ignore_checksum', True))
+            options['ignore_checksum'] = 'adler32' not in item and 'md5' not in item
             options.setdefault('transfer_timeout', item.pop('transfer_timeout', None))
 
             input_items.append(item)
@@ -1081,8 +1082,20 @@ class DownloadClient:
                                                      resolve_parents=True,
                                                      metalink=True)
             file_items = parse_replicas_from_string(metalink_str)
+
             logger.debug('num resolved files: %s' % len(file_items))
 
+            for input_did in item['dids']:
+                input_did_str = '%s:%s' % (input_did['scope'], input_did['name'])
+                did_exists = False
+                for file_item in file_items:
+                    if input_did_str == file_item['did'] or input_did_str in file_item['parent_dids']:
+                        did_exists = True
+                        break
+                if not did_exists:
+                    logger.error('No replicas found for DIDs: %s' % input_did_str)
+                    file_items.append({'did': input_did_str, 'adler32': None, 'md5': None, 'sources': []})
+
             nrandom = item.get('nrandom')
             if nrandom:
                 logger.info('Selecting %d random replicas from DID(s): %s' % (nrandom, item['dids']))
@@ -1091,6 +1104,7 @@ class DownloadClient:
                 merged_items_with_sources.append(file_items)
             else:
                 merged_items_with_sources.append(file_items)
+
         return merged_items_with_sources
 
     def _prepare_items_for_download(self, did_to_options, merged_items_with_sources, resolve_archives=True):
@@ -1130,7 +1144,6 @@ class DownloadClient:
         all_dest_file_paths = set()
 
         # get replicas for every file of the given dids
-        logger.debug('num list_replicas calls: %d' % len(merged_items_with_sources))
         for file_items in merged_items_with_sources:
             all_file_items.extend(file_items)
             for file_item in file_items:
diff --git a/lib/rucio/daemons/badreplicas/minos_temporary_expiration.py b/lib/rucio/daemons/badreplicas/minos_temporary_expiration.py
index 032bafee0..5b7c01739 100644
--- a/lib/rucio/daemons/badreplicas/minos_temporary_expiration.py
+++ b/lib/rucio/daemons/badreplicas/minos_temporary_expiration.py
@@ -19,9 +19,11 @@
 # PY3K COMPATIBLE
 
 import logging
+import math
 import os
 import socket
 import threading
+import traceback
 import time
 
 from sys import stdout, argv
@@ -31,6 +33,8 @@ from rucio.db.sqla.constants import BadFilesStatus, ReplicaState
 from rucio.db.sqla.session import get_session
 from rucio.common.config import config_get
 from rucio.common.utils import chunks
+from rucio.common.exception import DataIdentifierNotFound, ReplicaNotFound
+from rucio.core.did import get_metadata
 from rucio.core.replica import (update_replicas_states,
                                 bulk_delete_bad_replicas, list_expired_temporary_unavailable_replicas)
 
@@ -96,17 +100,48 @@ def minos_tu_expiration(bulk=1000, once=False, sleep_time=60):
                 bad_replicas.append({'scope': replica[0], 'name': replica[1], 'rse_id': replica[2], 'state': BadFilesStatus.TEMPORARY_UNAVAILABLE})
             session = get_session()
 
-            for chunk in chunks(replicas, chunk_size):
+            nchunk = 0
+            tot_chunk = int(math.ceil(len(replicas) / float(chunk_size)))
+            session = get_session()
+            for chunk in chunks(expired_replicas, chunk_size):
                 # Process and update the replicas in chunks
-                update_replicas_states(chunk, nowait=True, session=session)
-            for chunk in chunks(bad_replicas, chunk_size):
+                replicas = [{'scope': replica[0], 'name': replica[1], 'rse_id': replica[2], 'state': ReplicaState.AVAILABLE} for replica in chunk]
                 # Remove the replicas from bad_replicas table in chunks
-                bulk_delete_bad_replicas(chunk, session=session)
-
-            session.commit()  # pylint: disable=no-member
+                bad_replicas = [{'scope': replica[0], 'name': replica[1], 'rse_id': replica[2], 'state': BadFilesStatus.TEMPORARY_UNAVAILABLE} for replica in chunk]
+                try:
+                    nchunk += 1
+                    logging.debug(prepend_str + 'Running on %s chunk out of %s' % (nchunk, tot_chunk))
+                    update_replicas_states(replicas, nowait=True, session=session)
+                    bulk_delete_bad_replicas(bad_replicas, session=session)
+                    session.commit()  # pylint: disable=no-member
+                except (ReplicaNotFound, DataIdentifierNotFound) as error:
+                    session.rollback()  # pylint: disable=no-member
+                    logging.warning(prepend_str + 'One of the replicas does not exist anymore. Updating and deleting one by one. Error : %s' % str(error))
+                    for idx in range(len(chunk)):
+                        logging.debug(prepend_str + 'Working on %s' % (str(replicas[idx])))
+                        try:
+                            get_metadata(replicas[idx]['scope'], replicas[idx]['name'])
+                            update_replicas_states([replicas[idx], ], nowait=True, session=session)
+                            bulk_delete_bad_replicas([bad_replicas[idx], ], session=session)
+                            session.commit()  # pylint: disable=no-member
+                        except DataIdentifierNotFound as error:
+                            session.rollback()  # pylint: disable=no-member
+                            logging.warning(prepend_str + 'DID %s:%s does not exist anymore. ' % (bad_replicas[idx]['scope'], bad_replicas[idx]['name']))
+                            bulk_delete_bad_replicas([bad_replicas[idx], ], session=session)
+                            session.commit()  # pylint: disable=no-member
+                        except ReplicaNotFound as error:
+                            session.rollback()  # pylint: disable=no-member
+                            logging.warning(prepend_str + '%s:%s on RSEID %s does not exist anymore. ' % (replicas[idx]['scope'], replicas[idx]['name'], replicas[idx]['rse_id']))
+                            bulk_delete_bad_replicas([bad_replicas[idx], ], session=session)
+                            session.commit()  # pylint: disable=no-member
+                    session = get_session()
+                except Exception:
+                    session.rollback()  # pylint: disable=no-member
+                    logging.critical(traceback.format_exc())
+                    session = get_session()
 
         except Exception as error:
-            logging.error(prepend_str + '%s' % (str(error)))
+            logging.critical(traceback.format_exc())
 
         tottime = time.time() - start_time
         if once:
diff --git a/lib/rucio/daemons/hermes/hermes.py b/lib/rucio/daemons/hermes/hermes.py
index 9153a6373..3e3b3e0cb 100644
--- a/lib/rucio/daemons/hermes/hermes.py
+++ b/lib/rucio/daemons/hermes/hermes.py
@@ -173,6 +173,9 @@ def deliver_messages(once=False, brokers_resolved=None, thread=0, bulk=1000, del
         logging.fatal('No brokers resolved.')
         return
 
+    if not broker_timeout:  # Allow zero in config
+        broker_timeout = None
+
     logging.info('[broker] checking authentication method')
     use_ssl = True
     try:
diff --git a/lib/rucio/rse/rsemanager.py b/lib/rucio/rse/rsemanager.py
index 650cbf2cb..15da47048 100644
--- a/lib/rucio/rse/rsemanager.py
+++ b/lib/rucio/rse/rsemanager.py
@@ -32,7 +32,6 @@
 from __future__ import print_function
 
 import copy
-import os
 import random
 from time import sleep
 
@@ -218,92 +217,6 @@ def parse_pfns(rse_settings, pfns, operation='read', domain='wan'):
     return create_protocol(rse_settings, operation, urlparse(pfns[0]).scheme, domain).parse_pfns(pfns)
 
 
-def download(rse_settings, files, dest_dir=None, force_scheme=None, ignore_checksum=False, printstatements=False, domain='wan', transfer_timeout=None):
-    """
-        Copy a file from the connected storage to the local file system.
-        Providing a list indicates the bulk mode.
-
-
-        :param rse_settings:    RSE to use
-        :param files:           a single dict or a list with dicts containing 'scope' and 'name'
-                                if LFNs are provided and additional 'pfn' if PFNs are provided.
-                                Examples:
-                                [
-                                {'name': '2_rse_remote_get.raw', 'scope': 'user.jdoe'},
-                                {'name':'3_rse_remote_get.raw', 'scope': 'user.jdoe', 'pfn': 'user/jdoe/5a/98/3_rse_remote_get.raw'}
-                                ]
-        :param dest_dir:        path to the directory where the downloaded files will be stored. If not given, each scope is represented by its own directory.
-        :param force_scheme:    normally the scheme is dictated by the RSE object, when specifying the PFN it must be forced to the one specified in the PFN, overruling the RSE description.
-        :param ignore_checksum: do not verify the checksum - caution: should only be used for rucio download --pfn
-        :param transfer_timeout: set this timeout (in seconds) for the transfers, for protocols that support it
-
-        :returns: True/False for a single file or a dict object with 'scope:name' for LFNs or 'name' for PFNs as keys and True or the exception as value for each file in bulk mode
-
-        :raises SourceNotFound: remote source file can not be found on storage
-        :raises DestinationNotAccessible: local destination directory is not accessible
-        :raises FileConsistencyMismatch: the checksum of the downloaded file does not match the provided one
-        :raises ServiceUnavailable: for any other reason
-
-    """
-    ret = {}
-    gs = True  # gs represents the global status which inidcates if every operation workd in bulk mode
-
-    protocol = create_protocol(rse_settings, 'read', scheme=force_scheme, domain=domain)
-    protocol.connect()
-
-    files = [files] if not type(files) is list else files
-    for f in files:
-        pfn = f['pfn'] if 'pfn' in f else list(protocol.lfns2pfns(f).values())[0]
-        target_dir = "./%s" % f['scope'] if dest_dir is None else dest_dir
-        try:
-            if not os.path.exists(target_dir):
-                os.makedirs(target_dir)
-            # Each scope is stored into a separate folder
-            finalfile = '%s/%s' % (target_dir, f['name'])
-            # Check if the file already exists, if not download and validate it
-            if not os.path.isfile(finalfile):
-                if 'adler32' in f:
-                    tempfile = '%s/%s.part' % (target_dir, f['name'])
-                    if os.path.isfile(tempfile):
-                        if printstatements:
-                            print('%s already exists, probably from a failed attempt. Will remove it' % (tempfile))
-                        os.unlink(tempfile)
-                    protocol.get(pfn, tempfile, transfer_timeout=transfer_timeout)
-                    if printstatements:
-                        print('File downloaded. Will be validated')
-
-                    if ignore_checksum:
-                        if printstatements:
-                            print('Skipping checksum validation')
-                    else:
-                        ruciochecksum = f['adler32'] if f['adler32'] else f['md5']
-                        localchecksum = utils.adler32(tempfile) if f['adler32'] else utils.md5(tempfile)
-                        if localchecksum == ruciochecksum:
-                            if printstatements:
-                                print('File validated')
-                        else:
-                            os.unlink(tempfile)
-                            raise exception.FileConsistencyMismatch('Checksum mismatch : local %s vs recorded %s' % (str(localchecksum), str(ruciochecksum)))
-                    os.rename(tempfile, finalfile)
-                else:
-                    protocol.get(pfn, '%s/%s' % (target_dir, f['name']), transfer_timeout=transfer_timeout)
-                ret['%s:%s' % (f['scope'], f['name'])] = True
-            else:
-                ret['%s:%s' % (f['scope'], f['name'])] = True
-        except Exception as e:
-            gs = False
-            ret['%s:%s' % (f['scope'], f['name'])] = e
-
-    protocol.close()
-    if len(ret) == 1:
-        for x in ret:
-            if isinstance(ret[x], Exception):
-                raise ret[x]
-            else:
-                return ret[x]
-    return [gs, ret]
-
-
 def exists(rse_settings, files):
     """
         Checks if a file is present at the connected storage.