o ciR@sdZddlZddlZddlZddlZddlZddlZddlmZddl m m m Z ddl m m mZddlmZerFddlmZeeZde_GdddZdS) zaThis file implements a threaded stream controller to return logs back from the ray clientserver. N) TYPE_CHECKING)log_once)WorkerFc@seZdZdddefddZdejfddZdd d Zd e j de fd dZ de defddZde defddZde fddZdddZdddZd S)LogstreamClient client_workerrmetadatacCs4||_||_t|_||_|jd|_dS)zInitializes a thread-safe log stream over a Ray Client gRPC channel. Args: client_worker: The Ray Client worker that manages this client metadata: metadata to pass to gRPC requests N) r _metadataqueueQueue request_queue_start_logthread log_threadstartlast_req)selfrrrN/home/ubuntu/.local/lib/python3.10/site-packages/ray/util/client/logsclient.py__init__s     zLogstreamClient.__init__returncCstj|jdddS)NrT)targetargsdaemon) threadingThread _log_mainrrrrr +sz LogstreamClient._start_logthreadNc Cs d}|jjs|rt|_|jr|j|jt|jj }z|j t |jj d|j d}Wnty;tdYqwz|D]}|jdkrO|j|j|jd|j|j|jdq?WdStjy|}z||}|srWYd}~dSWYd}~nd}~ww|jjrdSdS)NF)r?rlevelmsg)r _in_shutdownr r r rputray_client_pb2_grpcRayletLogStreamerStubchannel Logstreamitergetr ValueErrortimesleepr stdstreamrloggrpcRpcError_process_rpc_error)r reconnectingstub log_streamrecorderrrr.s:      zLogstreamClient._log_mainr4cCsV|j|rtdrtdtdtddStd|jjs)t ddS) z Processes RPC errors that occur while reading from data stream. Returns True if the error can be recovered from, False otherwise. lost_reconnect_logszLog channel is reconnecting. Logs produced while the connection was down can be found on the head node of the cluster in `ray_client_server_[port].out`zLog channel dropped, retrying.rTzShutting down log channel.zUnexpected exception:F) r_can_reconnectrloggerwarningdebugr)r*r exception)rr4rrrr/Ms     z"LogstreamClient._process_rpc_errorrrcCstj||ddS)zLog the message from the log stream. By default, calls logger.log but this can be overridden. Args: level: The loglevel of the received log message msg: The content of the message rN)r7r,)rrrrrrr,bszLogstreamClient.logcCs&|dkrtjntj}t||dddS)zLog the stdout/stderr entry from the log stream. By default, calls print but this can be overridden. Args: level: The loglevel of the received log message msg: The content of the message )fileendN)sysstderrstdoutprint)rrr print_filerrrr+lszLogstreamClient.stdstreamcCs4t|t}d|_||_|j|||_dS)NT) r7setLevelray_client_pb2LogSettingsRequestenabledloglevelr r!r)rrreqrrrset_logstream_levelws   z#LogstreamClient.set_logstream_levelcCs(|jd|jdur|jdSdS)N)r r!r joinrrrrcloses  zLogstreamClient.closecCs$t}d|_|j|||_dS)NF)rErFrGr r!r)rrIrrr disable_logss  zLogstreamClient.disable_logs)rN)__name__ __module__ __qualname__listrrrr rr-r.boolr/intstrr,r+rJrLrMrrrrrs   r)__doc__r?loggingr rr)r-typingr!ray.core.generated.ray_client_pb2core generatedrE&ray.core.generated.ray_client_pb2_grpcr"ray.util.debugrray.util.client.workerr getLoggerrNr7 propagaterrrrrs