[celery] use attach/detach verbs instead of propagate/remove

Author: Emanuele Palazzetti

ddtrace/contrib/celery/signals.py

@@ -8,9 +8,9 @@
 from .util import (
     tags_from_context,
     retrieve_task_id,
-    propagate_span,
+    attach_span,
+    detach_span,
     retrieve_span,
-    remove_span,
 )
 
 
@@ -33,7 +33,7 @@ def trace_prerun(*args, **kwargs):
 
     # propagate the `Span` in the current task Context
     span = pin.tracer.trace(c.WORKER_ROOT_SPAN, service=c.WORKER_SERVICE, resource=task.name)
-    propagate_span(task, task_id, span)
+    attach_span(task, task_id, span)
 
 
 def trace_postrun(*args, **kwargs):
@@ -55,7 +55,7 @@ def trace_postrun(*args, **kwargs):
         span.set_tags(tags_from_context(kwargs))
         span.set_tags(tags_from_context(task.request))
         span.finish()
-        remove_span(task, task_id)
+        detach_span(task, task_id)
 
 
 def trace_before_publish(*args, **kwargs):
@@ -86,7 +86,7 @@ def trace_before_publish(*args, **kwargs):
     # Note: adding tags from `traceback` or `state` calls will make an
     # API call to the backend for the properties so we should rely
     # only on the given `Context`
-    propagate_span(task, task_id, span)
+    attach_span(task, task_id, span)
 
 
 def trace_after_publish(*args, **kwargs):
@@ -105,7 +105,7 @@ def trace_after_publish(*args, **kwargs):
         return
     else:
         span.finish()
-        remove_span(task, task_id)
+        detach_span(task, task_id)
 
 
 def trace_failure(*args, **kwargs):

ddtrace/contrib/celery/util.py

@@ -50,7 +50,7 @@ def tags_from_context(context):
     return tags
 
 
-def propagate_span(task, task_id, span):
+def attach_span(task, task_id, span):
     """Helper to propagate a `Span` for the given `Task` instance. This
     function uses a `WeakValueDictionary` that stores a Datadog Span using
     the `task_id` as a key. This is useful when information must be
@@ -64,7 +64,7 @@ def propagate_span(task, task_id, span):
     weak_dict[task_id] = span
 
 
-def remove_span(task, task_id):
+def detach_span(task, task_id):
     """Helper to remove a `Span` in a Celery task when it's propagated.
     This function handles tasks where the `Span` is not attached.
     """

tests/contrib/celery/test_utils.py

@@ -5,9 +5,9 @@
 from ddtrace.contrib.celery.util import (
     tags_from_context,
     retrieve_task_id,
-    propagate_span,
+    attach_span,
+    detach_span,
     retrieve_span,
-    remove_span,
 )
 
 from .base import CeleryBaseTestCase
@@ -72,7 +72,7 @@ def fn_task():
         # propagate and retrieve a Span
         task_id = '7c6731af-9533-40c3-83a9-25b58f0d837f'
         span_before = self.tracer.trace('celery.run')
-        propagate_span(fn_task, task_id, span_before)
+        attach_span(fn_task, task_id, span_before)
         span_after = retrieve_span(fn_task, task_id)
         ok_(span_before is span_after)
 
@@ -85,10 +85,10 @@ def fn_task():
         # propagate a Span
         task_id = '7c6731af-9533-40c3-83a9-25b58f0d837f'
         span = self.tracer.trace('celery.run')
-        propagate_span(fn_task, task_id, span)
+        attach_span(fn_task, task_id, span)
         # delete the Span
         weak_dict = getattr(fn_task, '__dd_task_span')
-        remove_span(fn_task, task_id)
+        detach_span(fn_task, task_id)
         ok_(weak_dict.get(task_id) is None)
 
     def test_span_delete_empty(self):
@@ -102,7 +102,7 @@ def fn_task():
         exception = None
         task_id = '7c6731af-9533-40c3-83a9-25b58f0d837f'
         try:
-            remove_span(fn_task, task_id)
+            detach_span(fn_task, task_id)
         except Exception as e:
             exception = e
         ok_(exception is None)
@@ -117,7 +117,7 @@ def fn_task():
 
         # propagate and finish a Span for `fn_task`
         task_id = '7c6731af-9533-40c3-83a9-25b58f0d837f'
-        propagate_span(fn_task, task_id, self.tracer.trace('celery.run'))
+        attach_span(fn_task, task_id, self.tracer.trace('celery.run'))
         weak_dict = getattr(fn_task, '__dd_task_span')
         ok_(weak_dict.get(task_id))
         # flush data and force the GC