ARCHVIE: Add retry functionality for archive operations that timeout while

waiting for the files to be marked as on-tape. The default max timeout is
set to 24 hours and can be changed by using the ARCHIVE_MAX_TIMEOUT
parameter in the eosarchived configuration.
Fixes EOS-6206

Author: esindril

archive/eosarch/configuration.py

@@ -41,7 +41,7 @@ def __init__(self):
 
         Args:
             fn_conf (string): Path to the configuration file, which in normal
-            conditions should be/etc/eosarchived.conf
+            conditions should be /etc/eosarchived.conf
         """
         try:
             LOG_DIR = os.environ["LOG_DIR"]
@@ -89,6 +89,7 @@ def __init__(self):
         self.__dict__['OPT_FORCE'] = 'force'
         self.__dict__['ARCH_FN'] = ".archive"
         self.__dict__['ARCH_INIT'] = ".archive.init"
+        self.__dict__['ARCHIVE_MAX_TIMEOUT'] = '86400'
 
         try:
             with open(archive_conf, 'r') as f:

archive/eosarch/exceptions.py

@@ -33,3 +33,8 @@ class CheckEntryException(Exception):
     """ Exception raised in cache a verify entry operation failes.
     """
     pass
+
+class NotOnTapeException(Exception):
+    """ Exception raised when a file is not on tape after the maximum
+        configured timeout per entry
+    """

archive/eosarch/transfer.py

@@ -250,7 +250,11 @@ def run(self):
             if self.do_retry:
                 self.do_retry_transfer()
             else:
-                self.do_transfer()
+                try:
+                    self.do_transfer()
+                except NotOnTapeException as _:
+                    self.logger.notice("Doing transfer re-try due to missing file on tape")
+                    self.do_retry_transfer()
         elif self.oper in [self.config.PURGE_OP, self.config.DELETE_OP]:
             self.archive_prepare()
             self.do_delete((self.oper == self.config.DELETE_OP))
@@ -437,7 +441,7 @@ def do_retry_transfer(self):
         self.set_status("verifying")
         check_ok, __ = self.archive.verify(False)
 
-        # For PUT operations what that all the files are on tape
+        # For PUT operations wait that all the files are on tape
         if self.archive.d2t:
            self.set_status("wait_on_tape")
            self.wait_on_tape()
@@ -1068,20 +1072,20 @@ def evict_disk_cache(self):
 
     def wait_on_tape(self):
         """ Check and wait that all the files are on tape, which in our case
-        means checking the "m" bit. If file is not on tape then suspend the
-        current thread for a period between 1 and 10 minutes depending on the
-        index of the failed file.
+        means checking the "m" bit. If a file is not on tape then suspend the
+        current thread for a period of 5 to 60 seconds but abort if the file
+        fails to be archived on tape afte 24h
         """
-        min_timeout, max_timeout = 5, 1
+        max_timeout_per_entry = int(self.config.ARCHIVE_MAX_TIMEOUT)
+        min_timeout, max_timeout = 5, 60
 
-        while True:
-            indx = 0 # index of the first file not on tape
-            all_on_tape = True
+        for fentry in self.archive.files():
+            start_ts = time.time()
+            __, dst = self.archive.get_endpoints(fentry[1])
+            url = client.URL(dst)
+            file_on_tape = False
 
-            for fentry in self.archive.files():
-                indx += 1
-                __, dst = self.archive.get_endpoints(fentry[1])
-                url = client.URL(dst)
+            while not file_on_tape:
                 st_stat, resp_stat = self.archive.fs_dst.stat(url.path)
 
                 if not st_stat.ok:
@@ -1092,21 +1096,24 @@ def wait_on_tape(self):
                 # Check file is on tape
                 if resp_stat.size != 0 and not (resp_stat.flags & StatInfoFlags.BACKUP_EXISTS):
                     self.logger.debug("File {0} is not yet on tape".format(dst))
-                    all_on_tape = False
-                    break
-
-            if all_on_tape:
-                break
-            else:
-                # Set timeout value
-                ratio = indx / int(self.archive.header['num_files'])
-                timeout = int(max_timeout * (1 - ratio))
+                    timeout = randrange(min_timeout, max_timeou)
+                    self.logger.info("Going to sleep for {0} seconds".format(timeout))
+                    sleep(timeout)
+
+                    if time.time() - start_ts > max_timeout_per_entry:
+                        self.logger.notice("Entry not archived within the maximum timeout."
+                                           " entry={0} archive_max_timeout={1}".format(
+                                               fentry[1], max_timeout_per_entry))
+                        break
+                    else:
+                        file_on_tape = True
+                else:
+                    file_on_tape = True
 
-                if timeout < min_timeout:
-                    timeout = min_timeout
+            if not file_on_tape:
+                # Throw exception to re-try the failed transfer
+                raise NotOnTapeException()
 
-                self.logger.info("Going to sleep for {0} seconds".format(timeout))
-                sleep(timeout)
 
     def backup_prepare(self):
         """ Prepare requested backup operation.

archive/eosarchived.conf

@@ -47,3 +47,8 @@ POLL_TIMEOUT=30000
 
 # Join timeout in seconds for running threads inside a process
 JOIN_TIMEOUT=1
+
+# Maximum timeout value in seconds for a file entry to be migrated to tape.
+# When this timeout expires the transfer process is retried. By default this
+# is 86400 seconds (1 day).
+#ARCHIVE_MAX_TIMEOUT=86400