o i@sddlmZddlmZddlmZddlmZddlm Z ddl m Z ddl m Z eZ egd Zd ed ed efd dZde deeefd dfddZd!ddZd!ddZd!ddZd!ddZd!ddZdd ZdS)")Any)WeakValueDictionary)set_flattened_tags)HTTPPropagator)Span)CTX_KEY)SPAN_KEY)) compressionzcelery.compression)correlation_idzcelery.correlation_id) countdownzcelery.countdown) delivery_infozcelery.delivery_info)etaz celery.eta)exchangezcelery.exchange)expireszcelery.expires)hostnamecelery.hostname)idz celery.id)priorityzcelery.priority)queuez celery.queue)reply_tozcelery.reply_to)retrieszcelery.retries) routing_keyzcelery.routing_key) serializerzcelery.serializer) timelimitzcelery.timelimit)originr)statez celery.statekeyvaluereturncCsJ|dus|dkr dS|dkrtdd|DrdS|dkr#|dkr#dSdS) NTrcss|]}|duVqdS)N).0_r!r!Y/home/ubuntu/.local/lib/python3.10/site-packages/ddtrace/contrib/internal/celery/utils.py 0sz,should_skip_context_value..rrF)all)rrr!r!r$should_skip_context_value)sr'spancontextNcCsDg}tD]\}}||}t||rq|||fqt||dS)z3Helper to extract meta values from a Celery ContextN)TAG_KEYSgetr'appendr)r(r) context_tagsrtag_namerr!r!r$set_tags_from_context:s   r/FcCsJi}t|j|t|td}|durt}t|t|||||df<dS)Ndistributed_context) propagatorinjectr)getattrrdictsetattr)tasktask_idr( is_publish trace_headers context_dictr!r!r$attach_span_contextHs  r;cCs(t|td}|dur dS|||dfS)HHelper to retrieve an active `Span` stored in a `Task` instance Nr0)r3rr+)r6r7r8r:r!r!r$retrieve_span_contextUs r=cCs6t|td}|durt}t|t|||||f<dS)aHelper to propagate a `Span` for the given `Task` instance. This function uses a `WeakValueDictionary` that stores a Datadog Span using the `(task_id, is_publish)` as a key. This is useful when information must be propagated from one Celery signal to another. DEV: We use (task_id, is_publish) for the key to ensure that publishing a task from within another task does not cause any conflicts. This mostly happens when either a task fails and a retry policy is in place, or when a task is manually retried (e.g. `task.retry()`), we end up trying to publish a task with the same id as the task currently running. Previously publishing the new task would overwrite the existing `celery.run` span in the `weak_dict` causing that span to be forgotten and never finished. NOTE: We cannot test for this well yet, because we do not run a celery worker, and cannot run `task.apply_async()` N)r3r rr5)r6r7r(r8 weak_dictr!r!r$ attach_spanas  r?cCs>t|td}|dur dSz|||f=WdStyYdSw)zHelper to remove a `Span` in a Celery task when it's propagated. This function handles tasks where the `Span` is not attached. N)r3r KeyErrorr6r7r8r>r!r!r$ detach_span|s  rBcCs&t|td}|dur dS|||fS)r<N)r3r r+rAr!r!r$ retrieve_spans rCcCsH|d}|d}|rd|vr|dS|r d|vr"|dSdSdS)zHelper to retrieve the `Task` identifier from the message `body`. This helper supports Protocol Version 1 and 2. The Protocol is well detailed in the official documentation: http://docs.celeryproject.org/en/latest/internals/protocol.html headersbodyrN)r+)r)rDrEr!r!r$retrieve_task_ids      rF)F)typingrweakrefr$ddtrace.contrib.internal.trace_utilsrddtrace.propagation.httpr ddtrace.tracer constantsrr r1 frozensetr*strboolr'r4r/r;r=r?rBrCrFr!r!r!r$s&