o ciGJ@shddlZddlZddlZddlZddlmZddlmZddlmZddl m Z ddl m Z m Z mZmZmZmZmZmZmZmZddlZddlZddlmZmZmZddlmZeeZ Gd d d Z!d a"da#d e$fd dZ%ddZ&deefddZ'de defddZ(Gddde)Z*dee e+fd e fddZ,GdddZ-edd d ed!fd"d#Z.d$e+fd%d&Z/d'e d(e fd e+fd)d*Z0d'e d(e fd e+fd+d,Z1d-ee+e d(e ffd.ee+e d(e fffd/d0Z2d-ee+e d(e ffd.ee+e d(e ffd e+fd1d2Z3d-ee+e d(e ffd.ee+e d(e ffd e+fd3d4Z4d5d6Z5d7d8Z6d9d:Z7d;d<Z8d=d>Z9dS)?N)contextmanager)wraps) Parameter) ModuleType) AnyCallableDict GeneratorListMutableMappingOptionalSequenceUnioncast)is_class_methodis_function_or_methodis_static_method)get_runtime_contextc@sPeZdZdZhdZddZddZddZd d Zd d Z d dZ ddZ dS)_OpenTelemetryProxya This proxy makes it possible for tracing to be disabled when opentelemetry is not installed on the cluster, but is installed locally. The check for `opentelemetry`'s existence must happen where the functions are executed because `opentelemetry` may be present where the functions are pickled. This can happen when `ray[full]` is installed locally by `ray` (no extra dependencies) is installed on the cluster. >traceContextcontext propagatecCs*|tjvrt|d|Std|)N_zAttribute does not exist: )rallowed_functionsgetattrAttributeError)selfnamerS/home/ubuntu/.local/lib/python3.10/site-packages/ray/util/tracing/tracing_helper.py __getattr__/s z_OpenTelemetryProxy.__getattr__cC |dS)Nzopentelemetry.trace _try_importrrrr _trace5 z_OpenTelemetryProxy._tracecCr")Nzopentelemetry.contextr#r%rrr _context8r'z_OpenTelemetryProxy._contextcCr")Nzopentelemetry.propagater#r%rrr _propagate;r'z_OpenTelemetryProxy._propagatecCs|}|r |jjSdSN)r(rr)rrrrr _Context>sz_OpenTelemetryProxy._ContextcCs$||||dSr*)r&r(r)r+r%rrr try_allEs z_OpenTelemetryProxy.try_allcCs>zt|WStytdddvrtdYdSw)NRAY_TRACING_ENABLEDFalse)true1zInstall opentelemetry with 'pip install opentelemetry-api==1.0.0rc1' and 'pip install opentelemetry-sdk==1.0.0rc1' to enable tracing. See more at docs.ray.io/tracing.html) importlib import_module ImportErrorosgetenvlower)rmodulerrr r$Ks  z_OpenTelemetryProxy._try_importN) __name__ __module__ __qualname____doc__rr!r&r(r)r+r,r$rrrr r"s  rFreturncCstS)zgChecks environment variable feature flag to see if tracing is turned on. Tracing is off by default.)_global_is_tracing_enabledrrrr _is_tracing_enabled\sr>cCsdatatdS)NT)r=r_opentelemetryr,rrrr _enable_tracingbs r@ params_listcCs8t|D]\}}|jtjkr||||Sq|S)z]Given a list of Parameters, if a kwargs Parameter exists, move it to the end of the list.) enumeratekindr VAR_KEYWORDappendpop)rAiparamrrr _sort_params_listis rIfunction new_paramcsPt|}t|j}tfdd|Dr|St|g}|j|d}|S)z/Add additional Parameter to function signature.c3s|] }|jjkVqdSr*)r).0rHrKrr xsz*_add_param_to_signature..) parameters)inspect signaturelistrOvaluesanyrIreplace)rJrKold_sigold_sig_list_repr new_paramsnew_sigrrMr _add_param_to_signaturess  rZc@s eZdZdS)_ImportFromStringErrorN)r8r9r:rrrr r[sr[ import_strc Cst|ts|S|d\}}}|r|sd}t|j|dzt|}WntyB}z|j|kr4|dd}t|j|dd}~ww|}z| dD]}t ||}qKW|St ygd}t|j||d w) zRGiven a string that is in format ":", import the attribute.:zEImport string "{import_str}" must be in format":".)r\Nz'Could not import module "{module_str}".) module_str.z;Attribute "{attrs_str}" not found in module "{module_str}".) attrs_strr^) isinstancestr partitionr[formatr1r2r3rsplitrr) r\r^rr`messager7excinstanceattr_strrrr _import_from_strings8     rjc@s<eZdZdeeeffddZdeeefddfddZdS) _DictPropagatorr<cCsi}tj||S)z*Inject trace context into otel propagator.)r?rinject context_dictrrr inject_current_contexts z&_DictPropagator.inject_current_contextrn_opentelemetry.ContextcCsttjtj|S)z,Given a trace context, extract as a Context.)rr?rrextractrmrrr rqsz_DictPropagator.extractN)r8r9r:rrrorqrrrr rksrkparent_contextrp)NNNc csL|dur|}nt}tj|}z dVWtj|dStj|w)z(Uses the Ray trace context for the span.N)r?rrattachdetach)rr new_contexttokenrrr _use_contexts rw function_namecCsvt}d|tt||d}tjjj j tjjj kr(| }|r(||d<t tjjj dd}|r9||d<|S)zXGet the Attributes of the function that will be reported as attributes in the trace.rJ) ray.remote ray.functionray.pid ray.job_id ray.node_idz ray.task_id worker_idN ray.worker_id)rrbr4getpid get_job_id get_node_idray_privateworker global_workermode WORKER_MODE get_task_idrhex)rxruntime_context span_argstask_idr~rrr _function_hydrate_span_argss   rfunc.cC |dS)z>Returns the function span name that has span kind of producer. ray.remoterrrrr _function_span_producer_name rcCr)z>Returns the function span name that has span kind of consumer. ray.remote_workerrrrrr _function_span_consumer_namerrclass_methodcCst|r|j}t|r|j}t}d|||d|tt||d}tj j j j tj j j kr=|}|r=||d<ttj j j dd}|rN||d<|S)zUGet the Attributes of the actor that will be reported as attributes in the trace.actorr_)ryzray.actor_classzray.actor_methodrzr{r|r} ray.actor_idr~Nr)callabler8rrbr4rrrrrrrrr get_actor_idrr)rrrractor_idr~rrr _actor_hydrate_span_argss*    rcC0t|ts|j}t|ts|j}|d|dS)z;Returns the actor span name that has span kind of producer.r_rrarbr8rrrrr _actor_span_producer_name   rcCr)z;Returns the actor span name that has span kind of consumer.r_rrrrrr _actor_span_consumer_namerrc s<t  d dtdtttfdtdtdtf fdd }|S) zfTrace the execution of a remote task. Inject the current span context into kwargs for propagation.Nargskwargs_args_kwargsr<cstr|jr|durd|vsJ|||g|Ri|Sd|vs#Jtjt}|jt|jtjj j t |jdt |d<|||g|Ri|WdS1sXwYdS)N_ray_trace_ctxrC attributes)r>_is_cross_languager?r get_tracerr8start_as_current_spanr_function_nameSpanKindPRODUCERrrkro)rrrrrtracerrrr _invocation_remote_span)s     $z9_tracing_task_invocation.._invocation_remote_spanNN)rrr )rrrrr _tracing_task_invocation%s rc sbtsSttjdtjjdd_tdddtdtt t tfdtdtffdd }|S) zWrap the function argument passed to RemoteFunction's __init__ so that future execution of that function will include tracing. Use the provided trace context from kwargs. rNdefaultrrrr<c s|dur |i|Stjt}jdj}tt|6|jt |tjj j t |d|i|WdWdS1sIwYWddS1sYwYdS)Nr_r) r?rrr8r9rwrkrqrrrCONSUMERr)rrrrrxrJrr _function_with_tracingUs  Rz=_inject_tracing_into_function.._function_with_tracing) r>rZrPr KEYWORD_ONLY __signature__rrr rrb)rJrrrr _inject_tracing_into_functionFs( rc s:ttdfdtdtttfdtdtffdd }|S)z`Trace the creation of an actor. Inject the current span context into kwargs for propagation.Nrrrrc s|duri}tsd|vsJ|||g|Ri|S|jj}d}d|vs(Jtjt}|jt||tjj j t ||d&}t |d<|||g|Ri|} |d| j| WdS1shwYdS)Nr__init__rrCrr)r>__ray_metadata__ class_namer?rrr8rrrrrrkro set_attribute _ray_actor_idr) rrrrrr method_namerspanresultrrr #_invocation_actor_class_remote_spanrs(    $zD_tracing_actor_creation.._invocation_actor_class_remote_span)rtuplerr )rrrrr _tracing_actor_creationns !rc s@t  d dttdtttfdtdtdtf fdd }|S) z(Trace the invocation of an actor method.Nrrrrr<c str|jjr|durd|vsJ|||g|Ri|S|jjj}|j}d|vs,Jtjt }|j t ||tjj j t||d%}t|d<|d|jj|||g|Ri|WdS1skwYdS)Nrrr)r>_actor_ray_is_cross_language'_ray_actor_creation_function_descriptorr _method_namer?rrr8rrrrrrkrorrr) rrrrrrrrrrrr _start_spans$      $z5_tracing_actor_method_invocation.._start_spanr)rr rr )rrrrr _tracing_actor_method_invocations rcCsdtdtfdtfdd}dtdtfdtfdd}t|t}|D]H\}}t||s/t|r0q"t|s:t|r;q"|dkr@q"t |tj d tj j d d |_ t |r\t|||}nt|||}t|||q"|S) zZGiven a class that will be made into an actor, inject tracing into all of the methods.r.r<c 8dddtdtdttttfdtdtf fdd}|S) Nrrrrrr<c str|dur|g|Ri|Stjt}tt|@|jt |j jtjj j t |j jd|g|Ri|WdWdS1sRwYWddS1sbwYdSzr Wrap the user's function with a function that will extract the trace context Nrr>r?rrr8rwrkrqrr __class__rrrrrrrrrrr _resume_spans    RzF_inject_tracing_into_class..span_wrapper.._resume_spanrr rrbrrrrr span_wrapperz0_inject_tracing_into_class..span_wrapperc r) Nrrrrrr<c str|dur|g|Ri|IdHStjt}tt|E|jt |j jjtjj j t |j jjd|g|Ri|IdHWdWdS1s[wYWddS1skwYdSrrrrrr rs"    RzL_inject_tracing_into_class..async_span_wrapper.._resume_spanrrrrr async_span_wrapperrz6_inject_tracing_into_class..async_span_wrapper__del__rNr)rrrP getmembersrrrisgeneratorfunctionisasyncgenfunctionrZrrriscoroutinefunctionrsetattr)_clsrrmethodsrrwrapped_methodrrr _inject_tracing_into_classs*    r):r1rPloggingr4 contextlibr functoolsrrtypesrtypingrrrr r r r r rrrray._private.workerray._private.inspect_utilrrrray.runtime_contextr getLoggerr8loggerrr=r?boolr>r@rIrZ Exceptionr[rbrjrkrwrrrrrrrrrrrrrrr sr    0   6  "  $    !() %