clearml_agent/commands/worker.py
def run_one_task(self, queue, task_id, worker_args, docker=None, task_session=None):
        # type: (Text, Text, WorkerParams, Optional[Text]) -> int
        """
        Run one task pulled from queue.
        :param queue: ID of queue that task was pulled from
        :param task_id: ID of task to run
        :param worker_args: Worker command line arguments
        :param task_session: The session for running operations on the passed task
        :param docker: Docker image in which the execution task will run

        :return: exit code (0 is success)
        """
        # start new process and execute task id
        # "Running task '{}'".format(task_id)
        print(self._task_logging_start_message.format(task_id))
        task_session = task_session or self._session
        # set task status to in_progress so we know it was popped from the queue
        # noinspection PyBroadException
        try:
            task_session.send_api(tasks_api.StartedRequest(task=task_id, status_message="pulled by agent", force=True))
        except Exception:
            print("Warning: Could not start task id '{}', skipping".format(task_id))
            return
        # setup console log
        temp_stdout_name = safe_mkstemp(
            suffix=".txt", prefix=".clearml_agent_out.", name_only=True
        )
        # temp_stderr_name = safe_mkstemp(suffix=".txt", prefix=".clearml_agent_err.", name_only=True)
        temp_stderr_name = None
        print(
            "Storing stdout and stderr log to '{}', '{}'".format(
                temp_stdout_name, temp_stderr_name or temp_stdout_name
            )
        )

        docker_image = None
        worker_id = '{}:service:{}'.format(self.worker_id, task_id) \
            if self._services_mode and not self._dynamic_gpus else self.worker_id

        if self.docker_image_func:
            # noinspection PyBroadException
            try:
                task_container = get_task_container(task_session, task_id)
            except Exception:
                task_container = {}

            default_docker = not bool(task_container.get('image'))
            docker_image = task_container.get('image') or self._docker_image
            docker_arguments = task_container.get(
                'arguments', self._docker_arguments if default_docker else None)
            docker_setup_script = task_container.get('setup_shell_script')

            self.send_logs(
                task_id=task_id,
                lines=
                ['Running Task {} inside {}docker: {} arguments: {}\n'.format(
                    task_id, "default " if default_docker else '',
                    docker_image, self._sanitize_docker_command(docker_arguments or []))]
                + (['custom_setup_bash_script:\n{}'.format(docker_setup_script)] if docker_setup_script else []),
                level="INFO",
                session=task_session,
            )

            # Update docker command
            docker_params = dict(
                docker_image=docker_image,
                docker_arguments=docker_arguments,
                docker_bash_setup_script=docker_setup_script,
            )
            if self._impersonate_as_task_owner:
                docker_params["auth_token"] = task_session.token
            elif self._session.access_key is None or self._session.secret_key is None:
                # We're using a token right now
                docker_params["auth_token"] = self._session.token
            if self._worker_tags:
                docker_params["worker_tags"] = self._worker_tags
            if self._services_mode:
                # if this is services mode, give the docker a unique worker id, as it will register itself.
                docker_params["worker_id"] = worker_id

            name_format = self._session.config.get('agent.docker_container_name_format', None)
            if name_format:
                try:
                    name = name_format.format(
                        task_id=re.sub(r'[^a-zA-Z0-9._-]', '-', task_id),
                        worker_id=re.sub(r'[^a-zA-Z0-9._-]', '-', worker_id),
                        rand_string="".join(sys_random.choice(string.ascii_lowercase) for _ in range(32))
                    )
                except Exception as ex:
                    print("Warning: failed generating docker container name: {}".format(ex))
                else:
                    if self._valid_docker_container_name(name):
                        docker_params["name"] = name
                    else:
                        print("Warning: generated docker container name is invalid: {}".format(name))

            full_docker_cmd = self.docker_image_func(env_task_id=task_id, **docker_params)

            # if we are using the default docker, update back the Task:
            if default_docker:
                # noinspection PyBroadException
                try:
                    set_task_container(
                        task_session,
                        task_id=task_id,
                        docker_image=docker_image,
                        docker_arguments=docker_arguments,
                        docker_bash_script=docker_setup_script,
                    )
                except Exception:
                    pass

            # if this is services_mode, change the worker_id to a unique name
            # abd use full-monitoring, ot it registers itself as a worker for this specific service.
            # notice, the internal agent will monitor itself once the docker is up and running
            full_docker_cmd[-1] = full_docker_cmd[-1] + 'execute {} {} --id {}'.format(
                '--full-monitoring' if self._services_mode else '--disable-monitoring',
                '--standalone-mode' if self._standalone_mode else '',
                task_id)

            display_docker_command = self._sanitize_docker_command(full_docker_cmd)

            # send the actual used command line to the backend
            self.send_logs(
                task_id=task_id,
                lines=['Executing: {}\n'.format(display_docker_command)],
                level="INFO",
                session=task_session,
            )

            cmd = Argv(*full_docker_cmd, display_argv=display_docker_command)

            print('Running Docker:\n{}\n'.format(str(cmd)))
        else:
            cmd = worker_args.get_argv_for_command("execute") + (
                '--full-monitoring' if self._services_mode else '--disable-monitoring',
                "--id",
                task_id,
            )

        events_service = self.get_service(Events)
        stop_signal = TaskStopSignal(
            command=self,
            session=task_session,
            events_service=events_service,
            task_id=task_id,
        )
        stop_signal_status = TaskStopSignal.default
        status = ExitStatus.interrupted
        try:
            # set WORKER ID
            ENV_WORKER_ID.set(worker_id)

            if self._docker_force_pull and docker_image:
                full_pull_cmd = ['docker', 'pull', docker_image]
                pull_cmd = Argv(*full_pull_cmd)
                status, stop_signal_status = self._log_command_output(
                    task_id=task_id,
                    cmd=pull_cmd,
                    stdout_path=temp_stdout_name,
                    stderr_path=temp_stderr_name,
                    daemon=True,
                    stop_signal=stop_signal,
                    session=task_session,
                )

            env = os.environ.copy()
            if self._impersonate_as_task_owner:
                env[ENV_AGENT_AUTH_TOKEN.vars[0]] = task_session.token
            if self._worker_tags:
                env[ENV_WORKER_TAGS.vars[0]] = " ".join(shlex.quote(t) for t in self._worker_tags)

            status, stop_signal_status = self._log_command_output(
                task_id=task_id,
                cmd=cmd,
                stdout_path=temp_stdout_name,
                stderr_path=temp_stderr_name,
                daemon=True,
                stop_signal=stop_signal,
                env=env,
                session=task_session,
            )
            errors = temp_stderr_name and Path(temp_stderr_name).read_text()
            if errors:
                print("\nEncountered errors:\n\n{}\n".format(errors))
            if status is None and self._services_mode:
                print(
                    "Service bootstrap completed: task '{}'".format(task_id)
                )
            elif status is None:
                print(
                    "DONE: Running task '{}' (user aborted)".format(task_id)
                )
            else:
                print("DONE: Running task '{}', exit status {}".format(task_id, status))
        except KeyboardInterrupt:
            self.handle_user_abort(task_id, session=task_session)
            status = ExitStatus.interrupted
        finally:
            if self._services_mode and stop_signal_status is None:
                print('Service started, docker running in the background')
            else:
                self.handle_task_termination(task_id, status, stop_signal_status, session=task_session)
                # remove temp files after we sent everything to the backend
                if self.docker_image_func:
                    shutdown_docker_process(docker_cmd_contains='--id {}\'\"'.format(task_id))
                safe_remove_file(temp_stdout_name)
                safe_remove_file(temp_stderr_name)
                if self._services_mode and status == ExitStatus.interrupted:
                    # unregister this worker, it was killed
                    self._unregister()

        return status

        Run one task pulled from queue.
        :param queue: ID of queue that task was pulled from
        :param task_id: ID of task to run
        :param worker_args: Worker command line arguments
        :param task_session: The session for running operations on the passed task
        :param docker: Docker image in which the execution task will run

        :return: exit code (0 is success)
        
Run one task pulled from queue.
Run one task pulled from queue .
def run_one_task(self, queue, task_id, worker_args, docker=None, task_session=None):
        print(self._task_logging_start_message.format(task_id))
        task_session = task_session or self._session
        try:
            task_session.send_api(tasks_api.StartedRequest(task=task_id, status_message="pulled by agent", force=True))
        except Exception:
            print("Warning: Could not start task id '{}', skipping".format(task_id))
            return
        temp_stdout_name = safe_mkstemp(
            suffix=".txt", prefix=".clearml_agent_out.", name_only=True
        )
        temp_stderr_name = None
        print(
            "Storing stdout and stderr log to '{}', '{}'".format(
                temp_stdout_name, temp_stderr_name or temp_stdout_name
            )
        )
        docker_image = None
        worker_id = '{}:service:{}'.format(self.worker_id, task_id) \
            if self._services_mode and not self._dynamic_gpus else self.worker_id
        if self.docker_image_func:
            try:
                task_container = get_task_container(task_session, task_id)
            except Exception:
                task_container = {}
            default_docker = not bool(task_container.get('image'))
            docker_image = task_container.get('image') or self._docker_image
            docker_arguments = task_container.get(
                'arguments', self._docker_arguments if default_docker else None)
            docker_setup_script = task_container.get('setup_shell_script')
            self.send_logs(
                task_id=task_id,
                lines=
                ['Running Task {} inside {}docker: {} arguments: {}\n'.format(
                    task_id, "default " if default_docker else '',
                    docker_image, self._sanitize_docker_command(docker_arguments or []))]
                + (['custom_setup_bash_script:\n{}'.format(docker_setup_script)] if docker_setup_script else []),
                level="INFO",
                session=task_session,
            )
            docker_params = dict(
                docker_image=docker_image,
                docker_arguments=docker_arguments,
                docker_bash_setup_script=docker_setup_script,
            )
            if self._impersonate_as_task_owner:
                docker_params["auth_token"] = task_session.token
            elif self._session.access_key is None or self._session.secret_key is None:
                docker_params["auth_token"] = self._session.token
            if self._worker_tags:
                docker_params["worker_tags"] = self._worker_tags
            if self._services_mode:
                docker_params["worker_id"] = worker_id
            name_format = self._session.config.get('agent.docker_container_name_format', None)
            if name_format:
                try:
                    name = name_format.format(
                        task_id=re.sub(r'[^a-zA-Z0-9._-]', '-', task_id),
                        worker_id=re.sub(r'[^a-zA-Z0-9._-]', '-', worker_id),
                        rand_string="".join(sys_random.choice(string.ascii_lowercase) for _ in range(32))
                    )
                except Exception as ex:
                    print("Warning: failed generating docker container name: {}".format(ex))
                else:
                    if self._valid_docker_container_name(name):
                        docker_params["name"] = name
                    else:
                        print("Warning: generated docker container name is invalid: {}".format(name))
            full_docker_cmd = self.docker_image_func(env_task_id=task_id, **docker_params)
            if default_docker:
                try:
                    set_task_container(
                        task_session,
                        task_id=task_id,
                        docker_image=docker_image,
                        docker_arguments=docker_arguments,
                        docker_bash_script=docker_setup_script,
                    )
                except Exception:
                    pass
            full_docker_cmd[-1] = full_docker_cmd[-1] + 'execute {} {} --id {}'.format(
                '--full-monitoring' if self._services_mode else '--disable-monitoring',
                '--standalone-mode' if self._standalone_mode else '',
                task_id)
            display_docker_command = self._sanitize_docker_command(full_docker_cmd)
            self.send_logs(
                task_id=task_id,
                lines=['Executing: {}\n'.format(display_docker_command)],
                level="INFO",
                session=task_session,
            )
            cmd = Argv(*full_docker_cmd, display_argv=display_docker_command)
            print('Running Docker:\n{}\n'.format(str(cmd)))
        else:
            cmd = worker_args.get_argv_for_command("execute") + (
                '--full-monitoring' if self._services_mode else '--disable-monitoring',
                "--id",
                task_id,
            )
        events_service = self.get_service(Events)
        stop_signal = TaskStopSignal(
            command=self,
            session=task_session,
            events_service=events_service,
            task_id=task_id,
        )
        stop_signal_status = TaskStopSignal.default
        status = ExitStatus.interrupted
        try:
            ENV_WORKER_ID.set(worker_id)
            if self._docker_force_pull and docker_image:
                full_pull_cmd = ['docker', 'pull', docker_image]
                pull_cmd = Argv(*full_pull_cmd)
                status, stop_signal_status = self._log_command_output(
                    task_id=task_id,
                    cmd=pull_cmd,
                    stdout_path=temp_stdout_name,
                    stderr_path=temp_stderr_name,
                    daemon=True,
                    stop_signal=stop_signal,
                    session=task_session,
                )
            env = os.environ.copy()
            if self._impersonate_as_task_owner:
                env[ENV_AGENT_AUTH_TOKEN.vars[0]] = task_session.token
            if self._worker_tags:
                env[ENV_WORKER_TAGS.vars[0]] = " ".join(shlex.quote(t) for t in self._worker_tags)
            status, stop_signal_status = self._log_command_output(
                task_id=task_id,
                cmd=cmd,
                stdout_path=temp_stdout_name,
                stderr_path=temp_stderr_name,
                daemon=True,
                stop_signal=stop_signal,
                env=env,
                session=task_session,
            )
            errors = temp_stderr_name and Path(temp_stderr_name).read_text()
            if errors:
                print("\nEncountered errors:\n\n{}\n".format(errors))
            if status is None and self._services_mode:
                print(
                    "Service bootstrap completed: task '{}'".format(task_id)
                )
            elif status is None:
                print(
                    "DONE: Running task '{}' (user aborted)".format(task_id)
                )
            else:
                print("DONE: Running task '{}', exit status {}".format(task_id, status))
        except KeyboardInterrupt:
            self.handle_user_abort(task_id, session=task_session)
            status = ExitStatus.interrupted
        finally:
            if self._services_mode and stop_signal_status is None:
                print('Service started, docker running in the background')
            else:
                self.handle_task_termination(task_id, status, stop_signal_status, session=task_session)
                if self.docker_image_func:
                    shutdown_docker_process(docker_cmd_contains='--id {}\'\"'.format(task_id))
                safe_remove_file(temp_stdout_name)
                safe_remove_file(temp_stderr_name)
                if self._services_mode and status == ExitStatus.interrupted:
                    self._unregister()
        return status
def run_one_task ( self , queue , task_id , worker_args , docker = None , task_session = None ) : print ( self . _task_logging_start_message . format ( task_id ) ) task_session = task_session or self . _session try : task_session . send_api ( tasks_api . StartedRequest ( task = task_id , status_message = "pulled by agent" , force = True ) ) except Exception : print ( "Warning: Could not start task id '{}', skipping" . format ( task_id ) ) return temp_stdout_name = safe_mkstemp ( suffix = ".txt" , prefix = ".clearml_agent_out." , name_only = True ) temp_stderr_name = None print ( "Storing stdout and stderr log to '{}', '{}'" . format ( temp_stdout_name , temp_stderr_name or temp_stdout_name ) ) docker_image = None worker_id = '{}:service:{}' . format ( self . worker_id , task_id ) if self . _services_mode and not self . _dynamic_gpus else self . worker_id if self . docker_image_func : try : task_container = get_task_container ( task_session , task_id ) except Exception : task_container = { } default_docker = not bool ( task_container . get ( 'image' ) ) docker_image = task_container . get ( 'image' ) or self . _docker_image docker_arguments = task_container . get ( 'arguments' , self . _docker_arguments if default_docker else None ) docker_setup_script = task_container . get ( 'setup_shell_script' ) self . send_logs ( task_id = task_id , lines = [ 'Running Task {} inside {}docker: {} arguments: {}\n' . format ( task_id , "default " if default_docker else '' , docker_image , self . _sanitize_docker_command ( docker_arguments or [ ] ) ) ] + ( [ 'custom_setup_bash_script:\n{}' . format ( docker_setup_script ) ] if docker_setup_script else [ ] ) , level = "INFO" , session = task_session , ) docker_params = dict ( docker_image = docker_image , docker_arguments = docker_arguments , docker_bash_setup_script = docker_setup_script , ) if self . _impersonate_as_task_owner : docker_params [ "auth_token" ] = task_session . token elif self . _session . access_key is None or self . _session . secret_key is None : docker_params [ "auth_token" ] = self . _session . token if self . _worker_tags : docker_params [ "worker_tags" ] = self . _worker_tags if self . _services_mode : docker_params [ "worker_id" ] = worker_id name_format = self . _session . config . get ( 'agent.docker_container_name_format' , None ) if name_format : try : name = name_format . format ( task_id = re . sub ( r'[^a-zA-Z0-9._-]' , '-' , task_id ) , worker_id = re . sub ( r'[^a-zA-Z0-9._-]' , '-' , worker_id ) , rand_string = "" . join ( sys_random . choice ( string . ascii_lowercase ) for _ in range ( 32 ) ) ) except Exception as ex : print ( "Warning: failed generating docker container name: {}" . format ( ex ) ) else : if self . _valid_docker_container_name ( name ) : docker_params [ "name" ] = name else : print ( "Warning: generated docker container name is invalid: {}" . format ( name ) ) full_docker_cmd = self . docker_image_func ( env_task_id = task_id , ** docker_params ) if default_docker : try : set_task_container ( task_session , task_id = task_id , docker_image = docker_image , docker_arguments = docker_arguments , docker_bash_script = docker_setup_script , ) except Exception : pass full_docker_cmd [ - 1 ] = full_docker_cmd [ - 1 ] + 'execute {} {} --id {}' . format ( '--full-monitoring' if self . _services_mode else '--disable-monitoring' , '--standalone-mode' if self . _standalone_mode else '' , task_id ) display_docker_command = self . _sanitize_docker_command ( full_docker_cmd ) self . send_logs ( task_id = task_id , lines = [ 'Executing: {}\n' . format ( display_docker_command ) ] , level = "INFO" , session = task_session , ) cmd = Argv ( * full_docker_cmd , display_argv = display_docker_command ) print ( 'Running Docker:\n{}\n' . format ( str ( cmd ) ) ) else : cmd = worker_args . get_argv_for_command ( "execute" ) + ( '--full-monitoring' if self . _services_mode else '--disable-monitoring' , "--id" , task_id , ) events_service = self . get_service ( Events ) stop_signal = TaskStopSignal ( command = self , session = task_session , events_service = events_service , task_id = task_id , ) stop_signal_status = TaskStopSignal . default status = ExitStatus . interrupted try : ENV_WORKER_ID . set ( worker_id ) if self . _docker_force_pull and docker_image : full_pull_cmd = [ 'docker' , 'pull' , docker_image ] pull_cmd = Argv ( * full_pull_cmd ) status , stop_signal_status = self . _log_command_output ( task_id = task_id , cmd = pull_cmd , stdout_path = temp_stdout_name , stderr_path = temp_stderr_name , daemon = True , stop_signal = stop_signal , session = task_session , ) env = os . environ . copy ( ) if self . _impersonate_as_task_owner : env [ ENV_AGENT_AUTH_TOKEN . vars [ 0 ] ] = task_session . token if self . _worker_tags : env [ ENV_WORKER_TAGS . vars [ 0 ] ] = " " . join ( shlex . quote ( t ) for t in self . _worker_tags ) status , stop_signal_status = self . _log_command_output ( task_id = task_id , cmd = cmd , stdout_path = temp_stdout_name , stderr_path = temp_stderr_name , daemon = True , stop_signal = stop_signal , env = env , session = task_session , ) errors = temp_stderr_name and Path ( temp_stderr_name ) . read_text ( ) if errors : print ( "\nEncountered errors:\n\n{}\n" . format ( errors ) ) if status is None and self . _services_mode : print ( "Service bootstrap completed: task '{}'" . format ( task_id ) ) elif status is None : print ( "DONE: Running task '{}' (user aborted)" . format ( task_id ) ) else : print ( "DONE: Running task '{}', exit status {}" . format ( task_id , status ) ) except KeyboardInterrupt : self . handle_user_abort ( task_id , session = task_session ) status = ExitStatus . interrupted finally : if self . _services_mode and stop_signal_status is None : print ( 'Service started, docker running in the background' ) else : self . handle_task_termination ( task_id , status , stop_signal_status , session = task_session ) if self . docker_image_func : shutdown_docker_process ( docker_cmd_contains = '--id {}\'\"' . format ( task_id ) ) safe_remove_file ( temp_stdout_name ) safe_remove_file ( temp_stderr_name ) if self . _services_mode and status == ExitStatus . interrupted : self . _unregister ( ) return status
Run one task pulled from queue.
Run one task pulled from queue .
# type: (Text, Text, WorkerParams, Optional[Text]) -> int """
        Run one task pulled from queue.
        :param queue: ID of queue that task was pulled from
        :param task_id: ID of task to run
        :param worker_args: Worker command line arguments
        :param task_session: The session for running operations on the passed task
        :param docker: Docker image in which the execution task will run

        :return: exit code (0 is success)
        """ # start new process and execute task id # "Running task '{}'".format(task_id) # set task status to in_progress so we know it was popped from the queue # noinspection PyBroadException # setup console log # temp_stderr_name = safe_mkstemp(suffix=".txt", prefix=".clearml_agent_err.", name_only=True) # noinspection PyBroadException # Update docker command # We're using a token right now # if this is services mode, give the docker a unique worker id, as it will register itself. # if we are using the default docker, update back the Task: # noinspection PyBroadException # if this is services_mode, change the worker_id to a unique name # abd use full-monitoring, ot it registers itself as a worker for this specific service. # notice, the internal agent will monitor itself once the docker is up and running # send the actual used command line to the backend # set WORKER ID # remove temp files after we sent everything to the backend # unregister this worker, it was killed
exit code (0 is success)
exit code ( 0 is success )
self
queue
ID of queue that task was pulled from
ID of queue that task was pulled from
task_id
ID of task to run
ID of task to run
worker_args
Worker command line arguments
Worker command line arguments
docker
Docker image in which the execution task will run
Docker image in which the execution task will run
task_session
The session for running operations on the passed task
The session for running operations on the passed task