shell_command: refactor retrires to attempts
This is consistent with the http_client. The argument "attempts" is more accurate than retries, as the first attempt is not a retry. Signed-off-by: Eric Callahan <arksine.code@gmail.com>
This commit is contained in:
Eric Callahan
parent
eee1eda6bc
commit
f1de614027
|
|
@ -460,7 +460,7 @@ class Machine:
|
|||
full_cmd = f"sudo -S {command}"
|
||||
shell_cmd: SCMDComp = self.server.lookup_component("shell_command")
|
||||
return await shell_cmd.exec_cmd(
|
||||
full_cmd, proc_input=proc_input, log_complete=False, retries=tries,
|
||||
full_cmd, proc_input=proc_input, log_complete=False, attempts=tries,
|
||||
timeout=timeout
|
||||
)
|
||||
|
||||
|
|
@ -959,7 +959,7 @@ class SystemdCliProvider(BaseProvider):
|
|||
)
|
||||
prop_args = ",".join(properties)
|
||||
props: str = await self.shell_cmd.exec_cmd(
|
||||
f"systemctl show -p {prop_args} {unit_name}", retries=5,
|
||||
f"systemctl show -p {prop_args} {unit_name}", attempts=5,
|
||||
timeout=10.
|
||||
)
|
||||
raw_props: Dict[str, Any] = {}
|
||||
|
|
@ -1301,7 +1301,7 @@ class SupervisordCliProvider(BaseProvider):
|
|||
else:
|
||||
cmd = f"supervisorctl {args}"
|
||||
return await self.shell_cmd.exec_cmd(
|
||||
cmd, proc_input=None, log_complete=False, retries=tries,
|
||||
cmd, proc_input=None, log_complete=False, attempts=tries,
|
||||
timeout=timeout, success_codes=success_codes
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -228,7 +228,7 @@ class ShellCommand:
|
|||
async def run_with_response(
|
||||
self,
|
||||
timeout: float = 2.,
|
||||
retries: int = 1,
|
||||
attempts: int = 1,
|
||||
log_complete: bool = True,
|
||||
sig_idx: int = 1,
|
||||
proc_input: Optional[str] = None,
|
||||
|
|
@ -236,11 +236,11 @@ class ShellCommand:
|
|||
) -> str:
|
||||
async with self.run_lock:
|
||||
self.factory.add_running_command(self)
|
||||
retries = max(1, retries)
|
||||
attempts = max(1, attempts)
|
||||
stdin: Optional[bytes] = None
|
||||
if proc_input is not None:
|
||||
stdin = proc_input.encode()
|
||||
while retries > 0:
|
||||
while attempts > 0:
|
||||
self._reset_command_data()
|
||||
timed_out = False
|
||||
stdout = stderr = b""
|
||||
|
|
@ -271,7 +271,7 @@ class ShellCommand:
|
|||
f"\n{stdout.decode(errors='ignore')}")
|
||||
if self.cancelled and not timed_out:
|
||||
break
|
||||
retries -= 1
|
||||
attempts -= 1
|
||||
await asyncio.sleep(.5)
|
||||
self.factory.remove_running_command(self)
|
||||
raise ShellCommandError(
|
||||
|
|
@ -375,7 +375,7 @@ class ShellCommandFactory:
|
|||
callback: OutputCallback = None,
|
||||
std_err_callback: OutputCallback = None,
|
||||
timeout: float = 2.,
|
||||
retries: int = 1,
|
||||
attempts: int = 1,
|
||||
verbose: bool = True,
|
||||
sig_idx: int = 1,
|
||||
proc_input: Optional[str] = None,
|
||||
|
|
@ -392,9 +392,9 @@ class ShellCommandFactory:
|
|||
scmd = ShellCommand(
|
||||
self, cmd, callback, std_err_callback, env, log_stderr, cwd
|
||||
)
|
||||
retries = max(1, retries)
|
||||
attempts = max(1, attempts)
|
||||
async def _wrapper() -> None:
|
||||
for _ in range(retries):
|
||||
for _ in range(attempts):
|
||||
if await scmd.run(
|
||||
timeout, verbose, log_complete, sig_idx,
|
||||
proc_input, success_codes
|
||||
|
|
@ -409,7 +409,7 @@ class ShellCommandFactory:
|
|||
self,
|
||||
cmd: str,
|
||||
timeout: float = 2.,
|
||||
retries: int = 1,
|
||||
attempts: int = 1,
|
||||
sig_idx: int = 1,
|
||||
proc_input: Optional[str] = None,
|
||||
log_complete: bool = True,
|
||||
|
|
@ -424,7 +424,7 @@ class ShellCommandFactory:
|
|||
scmd = ShellCommand(self, cmd, None, None, env,
|
||||
log_stderr, cwd)
|
||||
coro = scmd.run_with_response(
|
||||
timeout, retries, log_complete, sig_idx,
|
||||
timeout, attempts, log_complete, sig_idx,
|
||||
proc_input, success_codes
|
||||
)
|
||||
return asyncio.create_task(coro)
|
||||
|
|
|
|||
|
|
@ -424,7 +424,7 @@ class AppDeploy(BaseDeploy):
|
|||
try:
|
||||
await self.cmd_helper.run_cmd(
|
||||
f"{self.pip_cmd} install {args}", timeout=1200., notify=True,
|
||||
retries=3, env=env, log_stderr=True
|
||||
attempts=3, env=env, log_stderr=True
|
||||
)
|
||||
except Exception:
|
||||
self.log_exc("Error updating python requirements")
|
||||
|
|
@ -442,7 +442,7 @@ class AppDeploy(BaseDeploy):
|
|||
try:
|
||||
await self.cmd_helper.run_cmd(
|
||||
f"{self.pip_cmd} install pip=={update_ver}",
|
||||
timeout=1200., notify=True, retries=3
|
||||
timeout=1200., notify=True, attempts=3
|
||||
)
|
||||
except Exception:
|
||||
self.log_exc("Error updating python pip")
|
||||
|
|
@ -452,8 +452,9 @@ class AppDeploy(BaseDeploy):
|
|||
return None
|
||||
self.notify_status("Checking pip version...")
|
||||
try:
|
||||
data: str = await self.cmd_helper.run_cmd_with_response(
|
||||
f"{self.pip_cmd} --version", timeout=30., retries=3
|
||||
scmd = self.cmd_helper.get_shell_command()
|
||||
data: str = await scmd.exec_cmd(
|
||||
f"{self.pip_cmd} --version", timeout=30., attempts=3
|
||||
)
|
||||
match = re.match(
|
||||
r"^pip ([0-9.]+) from .+? \(python ([0-9.]+)\)$", data.strip()
|
||||
|
|
|
|||
|
|
@ -217,7 +217,8 @@ class GitDeploy(AppDeploy):
|
|||
try:
|
||||
await self.cmd_helper.run_cmd(
|
||||
"npm ci --only=prod", notify=True, timeout=600.,
|
||||
cwd=str(self.path))
|
||||
cwd=str(self.path)
|
||||
)
|
||||
except Exception:
|
||||
self.notify_status("Node Package Update failed")
|
||||
|
||||
|
|
@ -442,17 +443,17 @@ class GitRepo:
|
|||
" is not a valid git repo")
|
||||
return False
|
||||
await self._wait_for_lock_release()
|
||||
retries = 3
|
||||
while retries:
|
||||
attempts = 3
|
||||
while attempts:
|
||||
self.git_messages.clear()
|
||||
try:
|
||||
cmd = "status --porcelain -b"
|
||||
resp: Optional[str] = await self._run_git_cmd(cmd, retries=1)
|
||||
resp: Optional[str] = await self._run_git_cmd(cmd, attempts=1)
|
||||
except Exception:
|
||||
retries -= 1
|
||||
attempts -= 1
|
||||
resp = None
|
||||
# Attempt to recover from "loose object" error
|
||||
if retries and self.repo_corrupt:
|
||||
if attempts and self.repo_corrupt:
|
||||
if not await self._repair_loose_objects():
|
||||
# Since we are unable to recover, immediately
|
||||
# return
|
||||
|
|
@ -669,8 +670,8 @@ class GitRepo:
|
|||
async with self.git_operation_lock:
|
||||
for _ in range(3):
|
||||
try:
|
||||
await self._run_git_cmd(cmd, retries=1, corrupt_msg="error: ")
|
||||
except self.cmd_helper.scmd_error as err:
|
||||
await self._run_git_cmd(cmd, attempts=1, corrupt_msg="error: ")
|
||||
except self.cmd_helper.get_shell_command().error as err:
|
||||
if err.return_code == 1:
|
||||
return False
|
||||
if self.repo_corrupt:
|
||||
|
|
@ -788,7 +789,7 @@ class GitRepo:
|
|||
if self.git_remote == "?" or self.git_branch == "?":
|
||||
raise self.server.error("Cannot reset, unknown remote/branch")
|
||||
ref = f"{self.git_remote}/{self.git_branch}"
|
||||
await self._run_git_cmd(f"reset --hard {ref}", retries=2)
|
||||
await self._run_git_cmd(f"reset --hard {ref}", attempts=2)
|
||||
self.repo_corrupt = False
|
||||
|
||||
async def fetch(self) -> None:
|
||||
|
|
@ -800,7 +801,7 @@ class GitRepo:
|
|||
async def clean(self) -> None:
|
||||
self._verify_repo()
|
||||
async with self.git_operation_lock:
|
||||
await self._run_git_cmd("clean -d -f", retries=2)
|
||||
await self._run_git_cmd("clean -d -f", attempts=2)
|
||||
|
||||
async def pull(self) -> None:
|
||||
self._verify_repo()
|
||||
|
|
@ -859,7 +860,7 @@ class GitRepo:
|
|||
args = f"{cmd} {key} '{pattern}'" if pattern else f"{cmd} {key}"
|
||||
try:
|
||||
return await self.config_cmd(args)
|
||||
except self.cmd_helper.scmd_error as e:
|
||||
except self.cmd_helper.get_shell_command().error as e:
|
||||
if e.return_code == 1:
|
||||
return None
|
||||
raise
|
||||
|
|
@ -884,9 +885,9 @@ class GitRepo:
|
|||
for attempt in range(3):
|
||||
try:
|
||||
return await self._run_git_cmd(
|
||||
f"config {args}", retries=1, log_complete=verbose
|
||||
f"config {args}", attempts=1, log_complete=verbose
|
||||
)
|
||||
except self.cmd_helper.scmd_error as e:
|
||||
except self.cmd_helper.get_shell_command().error as e:
|
||||
if 1 <= (e.return_code or 10) <= 6 or attempt == 2:
|
||||
raise
|
||||
raise self.server.error("Failed to run git-config")
|
||||
|
|
@ -907,7 +908,7 @@ class GitRepo:
|
|||
|
||||
async def run_fsck(self) -> None:
|
||||
async with self.git_operation_lock:
|
||||
await self._run_git_cmd("fsck --full", timeout=300., retries=1)
|
||||
await self._run_git_cmd("fsck --full", timeout=300., attempts=1)
|
||||
|
||||
async def clone(self) -> None:
|
||||
if self.is_submodule_or_worktree():
|
||||
|
|
@ -1194,12 +1195,13 @@ class GitRepo:
|
|||
"Attempting to repair loose objects..."
|
||||
)
|
||||
try:
|
||||
await self.cmd_helper.run_cmd_with_response(
|
||||
shell_cmd = self.cmd_helper.get_shell_command()
|
||||
await shell_cmd.exec_cmd(
|
||||
"find .git/objects/ -type f -empty | xargs rm",
|
||||
timeout=10., retries=1, cwd=str(self.src_path))
|
||||
timeout=10., attempts=1, cwd=str(self.src_path))
|
||||
await self._run_git_cmd_async(
|
||||
"fetch --all -p", retries=1, fix_loose=False)
|
||||
await self._run_git_cmd("fsck --full", timeout=300., retries=1)
|
||||
"fetch --all -p", attempts=1, fix_loose=False)
|
||||
await self._run_git_cmd("fsck --full", timeout=300., attempts=1)
|
||||
except Exception:
|
||||
msg = (
|
||||
"Attempt to repair loose objects failed, "
|
||||
|
|
@ -1216,7 +1218,7 @@ class GitRepo:
|
|||
|
||||
async def _run_git_cmd_async(self,
|
||||
cmd: str,
|
||||
retries: int = 5,
|
||||
attempts: int = 5,
|
||||
need_git_path: bool = True,
|
||||
fix_loose: bool = True
|
||||
) -> None:
|
||||
|
|
@ -1231,10 +1233,11 @@ class GitRepo:
|
|||
git_cmd = f"git -C {self.src_path} {cmd}"
|
||||
else:
|
||||
git_cmd = f"git {cmd}"
|
||||
scmd = self.cmd_helper.build_shell_command(
|
||||
shell_cmd = self.cmd_helper.get_shell_command()
|
||||
scmd = shell_cmd.build_shell_command(
|
||||
git_cmd, callback=self._handle_process_output,
|
||||
env=env)
|
||||
while retries:
|
||||
while attempts:
|
||||
self.git_messages.clear()
|
||||
self.fetch_input_recd = False
|
||||
self.fetch_timeout_handle = event_loop.delay_callback(
|
||||
|
|
@ -1254,14 +1257,14 @@ class GitRepo:
|
|||
# Only attempt to repair loose objects once. Re-run
|
||||
# the command once.
|
||||
fix_loose = False
|
||||
retries = 2
|
||||
attempts = 2
|
||||
else:
|
||||
# since the attept to repair failed, bypass retries
|
||||
# since the attept to repair failed, bypass attempts
|
||||
# and immediately raise an exception
|
||||
raise self.server.error(
|
||||
"Unable to repair loose objects, use hard recovery"
|
||||
)
|
||||
retries -= 1
|
||||
attempts -= 1
|
||||
await asyncio.sleep(.5)
|
||||
await self._check_lock_file_exists(remove=True)
|
||||
raise self.server.error(f"Git Command '{cmd}' failed")
|
||||
|
|
@ -1303,21 +1306,22 @@ class GitRepo:
|
|||
self,
|
||||
git_args: str,
|
||||
timeout: float = 20.,
|
||||
retries: int = 5,
|
||||
attempts: int = 5,
|
||||
env: Optional[Dict[str, str]] = None,
|
||||
corrupt_msg: str = "fatal: ",
|
||||
log_complete: bool = True
|
||||
) -> str:
|
||||
shell_cmd = self.cmd_helper.get_shell_command()
|
||||
try:
|
||||
return await self.cmd_helper.run_cmd_with_response(
|
||||
return await shell_cmd.exec_cmd(
|
||||
f"git -C {self.src_path} {git_args}",
|
||||
timeout=timeout,
|
||||
retries=retries,
|
||||
attempts=attempts,
|
||||
env=env,
|
||||
sig_idx=2,
|
||||
log_complete=log_complete
|
||||
)
|
||||
except self.cmd_helper.scmd_error as e:
|
||||
except shell_cmd.error as e:
|
||||
stdout = e.stdout.decode().strip()
|
||||
stderr = e.stderr.decode().strip()
|
||||
msg_lines: List[str] = []
|
||||
|
|
|
|||
|
|
@ -183,8 +183,8 @@ class AptCliProvider(BasePackageProvider):
|
|||
f"{self.APT_CMD} update", timeout=600., notify=notify)
|
||||
|
||||
async def get_packages(self) -> List[str]:
|
||||
res = await self.cmd_helper.run_cmd_with_response(
|
||||
"apt list --upgradable", timeout=60.)
|
||||
shell_cmd = self.cmd_helper.get_shell_command()
|
||||
res = await shell_cmd.exec_cmd("apt list --upgradable", timeout=60.)
|
||||
pkg_list = [p.strip() for p in res.split("\n") if p.strip()]
|
||||
if pkg_list:
|
||||
pkg_list = pkg_list[2:]
|
||||
|
|
@ -195,7 +195,8 @@ class AptCliProvider(BasePackageProvider):
|
|||
self.cmd_helper.notify_update_response("Resolving packages...")
|
||||
search_regex = "|".join([f"^{pkg}$" for pkg in package_list])
|
||||
cmd = f"apt-cache search --names-only \"{search_regex}\""
|
||||
ret = await self.cmd_helper.run_cmd_with_response(cmd, timeout=600.)
|
||||
shell_cmd = self.cmd_helper.get_shell_command()
|
||||
ret = await shell_cmd.exec_cmd(cmd, timeout=600.)
|
||||
resolved = [
|
||||
pkg.strip().split()[0] for pkg in ret.split("\n") if pkg.strip()
|
||||
]
|
||||
|
|
@ -217,7 +218,7 @@ class AptCliProvider(BasePackageProvider):
|
|||
pkgs = " ".join(resolved)
|
||||
await self.cmd_helper.run_cmd(
|
||||
f"{self.APT_CMD} install --yes {pkgs}", timeout=timeout,
|
||||
retries=retries, notify=notify)
|
||||
attempts=retries, notify=notify)
|
||||
|
||||
async def upgrade_system(self) -> None:
|
||||
await self.cmd_helper.run_cmd(
|
||||
|
|
|
|||
|
|
@ -496,10 +496,6 @@ class CommandHelper:
|
|||
config.getboolean('enable_repo_debug', False, deprecate=True)
|
||||
if self.server.is_debug_enabled():
|
||||
logging.warning("UPDATE MANAGER: REPO DEBUG ENABLED")
|
||||
shell_cmd: SCMDComp = self.server.lookup_component('shell_command')
|
||||
self.scmd_error = shell_cmd.error
|
||||
self.build_shell_command = shell_cmd.build_shell_command
|
||||
self.run_cmd_with_response = shell_cmd.exec_cmd
|
||||
self.pkg_updater: Optional[PackageDeploy] = None
|
||||
|
||||
# database management
|
||||
|
|
@ -527,6 +523,9 @@ class CommandHelper:
|
|||
def get_server(self) -> Server:
|
||||
return self.server
|
||||
|
||||
def get_shell_command(self) -> SCMDComp:
|
||||
return self.server.lookup_component("shell_command")
|
||||
|
||||
def get_http_client(self) -> HttpClient:
|
||||
return self.http_client
|
||||
|
||||
|
|
@ -579,7 +578,7 @@ class CommandHelper:
|
|||
cmd: str,
|
||||
timeout: float = 20.,
|
||||
notify: bool = False,
|
||||
retries: int = 1,
|
||||
attempts: int = 1,
|
||||
env: Optional[Dict[str, str]] = None,
|
||||
cwd: Optional[str] = None,
|
||||
sig_idx: int = 1,
|
||||
|
|
@ -587,14 +586,10 @@ class CommandHelper:
|
|||
) -> None:
|
||||
cb = self.notify_update_response if notify else None
|
||||
log_stderr |= self.server.is_verbose_enabled()
|
||||
scmd = self.build_shell_command(
|
||||
cmd, callback=cb, env=env, cwd=cwd, log_stderr=log_stderr
|
||||
await self.get_shell_command().run_cmd_async(
|
||||
cmd, cb, timeout=timeout, attempts=attempts,
|
||||
env=env, cwd=cwd, sig_idx=sig_idx, log_stderr=log_stderr
|
||||
)
|
||||
for _ in range(retries):
|
||||
if await scmd.run(timeout=timeout, sig_idx=sig_idx):
|
||||
break
|
||||
else:
|
||||
raise self.server.error("Shell Command Error")
|
||||
|
||||
def notify_update_refreshed(self) -> None:
|
||||
vinfo: Dict[str, Any] = {}
|
||||
|
|
|
|||
Loading…
Reference in New Issue