o i@s:ddlZddlZddlmZddlmZddlmZddlmZddlm Z ddlm Z ddl m Z dd l mZdd l mZdd l mZdd l mZdd l mZddl mZddlmZddlmZddlmZddlmZddlmZddlmZddlmZee Z!d ddZ"ddZ#d ddZ$d ddZ%ddZ&dS)!N)signals)config) _DD_PIN_NAME)Pin)_SPAN_MEASURED_KEY) SPAN_KIND) trace_utils)trace_after_publish)trace_before_publish) trace_failure) trace_postrun) trace_prerun) trace_retry)SpanKind) SpanTypes)core)ddtrace_after_in_parent)ddtrace_before_fork) get_logger)tracerc Cst|ddrdSd|_|pttjdtjd}||tddttjd d d tdd ttjd |tj j tddt tjdtddt t jjtddt jjtddt jjtddt jjtddt jjtddt jjtdd|S)z\Attach the Pin class to the application and connect our handlers to Celery signals. __datadog_patchFNTworker_service_name)service_configz celery.beatzScheduler.apply_entry apply_entrycSs |djS)Nr)name)argsrW/home/ubuntu/.local/lib/python3.10/site-packages/ddtrace/contrib/internal/celery/app.py0s zpatch_app..zScheduler.ticktickzcelery.app.taskzTask.apply_async apply_asynczcelery.platformsclose_open_fds)weak)getattrrrrceleryontorwrap_traced_beat_functionbeat Scheduler_traced_apply_async_function_patched_close_open_fdsr task_prerunconnectr task_postrunr before_task_publishr after_task_publishr task_failurer task_retryrapppinrrr patch_apps0  r7cCst|ddsdSd|_t|}|durt|tttj j dttj j dttj j j dttjdtjttjttjttjttjttjtdS)zoRemove the Pin instance from the application and disconnect our handlers from Celery signal framework. rFNrr r!r")r$rrget_fromdelattrrrunwrapr%r)r*r5taskTask platformsrr- disconnectr r/r r0r r1r r2r r3rr4rrr unpatch_appHs        r?csfdd}|S)Ncst|}|r |s||i|Stjdtjt |d#}r*||_ | t t j|td||i|WdS1sHwYdS)Nzcelery.beat.{}) span_typer)rr8enabledrtraceformatrWORKERr ext_serviceresource _set_tag_strrrPRODUCER set_metricr)funcinstancerkwargsr6spanfn_nameintegration_config resource_fnrr_traced_beat_innerbs      $z1_traced_beat_function.._traced_beat_innerr)rQrPrRrSrrOrr(asr(cCs dd}|S)ao When apply_async is called, it calls various Celery signals in order, which gets used to start and close the span. Example: before_task_publish starts the span while after_task_publish closes the span. If an exception occurs anywhere inside Celery or its dependencies, this can interrupt the closing signals. The purpose of _traced_apply_async_function is to close the spans even if one of the closing signals don't get called over the course of the apply_task lifecycle. This is done by fetching the stored span and closing it if it hasn't already been closed by a closing signal. c SstdMz8z!||i|WWtd}|r!td||WdSty>td}|r=|jt wtd}|rQtd||ww1sUwYdS)N task_context task_spanzJThe after_task_publish signal was not called, so manually closing span: %r) rcontext_with_data find_itemlogdebugfinish Exception set_exc_infosysexc_info)rKrLrrMrUrrr_traced_apply_async_inners4      z?_traced_apply_async_function.._traced_apply_async_innerr)rQrPrRr_rrrr+ws r+c CsHtdtz||i|}Wttd|Sttdw)z Celery closes all open file descriptors to isolate some fork child. This causes the native runtime to panic because it expects to have a valid fd. We call fork hook to avoid panics when the native runtime interacts with closed fds. z/Shutting down native runtime before closing fdsz+Restarting native runtime after closing fds)rXrYrr)rKrLrrMresultrrrr,s   r,)N)'r]r%rddtracerddtrace._trace.pinrrddtrace.constantsrrddtrace.contribr'ddtrace.contrib.internal.celery.signalsr r r r r r ddtrace.extrrddtrace.internalrddtrace.internal.forksaferrddtrace.internal.loggerr ddtrace.tracer__name__rXr7r?r(r+r,rrrrs8                     *   %