ClearML/clearml-agent
1
0
Fork 0
mirror of https://github.com/clearml/clearml-agent synced 2025-02-12 07:38:04 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Fix "execute --clone" support

Browse Source
This commit is contained in:
allegroai 2020-03-12 18:38:35 +02:00
parent aa75b92e46
commit 15e9e6b778
1 changed files with 19 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
trains_agent/commands/worker.py
View File

@ -82,7 +82,11 @@ from trains_agent.helper.process import (
Argv, Argv,
COMMAND_SUCCESS, COMMAND_SUCCESS,
Executable, Executable,
get_bash_output, shutdown_docker_process, get_docker_id, commit_docker) get_bash_output,
shutdown_docker_process,
get_docker_id,
commit_docker
)
from trains_agent.helper.package.cython_req import CythonRequirement from trains_agent.helper.package.cython_req import CythonRequirement
from trains_agent.helper.repo import clone_repository_cached, RepoInfo, VCS from trains_agent.helper.repo import clone_repository_cached, RepoInfo, VCS
from trains_agent.helper.resource_monitor import ResourceMonitor from trains_agent.helper.resource_monitor import ResourceMonitor
@ -1096,8 +1100,9 @@ class Worker(ServiceCommandSection):
debug=self._session.debug_mode, debug=self._session.debug_mode,
trace=self._session.trace, trace=self._session.trace,
) )
self.report_monitor(ResourceMonitor.StatusReport(task=task_id)) self.report_monitor(ResourceMonitor.StatusReport(task=current_task.id))
self.run_one_task(queue='', task_id=task_id, worker_args=worker_params, docker=docker) self.run_one_task(queue='', task_id=current_task.id, worker_args=worker_params, docker=docker)
self.stop_monitor() self.stop_monitor()
self._unregister() self._unregister()
return return
@ -1113,7 +1118,7 @@ class Worker(ServiceCommandSection):
if not disable_monitoring: if not disable_monitoring:
self.log.debug("starting resource monitor") self.log.debug("starting resource monitor")
self.report_monitor(ResourceMonitor.StatusReport(task=task_id)) self.report_monitor(ResourceMonitor.StatusReport(task=current_task.id))
execution = self.get_execution_info(current_task) execution = self.get_execution_info(current_task)
@ -1161,7 +1166,7 @@ class Worker(ServiceCommandSection):
script_dir = (directory if isinstance(directory, Path) else Path(directory)).absolute().as_posix() script_dir = (directory if isinstance(directory, Path) else Path(directory)).absolute().as_posix()
# run code # run code
print("Running task id [%s]:" % task_id) print("Running task id [%s]:" % current_task.id)
extra = ['-u', ] extra = ['-u', ]
if optimization: if optimization:
extra.append( extra.append(
@ -1195,7 +1200,7 @@ class Worker(ServiceCommandSection):
) )
if repo_info: if repo_info:
self._update_commit_id(task_id, execution, repo_info) self._update_commit_id(current_task.id, execution, repo_info)
# Add the script CWD to the python path # Add the script CWD to the python path
python_path = get_python_path(script_dir, execution.entry_point, self.package_api) \ python_path = get_python_path(script_dir, execution.entry_point, self.package_api) \
@ -1213,7 +1218,7 @@ class Worker(ServiceCommandSection):
else: else:
use_execv = is_linux_platform() and not isinstance(self.package_api, (PoetryAPI, CondaAPI)) use_execv = is_linux_platform() and not isinstance(self.package_api, (PoetryAPI, CondaAPI))
print("Starting Task Execution:\n".format(task_id)) print("Starting Task Execution:\n".format(current_task.id))
exit_code = -1 exit_code = -1
try: try:
if disable_monitoring: if disable_monitoring:
@ -1242,13 +1247,13 @@ class Worker(ServiceCommandSection):
) )
print("Storing stdout and stderr log into [%s]" % temp_stdout_fname) print("Storing stdout and stderr log into [%s]" % temp_stdout_fname)
exit_code, _ = self._log_command_output( exit_code, _ = self._log_command_output(
task_id=task_id, task_id=current_task.id,
cmd=command, cmd=command,
stdout_path=temp_stdout_fname, stdout_path=temp_stdout_fname,
cwd=script_dir, cwd=script_dir,
) )
except KeyboardInterrupt: except KeyboardInterrupt:
self.handle_user_abort(task_id) self.handle_user_abort(current_task.id)
raise raise
except Exception as e: except Exception as e:
self.log.warning(str(e)) self.log.warning(str(e))
@ -1265,7 +1270,7 @@ class Worker(ServiceCommandSection):
if not disable_monitoring: if not disable_monitoring:
# we need to change task status according to exit code # we need to change task status according to exit code
self.handle_task_termination(task_id, exit_code, TaskStopReason.no_stop) self.handle_task_termination(current_task.id, exit_code, TaskStopReason.no_stop)
self.stop_monitor() self.stop_monitor()
# unregister the worker # unregister the worker
self._unregister() self._unregister()
@ -1863,6 +1868,10 @@ class Worker(ServiceCommandSection):
cmds = [cmds] cmds = [cmds]
extra_shell_script_str = " ; ".join(map(str, cmds)) + " ; " extra_shell_script_str = " ; ".join(map(str, cmds)) + " ; "
self.temp_config_path = self.temp_config_path or safe_mkstemp(
suffix=".cfg", prefix=".trains_agent.", text=True, name_only=True
)
docker_cmd = dict(worker_id=self.worker_id, docker_cmd = dict(worker_id=self.worker_id,
# docker_image=docker_image, # docker_image=docker_image,
# docker_arguments=docker_arguments, # docker_arguments=docker_arguments,