o -wi @sdZddlZddlZddlZddlZddlZddlZddlmZm Z m Z m Z m Z m Z mZddlZddlZddlmZddlmZddlmZddlmZddlmZdd lmZdd lmZdd lm Z m!Z!m"Z"m#Z#m$Z$dd l%m&Z&m'Z'dd l(m)Z)ddl*m+Z+m,Z,ddl-m.Z.ddl/m0Z0m1Z1m2Z2m3Z3m4Z4m5Z5ddl6m7Z7m8Z8e)dddddl9Z9ddl9m:Z:ddl;mZ>ddl?m@Z@ddlAmBZBddlCmDZDdZEdZFeGeHZIejJKdZLejJKdZMGd d!d!e7ZNGd"d#d#e7ZOGd$d%d%e8ZPd&e eQd'e e+d(e eRd)eSd*df d+d,ZTd-d.d/eRd0eRd1eRd*d2f d3d4ZUd-d.d5ed6eRd0eRd*e d2f d7d8ZVd9ed*e eQfd:d;ZWd9eeQeXfdZYd?ed*e eQfd@dAZZd?eeQeXfdBeRdCeRd*dfdDdEZ[d?eeQeXfdFeQd*dfdGdHZ\d?ee eXfdIe,d*dfdJdKZ]dS)Lz:Implementation of KubernetesRunner class for wandb launch.N)AnyDictIteratorListOptionalTupleUnion)Api) LaunchAgent)AbstractEnvironment)AbstractRegistry)AzureContainerRegistry) LocalRegistry)Status)WANDB_K8S_LABEL_AGENTWANDB_K8S_LABEL_MONITORWANDB_K8S_RUN_IDCustomResourceLaunchKubernetesMonitor)ExponentialBackoff retry_async) get_module) EntryPoint LaunchProject) LaunchError)CODE_MOUNT_DIR LOG_PREFIXMAX_ENV_LENGTHSPROJECT_SYNCHRONOUSget_kube_context_and_api_clientmake_name_dns_safe) AbstractRunAbstractRunnerkubernetes_asynciozfKubernetes runner requires the kubernetes package. Please install it with `pip install wandb[launch]`.)required)client) BatchV1Api) CoreV1Api)CustomObjectsApi)V1Secret) ApiException WANDB_LAUNCH_CODE_PVC_MOUNT_PATHWANDB_LAUNCH_CODE_PVC_NAMEc @seZdZdZ  dddddded eed ed d df d dZed efddZd eefddZ d e fddZ d e fddZ dddZdddZdS)KubernetesSubmittedRunz)Wrapper for a launched run on Kubernetes.defaultN batch_apir(core_apir)name namespacesecretr+returncCs(||_||_||_||_d|_||_dS)a,Initialize a KubernetesSubmittedRun. Other implementations of the AbstractRun interface poll on the run when `get_status` is called, but KubernetesSubmittedRun uses Kubernetes watch streams to update the run status. One thread handles events from the job object and another thread handles events from the rank 0 pod. These threads updated the `_status` attributed of the KubernetesSubmittedRun object. When `get_status` is called, the `_status` attribute is returned. Arguments: batch_api: Kubernetes BatchV1Api object. core_api: Kubernetes CoreV1Api object. name: Name of the job. namespace: Kubernetes namespace. secret: Kubernetes secret. Returns: None. rN)r2r3r4r5 _fail_countr6)selfr2r3r4r5r6r:f/home/ubuntu/sommelier/.venv/lib/python3.10/site-packages/wandb/sdk/launch/runner/kubernetes_runner.py__init__Ms  zKubernetesSubmittedRun.__init__cC|jS)zReturn the run id.r4r9r:r:r;idpzKubernetesSubmittedRun.idc szF|jjd|j|jdIdH}dd|jD}|s(td|jWdS|jj|d|jdIdH}|r=t|WStd|WdSt yd}zt t d |WYd}~dSd}~ww) Nz job-name=label_selectorr5cSg|]}|jjqSr:metadatar4.0pir:r:r; zz3KubernetesSubmittedRun.get_logs..z"Found no pods for kubernetes job: rr4r5zNo logs for kubernetes pod(s): zFailed to get pod logs: ) r3list_namespaced_podr4r5itemswandbtermwarnread_namespaced_pod_logstr Exception termerrorr)r9pods pod_nameslogser:r:r;get_logsus*    zKubernetesSubmittedRun.get_logscsf |IdH}ttd|jd|j|jdvrn tdIdHq|IdH|jdkS)zzWait for the run to finish. Returns: True if the run finished successfully, False otherwise. TNJob status: finishedfailed preemptedr-r]) get_statusrOtermlogrr4stateasynciosleep_delete_secretr9statusr:r:r;waits zKubernetesSubmittedRun.waitcs(t|j}|dvr|IdH|S)N)stoppedr^r]r_)rr`r4rerfr:r:r;r`s  z!KubernetesSubmittedRun.get_statusc snz|jj|j|jdIdH|IdHWdSty6}ztd|jd|jdt||d}~ww)zCancel the run.)r5r4Nz Failed to delete Kubernetes Job  in namespace : )r2delete_namespaced_jobr5r4rer,rrRr9rXr:r:r;cancels zKubernetesSubmittedRun.cancelcsHtjds |jr"|jj|jjj|jjjdIdHd|_dSdSdS)NWANDB_RELEASE_NAMErL) osenvirongetr6r3delete_namespaced_secretrFr4r5r?r:r:r;res z%KubernetesSubmittedRun._delete_secret)r1Nr7N)__name__ __module__ __qualname____doc__rRrr<propertyr@rYboolrhrr`rnrer:r:r:r;r0Js0 #  r0c@seZdZdZdededededededed d fd d Zed efd dZ d e efddZ d e fddZ dddZd efddZd S)CrdSubmittedRunz-Run submitted to a CRD backend, e.g. Volcano.groupversionpluralr4r5r3 custom_apir7NcCs4||_||_||_||_||_||_||_d|_dS)aCreate a run object for tracking the progress of a CRD. Arguments: group: The API group of the CRD. version: The API version of the CRD. plural: The plural name of the CRD. name: The name of the CRD instance. namespace: The namespace of the CRD instance. core_api: The Kubernetes core API client. custom_api: The Kubernetes custom object API client. Raises: LaunchError: If the CRD instance does not exist. rN)r|r}r~r4r5r3rr8)r9r|r}r~r4r5r3rr:r:r;r<s zCrdSubmittedRun.__init__cCr=)z"Get the name of the custom object.r>r?r:r:r;r@rAzCrdSubmittedRun.idc si}z-|jjd|j|jdIdH}dd|jD}|D]}|jj||jdIdH||<qWn!tyQ}ztd|jdt |WYd}~dSd}~ww|sVdSd d|D}d |S) zGet logs for custom object.z wandb/run-id=rBNcSrDr:rErGr:r:r;rJrKz,CrdSubmittedRun.get_logs..rLzFailed to get logs for rkcSs g|] \}}d|d|qS)zPod z: r:)rHpod_namelogr:r:r;rJs  ) r3rMr4r5rNrQr,rOrPrRjoin)r9rWrUrVrrX logs_as_arrayr:r:r;rYs*  zCrdSubmittedRun.get_logscst|jS)zGet status of custom object.)rr`r4r?r:r:r;r`s zCrdSubmittedRun.get_statusc slz|jj|j|j|j|j|jdIdHWdSty5}ztd|jd|jdt ||d}~ww)zCancel the custom object.)r|r}r5r~r4NzFailed to delete CRD rjrk) rdelete_namespaced_custom_objectr|r}r5r~r4r,rrRrmr:r:r;rns"zCrdSubmittedRun.cancelcsT |IdH}ttd|jd||jdvr!|jdkStdIdHq)z.Wait for this custom object to finish running.TNrZr[r\r]r-)r`rOrarr4rbrcrdrfr:r:r;rh s  zCrdSubmittedRun.waitrt)rurvrwrxrRr)r*r<ryr@rrYrr`rnrzrhr:r:r:r;r{s2 ! r{cseZdZdZdedeeefdede ddf fdd Z d eeefd eeefdefd d Z d eeefde dededdde eeefedff ddZde dedeefddZZS)KubernetesRunnerzLaunches runs onto kubernetes.apibackend_config environmentregistryr7Ncst||||_||_dS)a+Create a Kubernetes runner. Arguments: api: The API client object. backend_config: The backend configuration. environment: The environment to launch runs into. Raises: LaunchError: If the Kubernetes configuration is invalid. N)superr<rr)r9rrrr __class__r:r;r<s zKubernetesRunner.__init__ resource_argscontextcCsL|r |dddnd}|didp%|dp%|jdidp%|S)zGet the namespace to launch into. Arguments: resource_args: The resource args to launch. context: The k8s config context. Returns: The namespace to launch into. rr5r1rFrunner)rrr)r9rrdefault_namespacer:r:r; get_namespace,s zKubernetesRunner.get_namespacelaunch_project image_urir5r3r)r+c s>ddd}|||di}ddd}||di|d i} d d i} | | di| d ig} |d i|j|d t<d|d t<trXt|d t <|dslt d|j d|j d|d<t | D]%\} } d| vr| ddt| | d<d| vrdddgiddid| d<qp|jp|}|jr|| dd<ntdd| Ds|dusJ|| dd<t||j|j|IdH}|durdd|jig| d <t| ||j|jdu||jt|jj}d}| D]y} | d!pg}|D]g\}}|d"kr]|r]ts|jtr]tj d#}d$}|r(|d|7}n|d|j7}d%d&}t!t"t#j$d'd(t#j$d'd)t%d*t&|||||d+IdH}|'|d,|d-d.id/q|'||d0q|| d!<q| | d <| | d<| |d <||d<||d<t(|td|j)rt*||trt(|t t||fS)1aApply our default values, return job dict and api key secret. Arguments: resource_args (Dict[str, Any]): The resource args to launch. launch_project (LaunchProject): The launch project. builder (Optional[AbstractBuilder]): The builder. namespace (str): The namespace. core_api (CoreV1Api): The core api. Returns: Tuple[Dict[str, Any], Optional["V1Secret"]]: The resource args and api key secret. batch/v1Job) apiVersionkindrFr<) backoffLimitttlSecondsAfterFinishedspectemplate restartPolicyNever containerslabelstruer4zlaunch-- generateNamelaunchsecurityContextFdropALLtypeRuntimeDefault)allowPrivilegeEscalation capabilitiesseccompProfileimagecSsg|]}d|vqS)rr:)rHcontr:r:r;rJrKz5KubernetesRunner._inject_defaults..Nregcred-imagePullSecretsenv WANDB_API_KEYroz wandb-api-keycSstd|ddS)N3Exception when ensuring Kubernetes API key secret: z . Retrying...)rOrP)rXr:r:r;handle_exceptions z;KubernetesRunner._inject_defaults..handle_exceptionr")seconds)minutes) initial_sleep max_sleep max_retries)backofffnon_excr3 secret_namer5api_key secretKeyRefpassword)r4key)r4 valueFromr4value)+updaterr setdefaultrun_idrrr initializedr4rr! target_entitytarget_project enumeraterRoverride_entrypointget_job_entry_point docker_imageanymaybe_create_imagepull_secretrinject_entrypoint_and_args override_argsget_env_vars_dict_apirrrurNrrrprqrrdatetime timedeltaAPI_KEY_SECRET_MAX_RETRIESensure_api_key_secretappendadd_label_to_podsjob_base_imageapply_code_mount_configuration)r9rrrr5r3job job_metadatajob_spec pod_templatepod_specrir entry_pointr6env_varsapi_key_secretrrr release_namerrr:r:r;_inject_defaultsDs                     z!KubernetesRunner._inject_defaultsc# sFtIdH||di}|sttdtd|t t |IdH\}}|j durLt dus:t dur>td|}|tjt ||dd}|dvri||jt|jj}t|||d i|d <|d d i|d d <d |d d t<t|td trt|ttt|d d t<|j rt ||i} |j!r|j!| d <|j"r|j"j#| d <t$|| t%&|} |'||} |(d^} } }|d| } |d| } |d| }|)d}t*| | |d}tj+| |dz| j,| | | ||dIdH}Wn+t-y3}zt./|j0}t12|}td|d|j3d|j4d||d}~ww|d id}td|d|d dt5|| | | |t%6|| d}|j7t8rg|9IdH|St j%:|}t j%6|}|'||} |;|||| |IdH\}}d}d|vr|d|d7}t|zt jy}z-|j?D]#}t./|j0}|d}|d }td!|j@d"|d|j4d#|WYd}~nd}~wtAy}z td$tB|dd}~ww|d%} | jCj}!t+| tD|||!| |}"|j7t8r!|"9IdH|"S)&aExecute a launch project on Kubernetes. Arguments: launch_project: The launch project to execute. builder: The builder to use to build the image. Returns: The run object if the run was successful, otherwise None. N kuberneteszyNote: no resource args specified. Add a Kubernetes yaml spec or other options in a json file with --resource-args .z+Running Kubernetes job with resource args: a0WANDB_LAUNCH_SOURCE_CODE_PVC_ environment variables not set. Unable to mount source code PVC into base image. Use the `codeMountPvcName` variable in the agent helm chart to enable base image jobs for this agent. See https://github.com/wandb/helm-charts/tree/main/charts/launch-agent for more information.rr)rz batch/v1beta1rFrrargscommand/r|r}rs)r|r}r~)custom_resource)r|r}r5r~bodyzError creating CRD of kind rk rr4zCreated )r4r|r}r5r~r3rzCreating Kubernetes job)r5messagecodez(Failed to create Kubernetes job for run z (z): z3Unexpected exception when creating Kubernetes job: r)Erensure_initialized fill_macrosrrrOrar_loggerinfor r%rSOURCE_CODE_PVC_NAMESOURCE_CODE_PVC_MOUNT_PATHrget_image_source_stringchange_project_dirrppathrrrrrru add_wandb_envrrr rrr4rrrradd_entrypoint_args_overridesr'r*rsplitlowerrmonitor_namespacecreate_namespaced_custom_objectr,jsonloadsryamldumprgreasonr{r)rrrhr(rutilscreate_from_dictFailToCreateErrorapi_exceptionsrrSrRrFr0)#r9rrrr api_client code_subdir api_versionr overridesrr5r|r}_rr~rresponserXr body_yamlr4 submitted_runr2r3rr6msgexcrespr job_responsejob_name submitted_jobr:r:r;runs                                zKubernetesRunner.run)rurvrwrxr rrRrr r r<rrrrrr#r __classcell__r:r:rr;rsR      )rrrrshould_override_entrypointr7cCsJtt|D]}|r|||d<|r"||dr|r"|j||d<qdS)aPInject the entrypoint and args into the containers. Arguments: containers: The containers to inject the entrypoint and args into. entry_point: The entrypoint to inject. override_args: The args to inject. should_override_entrypoint: Whether to override the entrypoint. Returns: None rrN)rangelenrrr)rrrrrr:r:r;rs  rr3r)rr5rr+c s2dt|i}ddi}tj|tj|||dddd}zcz |||IdHWWSty}zK|j d kr}|j ||d IdH}|j |krt|j j ddkrj|j||d IdH|||IdHWYd}~WStd |d ||WYd}~WSd}~wwty}z td t|dd}~ww)a<Create a secret containing a user's wandb API key. Arguments: core_api: The Kubernetes CoreV1Api object. secret_name: The name to use for the secret. namespace: The namespace to create the secret in. api_key: The user's wandb API key Returns: The created secret rzwandb.ai/created-byz launch-agent)r4r5rSecretzkubernetes.io/basic-authdatarFrrNrLz.Kubernetes secret already exists in namespace z with incorrect data: rr)base64 b64encodeencodedecoder'r+ V1ObjectMetacreate_namespaced_secretr,rgread_namespaced_secretr rFrrrrsrrSrR) r3rr5r secret_datarr6rXexisting_secretr:r:r;rsT     rrrc s,d}t|ts t|trdS|IdH\}}d|jt|d|ddii}dtt |i}t j |t j d||dd d d }z2z |||IdHWWSty} z| jd krz|jd||dIdHWYd} ~ WSd} ~ wwty} z td t| dd} ~ ww)aICreate a secret for pulling images from a private registry. Arguments: core_api: The Kubernetes CoreV1Api object. registry: The registry to pull from. run_id: The run id. namespace: The namespace to create the secret in. Returns: A secret if one was created, otherwise None. Nauths:zdeprecated@wandblaunch.com)authemailz.dockerconfigjsonrrLrzkubernetes.io/dockerconfigjsonrr!z+Exception when creating Kubernetes secret: r) isinstancerr get_username_passwordurir"r#r$r%rdumpsr'r+r&r'r,rgr(rSrrR) r3rrr5r6unametoken creds_infor)rXr:r:r;rsJ    rrootccst|tr.|D]!\}}|dkrt|tr|EdHq t|ttfr+t|EdHq dSt|tr?|D] }t|EdHq5dSdS)zYield all container specs in a manifest. Recursively traverses the manifest and yields all container specs. Container specs are identified by the presence of a "containers" key in the value. rN)r/dictrNlistyield_containers)r6kvitemr:r:r;r9,s     r9rcCsPt|D]!}|dg}|dd|D||d<d|vr%|dqdS)aInjects wandb environment variables into specs. Recursively walks the spec and injects the environment variables into every container spec. Containers are identified by the "containers" key. This function treats the WANDB_RUN_ID and WANDB_GROUP_ID environment variables specially. If they are present in the spec, they will be overwritten. If a setting for WANDB_RUN_ID is provided in env_vars, then that environment variable will only be set in the first container modified by this function. Arguments: root: The spec to modify. env_vars: The environment variables to inject. Returns: None. rcSsg|] \}}||dqS)rr:)rHrrr:r:r;rJQsz!add_wandb_env.. WANDB_RUN_IDN)r9rextendrNpop)r6rrrr:r:r;r>s   rmanifestccs~t|tr|D] }t|EdHqdSt|tr;d|vr&d|dvr&|V|D]}t|ttfr:t|EdHq*dSdS)zYield all pod specs in a manifest. Recursively traverses the manifest and yields all pod specs. Pod specs are identified by the presence of a "spec" key with a "containers" key in the value. Nrr)r/r8 yield_podsr7values)r@r<rr:r:r;rAXs   rA label_key label_valuecCs2t|D]}|di}|di}|||<qdS)aAdd a label to all pod specs in a manifest. Recursively traverses the manifest and adds the label to all pod specs. Pod specs are identified by the presence of a "spec" key with a "containers" key in the value. Arguments: manifest: The manifest to modify. label_key: The label key to add. label_value: The label value to add. Returns: None. rFrN)rAr)r@rCrDpodrFrr:r:r;rjs    rrcCst|tr|D]}t||qdSt|trKd|vr?d|dvr?|dd}|D]}d|vr4|d|d<d|vr>|d|d<q(|D] }t||qCdSdS)aAdd entrypoint and args overrides to all containers in a manifest. Recursively traverses the manifest and adds the entrypoint and args overrides to all containers. Containers are identified by the presence of a "spec" key with a "containers" key in the value. Arguments: manifest: The manifest to modify. overrides: Dictionary with args and entrypoint keys. Returns: None. rrrrN)r/r8rr7rB)r@rr<r containerrr:r:r;rs"        rprojectcCstdusJ|}t|D]8}t|D]}d|vrg|d<|ddt|dt|d<q|d}d|vr:g|d<|dddtid qdS) aApply code mount configuration to all containers in a manifest. Recursively traverses the manifest and adds the code mount configuration to all containers. Containers are identified by the presence of a "spec" key with a "containers" key in the value. Arguments: manifest: The manifest to modify. project: The launch project. Returns: None. N volumeMountszwandb-source-code-volume)r4 mountPathsubPath workingDirrvolumes claimName)r4persistentVolumeClaim)rrrAr9rr)r@rG source_dirrErFrr:r:r;rs.    r)^rxrcr"rrloggingrptypingrrrrrrrrrOwandb.apis.internalr wandb.sdk.launch.agent.agentr %wandb.sdk.launch.environment.abstractr "wandb.sdk.launch.registry.abstractr 2wandb.sdk.launch.registry.azure_container_registryr (wandb.sdk.launch.registry.local_registryr wandb.sdk.launch.runner.abstractr*wandb.sdk.launch.runner.kubernetes_monitorrrrrrwandb.sdk.lib.retryrr wandb.utilr _project_specrrerrorsrrrrrrr r!abstractr#r$r%r'*kubernetes_asyncio.client.api.batch_v1_apir()kubernetes_asyncio.client.api.core_v1_apir)0kubernetes_asyncio.client.api.custom_objects_apir**kubernetes_asyncio.client.models.v1_secretr+kubernetes_asyncio.client.restr,TIMEOUTr getLoggerrurrqrrrrr0r{rr7rRrzrrrr9r8rrArrrr:r:r:r;s$                   o[   = 8&