Release/42.0.4

docs/CHANGELOG.rst

@@ -6,6 +6,17 @@ All notable changes to this project are documented in this file.
 This project adheres to `Semantic Versioning <http://semver.org/>`_.
 
 
+===================
+42.0.4 - 2024-12-09
+===================
+
+Changed
+-------
+- Catch ``EndpointConnectionError`` in retry decorator
+- Implement exponential backoff in retry decorator
+- Reduce the maximal number of threads for downolad data in the init container
+
+
 ===================
 42.0.3 - 2024-12-09
 ===================

resolwe/flow/executors/init_container.py

@@ -56,6 +56,9 @@
 DOWNLOAD_WAITING_TIMEOUT = 60  # in seconds
 RETRIES = 5
 
+# Max threads to use for data download.
+MAX_DOWNLOAD_THREADS = int(os.environ.get("GENESIS_MAX_DOWNLOAD_THREADS", 3))
+
 # Configure container logger. All logs are output to stdout for further
 # processing.
 # The log level defaults to debug except for boto and google loggers, which
@@ -112,8 +115,8 @@ async def transfer_inputs(communicator: BaseCommunicator, missing_data: dict):
 
     try:
         for connector_name in objects_to_transfer:
-            min_threads = 5
-            max_threads = 20
+            min_threads = 1
+            max_threads = MAX_DOWNLOAD_THREADS
             # Use from min_threads to max_threads threads. Assume each thread
             # can handle at least 100 files in reasonable time.
             files_count = len(objects_to_transfer[connector_name])

resolwe/storage/connectors/transfer.py

@@ -30,16 +30,17 @@
     gcs_exceptions = []
 
 try:
-    from botocore.exceptions import ClientError
+    from botocore.exceptions import ClientError, EndpointConnectionError
 
-    boto_exceptions = [ClientError]
+    boto_exceptions = [ClientError, EndpointConnectionError]
 except ModuleNotFoundError:
     boto_exceptions = []
 
 
 logger = logging.getLogger(__name__)
 ERROR_MAX_RETRIES = 3
-ERROR_TIMEOUT = 5  # In seconds.
+ERROR_INITIAL_TIMEOUT = 1  # In seconds.
+ERROR_MAX_TIMEOUT = 60  # In seconds.
 transfer_exceptions = tuple(
     boto_exceptions
     + gcs_exceptions
@@ -56,14 +57,17 @@ def retry_on_transfer_error(wrapped, instance, args, kwargs):
             return wrapped(*args, **kwargs)
         except transfer_exceptions:
             # Log the exception on retry for inspection.
-            if retry != ERROR_MAX_RETRIES:
+            if retry < ERROR_MAX_RETRIES:
+                timeout = min(
+                    ERROR_MAX_TIMEOUT, ERROR_INITIAL_TIMEOUT * (2 ** (retry - 1))
+                )
                 logger.exception(
                     "Retry %d/%d got exception, will retry in %d seconds.",
                     retry,
                     ERROR_MAX_RETRIES,
-                    ERROR_TIMEOUT,
+                    timeout,
                 )
-                sleep(ERROR_TIMEOUT)
+                sleep(timeout)
             # Raise exception when max retries are exceeded.
             else:
                 logger.exception("Final retry got exception, re-raising it.")

resolwe/storage/tests/test_transfer.py

@@ -41,7 +41,8 @@ def test_exception(self):
             with self.assertRaises(DataTransferError):
                 t.transfer_objects("test_url", [{}, {}])
 
-    @patch("resolwe.storage.connectors.transfer.ERROR_TIMEOUT", 0.1)
+    @patch("resolwe.storage.connectors.transfer.ERROR_INITIAL_TIMEOUT", 0.1)
+    @patch("resolwe.storage.connectors.transfer.ERROR_MAX_TIMEOUT", 0.1)
     @patch("resolwe.storage.connectors.transfer.ERROR_MAX_RETRIES", 3)
     def test_retry_transfer(self):
         t = Transfer(self.local, self.local)