o xiv@sdZddlmZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z ddl mZddlmZddlZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddl m!Z!m"Z"ddl#m$Z$ddl%m&Z&ddl'm(Z(ddl)m*Z*m+Z+ddl,m-Z-m.Z.m/Z/m0Z0ddl1m2Z2ddl3m4Z4dZ5dZ6dZ7dZ8dZ9dZ:dZ;dZd!Z?ej@Ad"ZBeBrzeCeBZDWneEye+d#eBwd$ZDeFeGZHe Gd%d&d&ZId>d*d+ZJd?d@d2d3ZKGd4d5d5ZL  dAdBd:d;ZMGdeZdZdddZdddZddd Zdd d Zdd d ZdS)InternalAgentLoggerrcCs|dk|_dS)Nr)_print_to_terminal)self verbosityr.r.r/__init__sszInternalAgentLogger.__init__messager(cC.|jr tt|tt|dSN)rEwandb termerrorr_loggererrorrFrIr.r.r/rOvzInternalAgentLogger.errorcCrJrK)rErLtermwarnrrNwarningrPr.r.r/warn{rQzInternalAgentLogger.warncCrJrK)rErLtermlogrrNinforPr.r.r/rVrQzInternalAgentLogger.infocCrJrK)rErLrUrrNdebugrPr.r.r/rWrQzInternalAgentLogger.debugN)r)rIr()r*r+r,rHrOrTrVrWr.r.r.r/rDrs     rD launch_config dict | None build_config|dr dS|did g}t|d kr2dS|d d d dgkr>dSdS)z5Determine whether a job/runSpec is a sweep scheduler.z4Received runSpec in _is_scheduler_job that was emptyuriFresourcez local-processr'T overrides entry_pointNrrL scheduler)rrWr=r PLACEHOLDER_URIlen)rFrcmdr.r.r/_is_scheduler_jobs  zLaunchAgent._is_scheduler_jobrun_queue_item_idrIphasefileslist[str] | NoneNonecs0|jrt|jj}|||||IdHdSdSrK)rrrfail_run_queue_item)rFrrIrrfail_rqir.r.r/r)s  zLaunchAgent.fail_run_queue_itemcCs@d|_|jrtjdddd}tj|j|j||jtd|_dSdS)NT)silent disable_gitdisable_job_creation)r|rzridjob_type) _wandb_runrrLSettingsinitrrrtHIDDEN_AGENT_RUN_TYPE)rFrr.r.r/r4s zLaunchAgent._init_agent_run list[int]cCs8|jt|jWdS1swYdS)z8Returns a list of keys running thread ids for the agent.N)rlistrkeysrFr.r.r/ thread_idsDs $zLaunchAgent.thread_idsr9cBjtfddjDWdS1swYdS)z%Return just the number of schedulers.csg|] }j|jr|qSr.r is_scheduler.0xrr.r/ Nz6LaunchAgent.num_running_schedulers..Nrrrrr.rr/num_running_schedulersJ$z"LaunchAgent.num_running_schedulerscr)z3Return the number of jobs not including schedulers.csg|] }j|js|qSr.rrrr.r/rTrz0LaunchAgent.num_running_jobs..Nrrr.rr/num_running_jobsPrzLaunchAgent.num_running_jobsr)c sbzt|jj}|||j|j|jdIdH}|WSty0}z td|WYd}~dSd}~ww)zPops an item off the runqueue to run as a job. Arguments: queue: Queue to pop from. Returns: Item popped off the queue. Raises: Exception: if there is an error popping from the queue. )rzr|agent_idNz Exception:)rrpop_from_run_queuerrr Exceptionprint)rFr)popupsrBr.r.r/pop_from_queueVs   zLaunchAgent.pop_from_queuecCst|_d}|jr||jd7}|j|jkr$|dd|jd7}|d|jd|jd7}tt ||jd krO|d dd d |j D7}t |d S)z'Prints the current status of the agent.agent  zpolling on queues ,z, zrunning z out of a maximum of z jobsrz: css|]}t|VqdSrK)r()rjob_idr.r.r/ }sz+LaunchAgent.print_status..N) timerrtrrjoinrrLrUrrrNrV)rF output_strr.r.r/ print_statusos    zLaunchAgent.print_statusstatuscsHt|jj}||j||jIdH}|ds"ttd|dSdS)zoUpdate the status of the agent. Arguments: status: Status to update the agent to. Nsuccessz!Failed to update agent status to )rrupdate_launch_agent_statusrrrLrMr)rFr_update_status update_retr.r.r/ update_statuss   zLaunchAgent.update_statusrzr|run_idrqi_idcCsdz|j|||}|dkrWdSWdSty1|jd|d|d|d|d YdSw)zfChecks the stateof the run to ensure it has been inited. Note this will not behave well with resuming.pendingTRun /z with rqi id: z did not have associated runF)r get_run_statelowerrrrV)rFrzr|rr run_stater.r.r/_check_run_exists_and_initeds   z(LaunchAgent._check_run_exists_and_inited thread_id exception$Exception | LaunchDockerError | Nonec s(|j |j|}Wdn1swY|jdur+|j|jkr+|jdn|durUtjt|||j d}|j d |dd}| |jt||j|IdHn|jdus_|jdurz|jd|jtd| |jd d IdHno|jdurd }d}d } ||j|j|j|j}|s|tkrn|s|d kr|jIdH}t|IdH|d9}q|sd}|jdkrd} nd} |r|j |dd}| |j| d|IdHn|jd|dtd|j |j|=Wdn1swYt|jdkr| t!IdHdSdS)z&Removes the job from our list for now.NzWSkipping check for completed run status because run is on a different entity than agent)valuetb error.logrOzZcalled finish_thread_id on thread whose tracker has no project or run id. RunQueueItemID: z;Missing project or run id on thread called finish thread idz=submitted job was finished without assigned project or run idagentFrTrfinishedzCThe submitted job exited successfully but failed to call wandb.initz.The submitted run was not successfully startedrunzFinish thread id z had no exception and no runzGlaunch agent called finish thread id on thread without run or exceptionr)"rrrzrrrV tracebackformat_exceptiontype __traceback__saver save_contentsrrrr( err_stager|rrLrMrrRUN_INFO_GRACE_PERIODget_logsasynciosleepcompleted_statusrrrrr AGENT_POLLING) rFrrjob_and_run_statustb_strfnames called_initlogsinterval_msgr.r.r/finish_thread_ids              zLaunchAgent.finish_thread_idr' file_saverrc st|}d|vrd|dvrd|dd<tdt|d}t|t|| t IdH|j d|d}| |t |d||}t||||j|j|dS) zYSet up project and run the job. Arguments: job: Job to run. runSpec_wandb_api_keyz zLaunch agent received job:  NzParsing launch specrunQueueItemId)rdeepcopyrpprintpformatrLrUrNrVr AGENT_RUNNINGr_assert_securerr create_task task_run_jobrr)rFr'r)rjob_copyr launch_spec job_trackerr.r.r/run_jobs*      zLaunchAgent.run_jobr(cCs|jsdS|didi}gd}|dididi}|D]}||vr1td|dq#|d g}|D] }d |vrDtd q:|d id rRtddS)zCIf secure mode is set, make sure no vulnerable keys are overridden.N resource_args kubernetes)hostPIDhostIPC hostNetworkinitContainersspectemplatez"This agent is configured to lock "z@" in pod spec but the job specification attempts to override it. containerscommandzoThis agent is configured to lock "command" in container spec but the job specification attempts to override it.rrzmThis agent is configured to lock the "entrypoint" override but the job specification attempts to override it.)rr=r>)rFr( k8s_configpod_secure_keyspod_specr7container_specscontainer_specr.r.r/r$&s.  zLaunchAgent._assert_securec sL||jdkr t}nt}zz d}|jd7_|j|j|j}|dr+t |j |j kr| IdH}|dur|j }|j}z1t|j|d}||dirh|j|jkrhttd|jd Wq||||IdHWn>ty}z2ttd tt||jtd d d }|j |dt!|d|dIdHWYd}~nd}~ww|jddkrt"|j#dkr|$t%IdHn|$t&IdHt''|j(|kr||j |j ks|durt)*t+IdHnt)*t,IdHqt y|$t-IdHt.td|YnwW|j/0dS|j/0w)zLoop infinitely to poll for jobs and run them. Raises: KeyboardInterrupt: if the agent is requested to stop. rTNr stopPollingrrz>Agent already running the maximum number of sweep schedulers: z@. To set this value use `max_schedulers` key in the agent configError running job: rrO)contentsfname file_sub_typer)rrIrrrzShutting down, active jobs:)1rrDEFAULT_PRINT_INTERVALVERBOSE_PRINT_INTERVALrrrrrKeyboardInterruptrrget_job_and_queuer'r)rrrr=rrrLrRrr*rrMr format_excrrr rr(rrrrr#rrrrAGENT_POLLING_INTERVALRECEIVED_JOB_POLLING_INTERVAL AGENT_KILLEDrUrclear) rFprint_intervalr'r job_and_queuer)rrBrr.r.r/loopCs     B zLaunchAgent.looprr)rc s|d}|s Jd}zz'|j ||j|<Wdn1s wY|||||||IdHWnntyY}zttd|jd|}t |WYd}~nYd}~wt y|}zttd||}t |WYd}~nBd}~wt y}zttdt |}t |WYd}~n)d}~wwW|||IdHdSW|||IdHdSW|||IdHdSW|||IdHdS|||IdHw)NrrzJ encountered an issue while starting Docker, see above output for details.r;)rr _task_run_jobrrLrMrrtrrrrrrCr) rFr(r'rryr)rrrBr.r.r/r&sJ   *zLaunchAgent.task_run_jobcs^t||}||||jt|j}||d|jIdH|||IdH|||j d| |j d| dpCd} t di} |j d| d} t|| \} } }|jp`|j}|}t| d i}t||}t| ||}t| || ||}|js|jst|ts|dusJ||||IdH}|j d t|tr|||IdH}n |sJ|||IdH}||r|jd |j|_Wdn1swYt t!d |j"d |j#d|s |j d |_$WddS1swYdS|j ||_Wdn 1swYt%%}d}|j&'r|(IdHj)}|dkrYt*dkrYt%%|t*krY|+IdHt,dt*d|-||IdHredS|.|j/IdHr|duryt%%}nt%%||j0kr|+IdHt12t3IdH|j&'s/t|t4r|j5dur|j56dSdSdS)Nrz"Fetching and validating project...zFetching resource...rzlocal-containerFzLoading backendr\r^zBackend loaded...Tz"Preparing to run sweep scheduler (r)startingrzRun failed to start within za seconds. If you want to increase this timeout, set WANDB_LAUNCH_START_TIMEOUT to a larger value.)7r from_spec_set_queue_and_rqi_in_projectr)rack_run_queue_itemrcheck_sweep_stateupdate_run_inforrVfetch_and_validate_projectr=rrj docker_imagejob_base_imageget_job_entry_pointrenvironment_from_configregistry_from_configbuilder_from_configrunner_from_configrur build_imagerrrrrrLrUrrrfailed_to_startrris_set get_statusstateRUN_START_TIMEOUTcancelr_check_run_finishedcheck_wandb_run_stoppedrrrrrDr _command_prockill)rFr(r'rryrr)r|ackrbackend_configoverride_build_config_rZrf image_uri entrypointr^r]r\backendr start_time stopped_timer_r.r.r/rKs                       zLaunchAgent._task_run_jobc s|drPzt|j}||d|d|ddIdH}Wnty:}z|jd|d}WYd}~nd}~ww|dkrR|dkrTtd |dd |d dSdSdS) z@Check the state of a sweep before launching a run for the sweep.sweep_idrzr|)sweeprzr|NzFetch sweep state error: rPAUSEDz-Launch agent picked up sweep job, but sweep (z) was in a terminal state (rL)r=rget_sweep_staterrrWr)rFr(ryrrr_rBr.r.r/rQs(   zLaunchAgent.check_sweep_statec sZ|jrdS|jdurt|jSd}z |j}|IdH}|j}|jD]+}||jvrO|j||j |j |dg}|sOt d|d|j |j|q$|dkr|j|jkr|} |j| d<| dd d | d<|j ||_Wdn1s~wY| dtkrttd |jd td WdSttd |jdd| vr| d=t| |j|jdWdS|dvr|jrttd|j|dkrz|j j|j|j|jddWnt y} zt!d| d} ~ wwn ttd|j|j ||_WdWdS1swYWdSWdSt!yb} z7t"td|jdt#| d}|j d|_Wdn1sKwYWYd} ~ |SWYd} ~ |Sd} ~ wt y} z>t"td|jt"t$%t &dt &dt &d|jt &t$%t &dt'(| WYd} ~ |Sd} ~ ww)NTF KuberneteszError adding warning z to run queue item preemptedr _resume_countrrrz has already resumed z times.z was preempted, requeuing...ro)r6 project_queue queue_name)stoppedfailedrrtzScheduler finished with ID: ryCANCELED)rprzr|r_zFailed to update sweep state: zJob finished with ID: zTerminating job z because it failed to start: zError getting status for job z---z&Caught exception while getting status.zJob ID: ))rrrwr\r^r_messagesrappendrupdate_run_queue_item_warningrrNrSremoverzrrrr=rMAX_RESUME_COUNTrLrUrr rr)rrset_sweep_stater|rrrMr(rrCrVrr) rFr)r( known_errorrrr_rSrr6rBr.r.r/rb2s            (     zLaunchAgent._check_run_finishedJobSpecAndQueue | NonecsP|jD]!}||IdH}|dur%|j||j|t||SqdSrK)rrr~r|r%)rFr)r'r.r.r/rBs   zLaunchAgent.get_job_and_queuercCs||_|j|_|d|_dS)Nr)rwr queue_entityr)rFr|r'r)r.r.r/rOsz)LaunchAgent._set_queue_and_rqi_in_project)rlrrmrr1rk)r1r()r1rw)ryrr6r&)rr&r1rwrK) rr(rIr(rr(rrr1r)r1r)r1r)r1r9)r)r(r1r)rr(r1r) rzr(r|r(rr(rr(r1rw)rr9rrr1r)r'r&r)r(rrr1r)r(r&r1r) r(r&r'r&rr&ryrr)rr1r)r(r&r'r&rr&ryrrr9r)rr1r)r(r&ryrr1r)r)rr(r&r1rw)r1r)r|rr'r&r)r(r1r) r*r+r,__doc__rnrp classmethodrvrxrHrrrpropertyrrrrrrrrr*r$rJr&rKrQrbrBrO __classcell__r.r.rrr/rksD    I         ` $  V " d  e rk)r0r(r1r()r)r6r&r7r(r8r9r1r:)NN)rXrYrZrYr1r[)Or __future__rrrloggingr`r!rrr dataclassesrmultiprocessingrtypingrrLwandb.analyticsrwandb.apis.internalr wandb.errorsrwandb.sdk.launch._launch_addr 'wandb.sdk.launch.runner.local_containerr %wandb.sdk.launch.runner.local_processr !wandb.sdk.launch.sweeps.schedulerr wandb.sdk.launch.utilsr r wandb.sdk.librrr _project_specrerrorsrrutilsrrrrjob_status_trackerrrun_queue_item_file_saverrrDrErr#rFrrr rr?r@rr= _env_timeoutr?r`r> getLoggerr*rNr%r5rCrDrjrkr.r.r.r/sx