Add timeout and retry to combine_restore kubectl cp (#4323)

Co-Authored-By: Claude Opus 4.8 (1M context) & Fable 5 <noreply@anthropic.com>

Author: imnasnainaec

maintenance/scripts/combine_app.py

@@ -4,13 +4,26 @@
 
 from enum import Enum, unique
 import json
+import logging
+import os
 from pathlib import Path
 import subprocess
 import sys
+import time
 from typing import Any, Dict, List, Optional
 
 from maint_utils import run_cmd
 
+# A `kubectl cp` streams a tar over the exec channel and can stall intermittently,
+# which would otherwise hang forever.  Bound each copy with a timeout so a stalled
+# stream gets killed, and retry a few times so a transient stall is recovered.
+# Clamp to a sane floor so a misconfigured env var can't skip the copy entirely.
+CP_ATTEMPTS = max(1, int(os.getenv("kubectl_cp_attempts", "3")))
+CP_TIMEOUT_S = max(10.0, float(os.getenv("kubectl_cp_timeout_s", "300")))
+# A fast-failing copy can exhaust all attempts within the same second, giving a
+# transient failure no time to clear; pause between attempts so the retries span time.
+CP_RETRY_DELAY = 3.0
+
 
 class CombineApp:
     """Run commands on the Combine services."""
@@ -39,6 +52,7 @@ def exec(
         *,
         exec_opts: Optional[List[str]] = None,
         check_results: bool = True,
+        timeout: Optional[float] = None,
     ) -> subprocess.CompletedProcess[str]:
         """
         Run a kubectl 'exec' command in a Combine Kubernetes cluster.
@@ -52,6 +66,8 @@ def exec(
                      command, for example, to specify a working directory or a
                      specific user to run the command.
             check_results: Indicate if subprocess should not check for failure.
+            timeout: If set, kill the command and raise subprocess.TimeoutExpired
+                     when it runs longer than this many seconds.
         Returns a subprocess.CompletedProcess.
         """
         exec_opts = exec_opts or []
@@ -65,11 +81,69 @@ def exec(
             + [pod_id, "--"]
             + cmd,
             check_results=check_results,
+            timeout=timeout,
         )
 
-    def kubectl(self, cmd: List[str]) -> subprocess.CompletedProcess[str]:
-        """Run kubectl command adding the configuration file and namespace."""
-        return run_cmd(["kubectl"] + self.kubectl_opts + cmd)
+    def kubectl(
+        self, cmd: List[str], *, check_results: bool = True, timeout: Optional[float] = None
+    ) -> subprocess.CompletedProcess[str]:
+        """Run kubectl command adding the configuration file and namespace.
+
+        Args:
+            cmd: The kubectl subcommand and its arguments.
+            check_results: Indicate if subprocess should not check for failure.
+            timeout: If set, kill the command and raise subprocess.TimeoutExpired
+                     when it runs longer than this many seconds.
+        """
+        return run_cmd(
+            ["kubectl"] + self.kubectl_opts + cmd,
+            check_results=check_results,
+            timeout=timeout,
+        )
+
+    def cp_with_retry(
+        self,
+        cp_args: List[str],
+        *,
+        label: str,
+        timeout: float = CP_TIMEOUT_S,
+        attempts: int = CP_ATTEMPTS,
+    ) -> None:
+        """Run a `kubectl cp`, bounding it with a timeout and retrying transient stalls.
+
+        `kubectl cp` streams a tar over the exec channel and can stall intermittently
+        with no output.  Each attempt is killed after `timeout` seconds and the copy is
+        tried up to `attempts` times, pausing briefly between attempts.  If every
+        attempt fails, the failing copy is logged and the process exits non-zero so
+        the failure surfaces instead of hanging silently.
+
+        Args:
+            cp_args: Arguments to `kubectl cp` (source and destination, plus any flags).
+            label: Human-readable description of what is being copied, for logging.
+            timeout: Per-attempt timeout in seconds.
+            attempts: Total number of attempts before giving up.
+        """
+        for attempt in range(1, attempts + 1):
+            try:
+                proc = self.kubectl(["cp"] + cp_args, check_results=False, timeout=timeout)
+            except subprocess.TimeoutExpired:
+                logging.warning(
+                    f"Copy of {label} timed out after {timeout:g}s "
+                    f"(attempt {attempt}/{attempts})."
+                )
+            else:
+                if proc.returncode == 0:
+                    logging.debug(f"stderr:\n{proc.stderr.strip()}")
+                    logging.debug(f"stdout:\n{proc.stdout.strip()}")
+                    return
+                logging.warning(
+                    f"Copy of {label} failed with return code {proc.returncode} "
+                    f"(attempt {attempt}/{attempts}).\n{proc.stderr.strip()}"
+                )
+            if attempt < attempts:
+                time.sleep(CP_RETRY_DELAY)
+        logging.error(f"Failed to copy {label} after {attempts} attempts; aborting.")
+        sys.exit(1)
 
     def get_pod_id(self, service: CombineApp.Component, *, instance: int = 0) -> str:
         """Look up the Kubernetes pod id for the specified service."""

maintenance/scripts/combine_backup.py

@@ -96,23 +96,23 @@ def main() -> None:
 
         with tarfile.open(backup_file, "x:gz") as tar:
             # cp, tar, and rm the db subdir
-            combine.kubectl(
+            combine.cp_with_retry(
                 [
-                    "cp",
                     f"{db_pod}:/{db_files_subdir}",
                     str(Path(backup_dir) / db_files_subdir),
-                ]
+                ],
+                label=f"database dump ({db_files_subdir})",
             )
             tar.add(db_files_subdir)
             rmtree(db_files_subdir)
 
             # cp, tar, and rm the backend subdir
-            combine.kubectl(
+            combine.cp_with_retry(
                 [
-                    "cp",
                     f"{backend_pod}:/home/app/{backend_files_subdir}/",
                     str(Path(backup_dir) / backend_files_subdir),
-                ]
+                ],
+                label=f"backend files ({backend_files_subdir})",
             )
             tar.add(backend_files_subdir)
             rmtree(backend_files_subdir)

maintenance/scripts/combine_restore.py

@@ -160,9 +160,9 @@ def safe_extract(
             sys.exit(1)
 
         logging.debug(f"Copying {db_files_subdir} to {db_pod} ...")
-        cp_proc = combine.kubectl(["cp", db_files_subdir, f"{db_pod}:/"])
-        logging.debug(f"stderr:\n{cp_proc.stderr.strip()}")
-        logging.debug(f"stdout:\n{cp_proc.stdout.strip()}")
+        combine.cp_with_retry(
+            [db_files_subdir, f"{db_pod}:/"], label=f"database dump ({db_files_subdir})"
+        )
 
         logging.debug(f"Running mongorestore on {db_pod} ...")
         mongorestore_proc = combine.exec(
@@ -203,7 +203,10 @@ def safe_extract(
                 continue
             logging.debug(f"Copying {item} ...")
             local_item = os.path.join(backend_files_subdir, item)
-            combine.kubectl(["cp", local_item, remote_subdir, "--no-preserve"])
+            combine.cp_with_retry(
+                [local_item, remote_subdir, "--no-preserve"],
+                label=f"backend files for {item}",
+            )
 
 
 if __name__ == "__main__":

maintenance/scripts/maint_utils.py

@@ -6,7 +6,7 @@
 import os
 import subprocess
 import sys
-from typing import List
+from typing import List, Optional
 
 
 def check_env_vars(var_names: List[str]) -> tuple[str, ...]:
@@ -26,15 +26,26 @@ def check_env_vars(var_names: List[str]) -> tuple[str, ...]:
     return tuple(value_list)
 
 
-def run_cmd(cmd: List[str], *, check_results: bool = True) -> subprocess.CompletedProcess[str]:
-    """Run a command with subprocess and catch any CalledProcessErrors."""
+def run_cmd(
+    cmd: List[str], *, check_results: bool = True, timeout: Optional[float] = None
+) -> subprocess.CompletedProcess[str]:
+    """Run a command with subprocess and catch any CalledProcessErrors.
+
+    Args:
+        cmd: the command, as an argument list, to run.
+        check_results: if true, exit when the command returns a non-zero code.
+        timeout: if set, the command is killed and subprocess.TimeoutExpired is
+                 raised when it runs longer than this many seconds.  Callers that
+                 pass a timeout are responsible for handling TimeoutExpired.
+    """
     try:
         return subprocess.run(
             cmd,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             universal_newlines=True,
             check=check_results,
+            timeout=timeout,
         )
     except subprocess.CalledProcessError as err:
         print("CalledProcessError")