Retry storage operations in case of ClientError.

CHANGELOG.D/1107.feature

@@ -0,0 +1 @@
+Retry storage operations in case of some errors.

neuromation/api/storage.py

@@ -46,7 +46,7 @@
     normalize_storage_path_uri,
 )
 from .users import Action
-from .utils import NoPublicConstructor
+from .utils import NoPublicConstructor, retries
 
 
 MAX_OPEN_FILES = 100
@@ -384,9 +384,12 @@ async def _upload_file(
         progress: AbstractFileProgress,
         queue: "asyncio.Queue[ProgressQueueItem]",
     ) -> None:
-        await self.create(
-            dst, self._iterate_file(src_path, dst, progress=progress, queue=queue)
-        )
+        for retry in retries(f"Fail to upload {dst}"):
+            async with retry:
+                await self.create(
+                    dst,
+                    self._iterate_file(src_path, dst, progress=progress, queue=queue),
+                )
 
     async def upload_dir(
         self,
@@ -428,14 +431,20 @@ async def _upload_dir(
             exists = False
             if update:
                 try:
-                    dst_files = {
-                        item.name: item for item in await self.ls(dst) if item.is_file()
-                    }
+                    for retry in retries(f"Fail to list {dst}"):
+                        async with retry:
+                            dst_files = {
+                                item.name: item
+                                for item in await self.ls(dst)
+                                if item.is_file()
+                            }
                     exists = True
                 except ResourceNotFound:
                     update = False
             if not exists:
-                await self.mkdir(dst, exist_ok=True)
+                for retry in retries(f"Fail to create {dst}"):
+                    async with retry:
+                        await self.mkdir(dst, exist_ok=True)
         except (FileExistsError, neuromation.api.IllegalArgumentError):
             raise NotADirectoryError(errno.ENOTDIR, "Not a directory", str(dst))
         await queue.put((progress.enter, StorageProgressEnterDir(src, dst)))
@@ -532,13 +541,18 @@ async def _download_file(
         async with self._file_sem:
             with dst_path.open("wb") as stream:
                 await queue.put((progress.start, StorageProgressStart(src, dst, size)))
-                pos = 0
-                async for chunk in self.open(src):
-                    pos += len(chunk)
-                    await queue.put(
-                        (progress.step, StorageProgressStep(src, dst, pos, size))
-                    )
-                    await loop.run_in_executor(None, stream.write, chunk)
+                for retry in retries(f"Fail to download {src}"):
+                    async with retry:
+                        pos = 0
+                        async for chunk in self.open(src):
+                            pos += len(chunk)
+                            await queue.put(
+                                (
+                                    progress.step,
+                                    StorageProgressStep(src, dst, pos, size),
+                                )
+                            )
+                            await loop.run_in_executor(None, stream.write, chunk)
                 await queue.put(
                     (progress.complete, StorageProgressComplete(src, dst, size))
                 )
@@ -586,7 +600,11 @@ async def _download_dir(
                         item.name: item for item in dst_path.iterdir() if item.is_file()
                     },
                 )
-        folder = await self.ls(src)
+
+        for retry in retries(f"Fail to list {src}"):
+            async with retry:
+                folder = await self.ls(src)
+
         for child in folder:
             name = child.name
             if child.is_file():

neuromation/api/utils.py

@@ -1,16 +1,23 @@
+import asyncio
+import logging
 import sys
 from types import TracebackType
 from typing import (
     Any,
+    AsyncIterator,
     Awaitable,
+    Callable,
     Coroutine,
     Generator,
     Generic,
+    Iterator,
     Optional,
     Type,
     TypeVar,
 )
 
+import aiohttp
+
 
 if sys.version_info >= (3, 7):  # pragma: no cover
     from contextlib import asynccontextmanager  # noqa
@@ -73,3 +80,31 @@ async def __aexit__(
         # Need to teach mypy about this facility
         await self._ret.close()  # type: ignore
         return None
+
+
+log = logging.getLogger(__name__)
+
+
+def retries(
+    msg: str, attempts: int = 10, logger: Callable[[str], None] = log.info
+) -> Iterator[AsyncContextManager[None]]:
+    sleeptime = 0.0
+    while attempts:
+
+        @asynccontextmanager
+        async def retry() -> AsyncIterator[None]:
+            nonlocal attempts
+            if attempts:
+                try:
+                    yield
+                except aiohttp.ClientError as err:
+                    logger(f"{msg}: {err}.  Retry...")
+                    await asyncio.sleep(sleeptime)
+                else:
+                    attempts = 0
+            else:
+                yield
+
+        sleeptime += 0.1
+        attempts -= 1
+        yield retry()