o ;iZ@sbddlZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z m Z ddl ZddlZddlZddlmZmZddlmZddlmZddlmZmZddlmZmZdd lmZd d l m!Z!d d l"m#Z#d dl$m%Z%m&Z&de'de(fddZ)de'de e*fddZ+dd ddde*de*de e,fddZ-de'dej.fdd Z/d!d"Z0Gd#d$d$Z1dS)%N)suppress)AsyncGeneratorOptional) GRPCErrorStatus)StreamTerminatedError)logger) ConflictErrorExecTimeoutError)api_pb2task_command_router_pb2)TaskCommandRouterStub)grpc_error_converter)aclosing)RETRYABLE_GRPC_STATUS_CODESconnect_channeldatareturncCs dt| d}t||S)z9Decode a base64url string with missing padding tolerated.=)lenbase64urlsafe_b64decode)rpaddingr[/home/ubuntu/.local/lib/python3.10/site-packages/modal/_utils/task_command_router_client.py_b64url_decodesr jwt_tokencCszz-|d}t|dkrWdSt|d}t|}|d}t|ttfr+t|WSWdSt y<t dYdSw)zParse exp from a JWT without verification. Returns UNIX time seconds or None. This is best-effort; if parsing fails or claim missing, returns None. .NrexpzFailed to parse JWT expiration) splitrrjsonloadsget isinstanceintfloat Exceptionrwarning)rparts payload_bpayloadr"rrr_parse_jwt_expiration"s       r/{Gz? base_delay_secs delay_factor max_retriesr3r4r5c sr|ddtffdd } z|IdHWStyA}z|dus(|kr5|jtvr5||IdHn|WYd}~n{d}~wtyj}z|dusP|kr^dt|vr^||IdHn|WYd}~nRd}~wty}z|dusy|kr||IdHn|WYd}~n/d}~wttj fy}z|dus|kr||IdHnt t|WYd}~nd}~wwq)zCall func() with transient error retries and exponential backoff. Authentication retries are expected to be handled by the caller. recs<tdd|tIdH9d7dS)NzRetrying RPC with delay s due to error: r)rdebugasynciosleepr6r4 delay_secs num_retriesrr0sleep_and_update_delay_and_num_retries_remainingEs  z_call_with_retries_on_transient_errors..sleep_and_update_delay_and_num_retries_remainingTN_write_appdata) r*rstatusrAttributeErrorstrrOSErrorr9 TimeoutErrorConnectionError)funcr3r4r5r?r6rr<r%call_with_retries_on_transient_errors7sB      rHtask_idcs|jtj|dIdHS)z:Fetch direct command router access info from Modal server.)rIN)stubTaskGetCommandRouterAccessr !TaskGetCommandRouterAccessRequest) server_clientrIrrrfetch_command_router_accesshs  rNcCsF|s!tt||jWddS1swYdSdSN) is_closedrr*call_soon_threadsafeclose)loopchannelrrr_finalize_channelos  "rUc@seZdZdZedededfddZdddd ded ed ed ej j d e j de j dedededdfddZddZd6ddZdejdejfddZ d7dededddeedeejdff d d!Zdeded"ed#ed$edejf d%d&Z d7dededeedejfd'd(Z d7dededeedej fd)d*Z!d6d+d,Z"d-d.Z# d7dededd/deedeejdff d0d1Z$dej%fd2d3Z&dej'dej(fd4d5Z)dS)8TaskCommandRouterClientz Client used to talk directly to TaskCommandRouter service on worker hosts. A new instance should be created per task. rIrc sz t||IdH}Wntytd|YdSwtd|tj|j}|jdkr:t d|j|j d\}}}|rIt |nd}t } |jr^tdd | _t j| _tjj||| tjjd d d d } t| IdHt} t} ||||j|j| | | S) zAttempt to initialize a TaskCommandRouterClient by fetching direct access. Returns None if command router access is not enabled (FAILED_PRECONDITION). Nz.Command router access is not enabled for task z%Using command router access for task httpsz$Task router URL must be https, got: :izTUsing insecure TLS for task command router because server client points to localhostFi)http2_connection_window_sizehttp2_stream_window_size)sslconfig)rNr rr8urllibparseurlparseurlscheme ValueErrornetloc partitionr(r[create_default_context _is_localhostr+check_hostname CERT_NONE verify_modegrpclibclientChannelr\ Configurationrr9get_running_loopLockjwt) clsrMrIrespohost_port_strport ssl_contextrTrSjwt_refresh_lockrrrtry_inits>     z TaskCommandRouterClient.try_initr0rr1)stream_stdio_retry_delay_secsstream_stdio_retry_delay_factorstream_stdio_max_retries server_urlrprTrSryr{r|r}Nc Csn||_||_||_||_||_||_||_| |_| |_t ||_ ||_ d|_ t |t|||_t|j|_dS)zUCallers should not use this directly. Use TaskCommandRouterClient.try_init() instead.FN)_loop_server_client_task_id _server_url_jwt_channelr{r|r}r/_jwt_exp_jwt_refresh_lock_closedweakreffinalizerU_channel_finalizerr _stub) selfrMrIr~rprTrSryr{r|r}rrr__init__s& z TaskCommandRouterClient.__init__cCsdd|jiS)N authorizationzBearer )rrrrr _get_metadatasz%TaskCommandRouterClient._get_metadatacs6|jrdSd|_|j|jjr|jdSdS)zClose the client.NT)rrrRralivedetachrrrrrRs zTaskCommandRouterClient.closerequestcsDttfddIdHWdS1swYdS)z?Start an exec'd command, properly retrying on transient errors.cjjSrO)_call_with_auth_retryr TaskExecStartrrrrrz4TaskCommandRouterClient.exec_start..NrrHrrrrr exec_starts  $z"TaskCommandRouterClient.exec_startexec_idfile_descriptorz api_pb2.FileDescriptor.ValueTypedeadlinec Cs|tjkr tj}n!|tjkrtj}n|tjks|tjkr$td|td|t Bt | ||||4IdH}|2z 3dHW}|Vq@6WdIdHn1IdHs[wYWddSWddS1sswYdS)a.Stream stdout/stderr batches from the task, properly retrying on transient errors. Args: task_id: The task ID of the task running the exec'd command. exec_id: The execution ID of the command to read from. file_descriptor: The file descriptor to read from. deadline: The deadline by which all output must be streamed. If None, wait forever. If the deadline is exceeded, raises an ExecTimeoutError. Returns: AsyncGenerator[sr_pb2.TaskExecStdioReadResponse, None]: A stream of stdout/stderr batches. Raises: ExecTimeoutError: If the deadline is exceeded. Other errors: If retries are exhausted on transient errors or if there's an error from the RPC itself. zUnsupported file descriptor: zInvalid file descriptor: N) r FILE_DESCRIPTOR_STDOUTsr_pb2&TASK_EXEC_STDIO_FILE_DESCRIPTOR_STDOUTFILE_DESCRIPTOR_STDERR&TASK_EXEC_STDIO_FILE_DESCRIPTOR_STDERRFILE_DESCRIPTOR_INFOFILE_DESCRIPTOR_UNSPECIFIEDrbrr _stream_stdio)rrIrrrsr_fdstreamitemrrrexec_stdio_reads"  *"z'TaskCommandRouterClient.exec_stdio_readoffsetreofcsXtj|||||dttfddIdHWdS1s%wYdS)aBWrite to the stdin stream of an exec'd command, properly retrying on transient errors. Args: task_id: The task ID of the task running the exec'd command. exec_id: The execution ID of the command to write to. offset: The offset to start writing to. data: The data to write to the stdin stream. eof: Whether to close the stdin stream after writing the data. Raises: Other errors: If retries are exhausted on transient errors or if there's an error from the RPC itself. )rIrrrrcrrO)rrTaskExecStdinWriterrrrr+rz:TaskCommandRouterClient.exec_stdin_write..N)rTaskExecStdinWriteRequestrrH)rrIrrrrrrrexec_stdin_writes $z(TaskCommandRouterClient.exec_stdin_writec stj||d|dur|tnd}|dur#|dkr#td|t*ztjtfdd|dIdHWWdStj yOtd|w1sSwYdS)a2Poll for the exit status of an exec'd command, properly retrying on transient errors. Args: task_id: The task ID of the task running the exec'd command. exec_id: The execution ID of the command to poll on. Returns: sr_pb2.TaskExecPollResponse: The exit status of the command if it has completed. Raises: ExecTimeoutError: If the deadline is exceeded. Other errors: If retries are exhausted on transient errors or if there's an error from the RPC itself. rIrNrz)Deadline exceeded while polling for exec crrO)rr TaskExecPollrrrrrIrz3TaskCommandRouterClient.exec_poll..timeout) rTaskExecPollRequesttime monotonicr rr9wait_forrHrErrIrrrrrr exec_poll.s$  z!TaskCommandRouterClient.exec_pollc stj||d|dur|tnd}|dur#|dkr#td|t.ztjtfdddddd|d IdHWWdStj yStd|w1sWwYdS) aWait for an exec'd command to exit and return the exit code, properly retrying on transient errors. Args: task_id: The task ID of the task running the exec'd command. exec_id: The execution ID of the command to wait on. Returns: Optional[sr_pb2.TaskExecWaitResponse]: The exit code of the command. Raises: ExecTimeoutError: If the deadline is exceeded. Other errors: If there's an error from the RPC itself. rNrz)Deadline exceeded while waiting for exec csjjjddS)N<r)rr TaskExecWaitrrrrrtsz3TaskCommandRouterClient.exec_wait..rr2r) rTaskExecWaitRequestrrr rr9rrHrErrrr exec_waitPs*  z!TaskCommandRouterClient.exec_waitc s|j4IdHo|jr WdIdHdS|jdur<|jtdkr.sleep_and_update_delay_and_num_retries_remainingTN)rr)rIrrr)endr@)r{r|r}r*maxrrrTaskExecStdioReadopenrrTaskExecStdioReadRequest send_messagerrrrArrrrrBrCrr9rErFrD)rrIrrrrdid_auth_retryr?rrsreqrrr6rrrrs              z%TaskCommandRouterClient._stream_stdiocDttfddIdHWdS1swYdS)NcrrO)rrTaskMountDirectoryrrrrrrz5TaskCommandRouterClient.mount_image..rrrrr mount_images  $z#TaskCommandRouterClient.mount_imagecr)NcrrO)rrTaskSnapshotDirectoryrrrrr rz.rrrrrsnapshot_directorys  $z*TaskCommandRouterClient.snapshot_directory)rNrO)*__name__ __module__ __qualname____doc__ classmethodrCrrzrjrkrlr9AbstractEventLoopror)r(rrrRrTaskExecStartRequestTaskExecStartResponserrTaskExecStdioReadResponserbytesboolTaskExecStdinWriteResponserTaskExecPollResponserTaskExecWaitResponserrrrTaskMountDirectoryRequestrTaskSnapshotDirectoryRequestTaskSnapshotDirectoryResponserrrrrrVxs :    .     &  & .  ^rV)2r9rr$r[r urllib.parser]r contextlibrtypingrrgrpclib.clientrjgrpclib.configgrpclib.eventsrrgrpclib.exceptionsr modal.configrmodal.exceptionr r modal_protor r r$modal_proto.task_command_router_grpcr _grpc_clientr async_utilsr grpc_utilsrrrCrrr)r/r(rH"TaskGetCommandRouterAccessResponserNrUrVrrrrsH       1