Merge pull request #453 from aiven/rdunklau-workaround_tar_concurrency

Workaround tar concurrency issue

#453

Author: rikonen

pghoard/restore.py

@@ -33,7 +33,7 @@
 
 from pghoard.common import BaseBackupFormat, StrEnum
 from pghoard.rohmu import compat, dates, get_transfer, rohmufile
-from pghoard.rohmu.errors import Error, InvalidConfigurationError
+from pghoard.rohmu.errors import (Error, InvalidConfigurationError, MaybeRecoverableError)
 
 from . import common, config, logutil, version
 from .postgres_command import PGHOARD_HOST, PGHOARD_PORT
@@ -640,6 +640,7 @@ def __init__(self, *, app_config, debug, site, pgdata, tablespaces, data_files:
         self.manager_class = multiprocessing.Manager if self._process_count() > 1 else ThreadingManager
         self.max_stale_seconds = 120
         self.pending_jobs = set()
+        self.jobs_to_retry = set()
         self.pgdata = pgdata
         # There's no point in spawning child processes if process count is 1
         self.pool_class = multiprocessing.Pool if self._process_count() > 1 else multiprocessing.pool.ThreadPool
@@ -648,6 +649,7 @@ def __init__(self, *, app_config, debug, site, pgdata, tablespaces, data_files:
         self.sleep_fn = time.sleep
         self.tablespaces = tablespaces
         self.total_download_size = 0
+        self.retry_per_file = {}
 
     def fetch_all(self):
         for retry in range(3):
@@ -656,7 +658,7 @@ def fetch_all(self):
                     self._setup_progress_tracking(manager)
                     with self.pool_class(processes=self._process_count()) as pool:
                         self._queue_jobs(pool)
-                        self._wait_for_jobs_to_complete()
+                        self._wait_for_jobs_to_complete(pool)
                         # Context manager does not seem to properly wait for the subprocesses to exit, let's join
                         # the pool manually (close need to be called before joining)
                         pool.close()
@@ -738,15 +740,33 @@ def job_completed(self, key):
             if key in self.pending_jobs:
                 self.pending_jobs.remove(key)
                 self.completed_jobs.add(key)
+                self.retry_per_file.pop(key, None)
 
     def job_failed(self, key, exception):
+        if isinstance(exception, MaybeRecoverableError):
+            self.log.warning("Got error which can be recoverable from chunk download %s", exception)
+            with self.lock:
+                if key in self.pending_jobs:
+                    retries = self.retry_per_file.get(key, 0) + 1
+                    self.retry_per_file[key] = retries
+                    self.pending_jobs.remove(key)
+                    if retries <= 2:
+                        self.jobs_to_retry.add(key)
+                        return
+                    self.errors += 1
+                    self.completed_jobs.add(key)
+                    self.retry_per_file.pop(key, None)
+                    self.log.error("Giving up on recoverable error: %s", exception)
+                    return
+
         self.log.error("Got error from chunk download: %s", exception)
         self.last_progress_ts = time.monotonic()
         with self.lock:
             if key in self.pending_jobs:
                 self.errors += 1
                 self.pending_jobs.remove(key)
                 self.completed_jobs.add(key)
+                self.retry_per_file.pop(key, None)
 
     def jobs_in_progress(self):
         with self.lock:
@@ -788,8 +808,18 @@ def _write_status_output_to_file(self, output_file):
             }
         )
 
-    def _wait_for_jobs_to_complete(self):
+    def _wait_for_jobs_to_complete(self, pool):
         while self.jobs_in_progress():
+            to_queue = []
+            with self.lock:
+                if self.jobs_to_retry:
+                    for item in self.data_files:
+                        if item.id in self.jobs_to_retry:
+                            self.pending_jobs.add(item.id)
+                            self.jobs_to_retry.remove(item.id)
+                            to_queue.append(item)
+            for item in to_queue:
+                self._queue_job(pool, item)
             self._print_download_progress()
             if self.status_output_file:
                 self._write_status_output_to_file(self.status_output_file)
@@ -910,12 +940,16 @@ def _fetch_delta_file(self, metadata, fetch_fn):
         )
 
     def _fetch_and_extract_one_backup(self, metadata, file_size, fetch_fn):
+        # Force tar to use the C locale to match errors in stderr
+        tar_env = os.environ.copy()
+        tar_env["LANG"] = "C"
         with subprocess.Popen(
             self._build_tar_args(metadata),
             bufsize=0,
             stdin=subprocess.PIPE,
             stdout=subprocess.DEVNULL,
-            stderr=subprocess.PIPE
+            stderr=subprocess.PIPE,
+            env=tar_env
         ) as tar:
             common.increase_pipe_capacity(tar.stdin, tar.stderr)
             sink = rohmufile.create_sink_pipeline(
@@ -939,7 +973,18 @@ def _fetch_and_extract_one_backup(self, metadata, file_size, fetch_fn):
             exit_code = tar.wait()
             file_name = "<mem_bytes>" if isinstance(self.file_info, FileDataInfo) else self.file_info.name
             if exit_code != 0:
-                raise Exception("tar exited with code {!r} for file {!r}, output: {!r}".format(exit_code, file_name, output))
+                ex_message = "tar exited with code {!r} for file {!r}, output: {!r}".format(exit_code, file_name, output)
+                # Running multiple tar commands in parallel in the same
+                # directory can lead to race conditions while creating the
+                # intermediate directories.
+                # In that case, try to recover from it.
+                # See issue #452 and https://savannah.gnu.org/bugs/index.php?61015
+                if exit_code == 2 and b"Cannot open: No such file or directory" in output:
+                    raise MaybeRecoverableError(ex_message)
+                else:
+                    raise Exception(
+                        "tar exited with code {!r} for file {!r}, output: {!r}".format(exit_code, file_name, output)
+                    )
             self.log.info("Processing of %r completed successfully", file_name)
 
 

pghoard/rohmu/errors.py

@@ -28,3 +28,7 @@ class LocalFileIsRemoteFileError(StorageError):
 
 class MissingLibraryError(Exception):
     """Missing dependency library"""
+
+
+class MaybeRecoverableError(Error):
+    """An error that may be recoverable"""

test/basebackup_one_chunk/chunks/00000002.pghoard

@@ -0,0 +1 @@
+{"compression-algorithm": "snappy", "encryption-key-id": "5ba999de817c49a682ffed124abf9a2e", "format": "pghoard-bb-v2", "original-file-size": "20480"}

test/basebackup_one_chunk/chunks/00000002.pghoard.metadata

@@ -0,0 +1,23 @@
+{
+    "backup_sites": {
+        "f73f56ee-6b9f-4ce0-b7aa-a170d58da833": {
+            "encryption_key_id": "5ba999de817c49a682ffed124abf9a2e",
+            "encryption_keys": {
+                "5ba999de817c49a682ffed124abf9a2e": {
+                    "private": "-----BEGIN PRIVATE KEY-----\nMIIG/AIBADANBgkqhkiG9w0BAQEFAASCBuYwggbiAgEAAoIBgQC/e0jNVBCB8pxK\nwPmJUlus6q+mKQ9QD0esP/TzBZ6TwKiMlMukwh0FCah88UTf/9VNDEvgzFhrcbEc\n0O0ZKywSHNFOEq6onm3QWRqnMvXZLlTyhIBZRuLB8Vt3WH3Atv6BwbLRgFuT6Rfx\nopOGptmNQsOMT4z9lB2n2JiTBJsg7+iAfw6ZltuBSIjC8/5flcmYkkTQFEHUJ3RY\nOTjlqTY7y8J464qvXgQIUE/kCx7np4pdvWc3Zf9l1hgv5Ol/escpr8Mo2Cg5Qhjg\ntLzB4emx8dfnoV1oc77F4XEFj5SzBZqgKy/pV5yy8UtJ6NDBEwG4cK5kXXFyu31q\nc+XYzgD2SdmG3Fqvy5Ikwj8Sx82zrwBWEM8etEy7CniyZi7+Sr9G4NfFvBG8OOQ2\nXGMfnl8doEFYRtVdjU0o3VSn37ASCW8XsHgd/Zobu6k9sMKdU1iNrEjkiSGQtlbD\nsq8GMuq8saSbPyRUNQO7mFQC2F6K49KGgxpn4d65G0wQMMnbq4sCAwEAAQKCAYAK\nYYOr5g/TC7UfdGDS6g0gTcTiDD3RSFLJato7xqU3O22n2XVE5GUwXbqts2LZhgQp\nXi5K7KkqggppFoaUI7wK61cJlYe0iopHjl0cjW24rYNbdoWC0Y3/l7cuvDRtGz6n\nCDpKk1vjo/JxXjADT85hkyoI1FM/eCU3cU2sQsaqPXdsZ/cBqqUR2D3Z2+KBihxY\n0i063q5G8zCii8+i286d5UkQxyxIn582WCxMn7G4O2QL+vW6kiQLgFTlW9Kw35YO\nfbM04zUmpvZpjCJuqLDw/2x7/sJn17vzQ+LsdBz/JXaLEidkI4Tr4caWqSjJZVBK\nb5LzeOrqCBcsppXUhN+4yvqCAtK0iUCLbFwPuqqmv3Ly2GpWQBPEgozdaXZOOGzc\ne7M5xCDH5bQRR/3e6Md9fHl/ATKD5eSun+7dyjNq8FJHeUAEKS2okd5oFmTD4mkZ\nxV1GTHTLzUy0qeyiUoGCxnCnS8q9rx2Os4j4/Y+aQI6xzUWYK+Zb57o9w7kIirEC\ngcEA8BHiDzTM4HgmKJ5kPkB38JYkoI0hFzofesvvKg8Nx4hEg6KBL3t7kJPqdJgm\ntZHwhg3EwzKIRCY+wHCpYy0ouX/MAfKjWBI9uAYhTB5nbJEqxFNo18uK40PoWRMt\nNSralCrRcIb5z/Kl5WTsXN6DHJFTDwsGHF41LWrnCXRNM/XfzJvE7X0VlcjbL/A9\n0cxfEuTb7k2xhJNBbizCvdw92mmDemFxq9PO63966nnOIjwaX3m4G1yg3nsUGeTY\nVmpZAoHBAMwwBaj9x29Cxjn1Dy4rF5GiNErO5JY9mZnroa4I43zbYuXww5mWTpHs\nORntM1XtKt2E5KeW9fBUdJPWh4epnh2cyFmodSxaEk35wfnEGnMPlwHTgZvUE7kz\n6VXqDEGK4mYmzof6edbpg+tejx8SsK6Pwrt1Moj9evGs+I38oK2IRJchIS7Ur+qj\njLS13Z1mxLmAquDsmzAVgEL13rgUsBsiV+ghEEpvPS8gb9yZmpTiKD07JKml18VK\n06FsF9VAgwKBwBT/O2phD9pCJ/Q9hj77nIHqX+G69j3103MGCzD+iBH/lR5+RBZH\nEpOenE7+T8Rps0PGSINaFBkBz1M9h1MpS/qNduZktmypi8RgpODnd9xDBh6NvQnC\nv68I7XV8++M+kEeNRyw0Yf0SF/hsT1AAFi+VdlJGgI5SnwwN4Y8uIOJ+ish2h07O\nNekX3DPhK0cCPP6GDcZV/US/LGXafF3muXI08E7v3uVMbTijubhwVtsfrp7TIosi\nGt/am/N31IQaYQKBwAmUmz9hoPDsfiaMBAlThkiUBsYXzQvrmgBp2O00h4/9LzfA\nwzy6m7cnEUrRIV5/wUohiST/5UxAejPRlgxcfgm/qHrkd5L8Ku2zsVFJzT/m1FwG\nk4c/PSmscN9SGv8cSCEo4vnoW70kucbaafa4Rsf6ANYQ2q0oz5L1XbgzyUo7IZTB\nvi/XVOW6hMiZ2+sdvk9B5UKmd2WbLKh3ptqWRekQBHXkz0He1E0YxYbhQiqILgEp\nfD/lgylDqIhjbP7ZhwKBwFYIBmEYEs/+ixHkHGbzE4tP+VhgzQ7we1+xyqN7Z69e\nf7StEuWWIZ5Os9JpdhZmN/9qb4kzL+Pb0vELuNHLwv08MnJ792wA9dPeCuYLKKiK\nUBuBwnslBy69tCRSXQ8ltH3kX67lxCu28hDRg+oN83bEczPJ2I/x7BKDrWNYZvrq\nJw7Ijp5l5gLqtdb7eOI8mSs5F7yaUKR6yVhl3ifc6bQUmmGmmElobpCpcGqAmXSj\noAeJCZug3MWS2fxhluthxQ==\n-----END PRIVATE KEY-----\n",
+                    "public": "-----BEGIN PUBLIC KEY-----\nMIIBojANBgkqhkiG9w0BAQEFAAOCAY8AMIIBigKCAYEAv3tIzVQQgfKcSsD5iVJb\nrOqvpikPUA9HrD/08wWek8CojJTLpMIdBQmofPFE3//VTQxL4MxYa3GxHNDtGSss\nEhzRThKuqJ5t0FkapzL12S5U8oSAWUbiwfFbd1h9wLb+gcGy0YBbk+kX8aKThqbZ\njULDjE+M/ZQdp9iYkwSbIO/ogH8OmZbbgUiIwvP+X5XJmJJE0BRB1Cd0WDk45ak2\nO8vCeOuKr14ECFBP5Ase56eKXb1nN2X/ZdYYL+Tpf3rHKa/DKNgoOUIY4LS8weHp\nsfHX56FdaHO+xeFxBY+UswWaoCsv6VecsvFLSejQwRMBuHCuZF1xcrt9anPl2M4A\n9knZhtxar8uSJMI/EsfNs68AVhDPHrRMuwp4smYu/kq/RuDXxbwRvDjkNlxjH55f\nHaBBWEbVXY1NKN1Up9+wEglvF7B4Hf2aG7upPbDCnVNYjaxI5IkhkLZWw7KvBjLq\nvLGkmz8kVDUDu5hUAtheiuPShoMaZ+HeuRtMEDDJ26uLAgMBAAE=\n-----END PUBLIC KEY-----\n"
+                }
+            },
+            "object_storage": {
+                "directory": "test/basebackup/chunks",
+                "storage_type": "local"
+            },
+            "prefix": "1052a492-1a01-459d-a126-9db8518724c0/f73f56ee-6b9f-4ce0-b7aa-a170d58da833"
+        }
+    },
+    "compression": {
+        "algorithm": "snappy"
+    },
+    "log_level": "INFO",
+    "restore_process_count": 2
+}

test/basebackup_one_chunk/config.json

@@ -0,0 +1,5 @@
+#!/bin/bash
+# This is a mock program to simulate tar race conditions failure
+cat - > /dev/null
+echo "Cannot open: No such file or directory" >&2
+exit 2

test/tar_failer

@@ -235,6 +235,42 @@ def test_real_processing(self):
         for tar in ["tar", "pghoard/gnutaremu.py"]:
             self.run_restore_test("basebackup", tar, self.real_processing)
 
+    # Simulate a tar failure similar to what we have when running into
+    # concurrency issues
+    # We use a basebackup with one chunk to make sure all retries
+    # deal with the same file
+    def test_tar_recovery(self):
+        pass_after_retry = 2
+
+        def simulate_tar_failure(fetcher, restore_dir):
+            # Mock "_build_tar_args" to get the information we need
+            retry = 0
+            original_tar_executable = fetcher.config["tar_executable"]
+            fetcher.config["tar_executable"] = "test/tar_failer"
+            original_build_tar_args = ChunkFetcher._build_tar_args  # pylint: disable=protected-access
+
+            def _build_tar_args(self, metadata):
+                nonlocal retry, pass_after_retry
+                retry += 1
+                if retry >= pass_after_retry:
+                    self.config["tar_executable"] = original_tar_executable
+                return original_build_tar_args(self, metadata)
+
+            with patch("pghoard.restore.ChunkFetcher._build_tar_args", new=_build_tar_args):
+                # Now launch the test progress_percent
+                fetcher.fetch_all()
+                self.check_sha256(
+                    os.path.join(restore_dir, "base", "1", "3608"),
+                    "cd461a152a9259c2d311ee348a4fa6722c119c1ff9a5b3147a86058d76f9bba8"
+                )
+
+        # Check that after 1 failure, we succeed
+        self.run_restore_test("basebackup_one_chunk", "tar", simulate_tar_failure)
+        pass_after_retry = 10
+        # Check that after 3 failure, we fail
+        with pytest.raises(RestoreError):
+            self.run_restore_test("basebackup_one_chunk", "tar", simulate_tar_failure)
+
     def real_processing(self, fetcher, restore_dir):
         assert fetcher.pool_class == multiprocessing.Pool  # pylint: disable=comparison-with-callable
         fetcher.fetch_all()