We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Celery Instrumentation with apply_async function
I have a basic django application which uses celery to call a method using apply_async and delay functions. I am trying to integrate Otel into my django application. I am using  0.10b0 version of all Otel libraries. I have used celery-auto instrumentation for produce the spans/traces. But I see that the celery-auto instrumentation isnt working. 

**Steps to reproduce**
I have attached my code in order to reproduce the issue. 
1. Run the python application using 
`python manage.py runserver`
<img width="684" alt="Screen Shot 2020-07-01 at 1 54 05 PM" src="https://user-images.githubusercontent.com/11060386/86276134-5bf37780-bba2-11ea-951e-75cca763ee55.png">

2. Then also start the celery worker using
celery -A django_email_celery worker --loglevel=info
<img width="1352" alt="Screen Shot 2020-07-01 at 1 55 11 PM" src="https://user-images.githubusercontent.com/11060386/86276219-82b1ae00-bba2-11ea-9a8d-96454683a694.png">


**What is the expected behavior?**
Once you hit the `localhost` url, I see the trace and spans created by the Otel. But I was expecting the `traceid` to continue from method `index()` in `views.py` to `send_log` in `tasks.py`. 

**What is the actual behavior?**
But then the a new trace is being created at the `send_log`. 
<img width="748" alt="Screen Shot 2020-07-01 at 1 59 29 PM" src="https://user-images.githubusercontent.com/11060386/86276630-331fb200-bba3-11ea-9eab-5df16112a275.png">
<img width="1336" alt="Screen Shot 2020-07-01 at 2 01 44 PM" src="https://user-images.githubusercontent.com/11060386/86276746-737f3000-bba3-11ea-8750-be0e416399f0.png">

**Other Info**
I have added celery instrumentation in `celery.py` file

[django_email_celery.zip](https://github.com/open-telemetry/opentelemetry-python/files/4859611/django_email_celery.zip)

@ocelotl @codeboten 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py`
Content:
```
1 # Copyright The OpenTelemetry Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """
15 Instrument `celery`_ to trace Celery applications.
16 
17 .. _celery: https://pypi.org/project/celery/
18 
19 Usage
20 -----
21 
22 * Start broker backend
23 
24 .. code::
25 
26     docker run -p 5672:5672 rabbitmq
27 
28 
29 * Run instrumented task
30 
31 .. code:: python
32 
33     from opentelemetry.instrumentation.celery import CeleryInstrumentor
34 
35     CeleryInstrumentor().instrument()
36 
37     from celery import Celery
38 
39     app = Celery("tasks", broker="amqp://localhost")
40 
41     @app.task
42     def add(x, y):
43         return x + y
44 
45     add.delay(42, 50)
46 
47 API
48 ---
49 """
50 
51 import logging
52 import signal
53 
54 from celery import signals  # pylint: disable=no-name-in-module
55 
56 from opentelemetry import trace
57 from opentelemetry.instrumentation.celery import utils
58 from opentelemetry.instrumentation.celery.version import __version__
59 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
60 from opentelemetry.trace.status import Status, StatusCanonicalCode
61 
62 logger = logging.getLogger(__name__)
63 
64 # Task operations
65 _TASK_TAG_KEY = "celery.action"
66 _TASK_APPLY_ASYNC = "apply_async"
67 _TASK_RUN = "run"
68 
69 _TASK_RETRY_REASON_KEY = "celery.retry.reason"
70 _TASK_REVOKED_REASON_KEY = "celery.revoked.reason"
71 _TASK_REVOKED_TERMINATED_SIGNAL_KEY = "celery.terminated.signal"
72 _TASK_NAME_KEY = "celery.task_name"
73 _MESSAGE_ID_ATTRIBUTE_NAME = "messaging.message_id"
74 
75 
76 class CeleryInstrumentor(BaseInstrumentor):
77     def _instrument(self, **kwargs):
78         tracer_provider = kwargs.get("tracer_provider")
79 
80         # pylint: disable=attribute-defined-outside-init
81         self._tracer = trace.get_tracer(__name__, __version__, tracer_provider)
82 
83         signals.task_prerun.connect(self._trace_prerun, weak=False)
84         signals.task_postrun.connect(self._trace_postrun, weak=False)
85         signals.before_task_publish.connect(
86             self._trace_before_publish, weak=False
87         )
88         signals.after_task_publish.connect(
89             self._trace_after_publish, weak=False
90         )
91         signals.task_failure.connect(self._trace_failure, weak=False)
92         signals.task_retry.connect(self._trace_retry, weak=False)
93 
94     def _uninstrument(self, **kwargs):
95         signals.task_prerun.disconnect(self._trace_prerun)
96         signals.task_postrun.disconnect(self._trace_postrun)
97         signals.before_task_publish.disconnect(self._trace_before_publish)
98         signals.after_task_publish.disconnect(self._trace_after_publish)
99         signals.task_failure.disconnect(self._trace_failure)
100         signals.task_retry.disconnect(self._trace_retry)
101 
102     def _trace_prerun(self, *args, **kwargs):
103         task = utils.retrieve_task(kwargs)
104         task_id = utils.retrieve_task_id(kwargs)
105 
106         if task is None or task_id is None:
107             return
108 
109         logger.debug("prerun signal start task_id=%s", task_id)
110 
111         span = self._tracer.start_span(task.name, kind=trace.SpanKind.CONSUMER)
112 
113         activation = self._tracer.use_span(span, end_on_exit=True)
114         activation.__enter__()
115         utils.attach_span(task, task_id, (span, activation))
116 
117     @staticmethod
118     def _trace_postrun(*args, **kwargs):
119         task = utils.retrieve_task(kwargs)
120         task_id = utils.retrieve_task_id(kwargs)
121 
122         if task is None or task_id is None:
123             return
124 
125         logger.debug("postrun signal task_id=%s", task_id)
126 
127         # retrieve and finish the Span
128         span, activation = utils.retrieve_span(task, task_id)
129         if span is None:
130             logger.warning("no existing span found for task_id=%s", task_id)
131             return
132 
133         # request context tags
134         span.set_attribute(_TASK_TAG_KEY, _TASK_RUN)
135         utils.set_attributes_from_context(span, kwargs)
136         utils.set_attributes_from_context(span, task.request)
137         span.set_attribute(_TASK_NAME_KEY, task.name)
138 
139         activation.__exit__(None, None, None)
140         utils.detach_span(task, task_id)
141 
142     def _trace_before_publish(self, *args, **kwargs):
143         task = utils.retrieve_task_from_sender(kwargs)
144         task_id = utils.retrieve_task_id_from_message(kwargs)
145 
146         if task is None or task_id is None:
147             return
148 
149         span = self._tracer.start_span(task.name, kind=trace.SpanKind.PRODUCER)
150 
151         # apply some attributes here because most of the data is not available
152         span.set_attribute(_TASK_TAG_KEY, _TASK_APPLY_ASYNC)
153         span.set_attribute(_MESSAGE_ID_ATTRIBUTE_NAME, task_id)
154         span.set_attribute(_TASK_NAME_KEY, task.name)
155         utils.set_attributes_from_context(span, kwargs)
156 
157         activation = self._tracer.use_span(span, end_on_exit=True)
158         activation.__enter__()
159         utils.attach_span(task, task_id, (span, activation), is_publish=True)
160 
161     @staticmethod
162     def _trace_after_publish(*args, **kwargs):
163         task = utils.retrieve_task_from_sender(kwargs)
164         task_id = utils.retrieve_task_id_from_message(kwargs)
165 
166         if task is None or task_id is None:
167             return
168 
169         # retrieve and finish the Span
170         _, activation = utils.retrieve_span(task, task_id, is_publish=True)
171         if activation is None:
172             logger.warning("no existing span found for task_id=%s", task_id)
173             return
174 
175         activation.__exit__(None, None, None)
176         utils.detach_span(task, task_id, is_publish=True)
177 
178     @staticmethod
179     def _trace_failure(*args, **kwargs):
180         task = utils.retrieve_task_from_sender(kwargs)
181         task_id = utils.retrieve_task_id(kwargs)
182 
183         if task is None or task_id is None:
184             return
185 
186         # retrieve and pass exception info to activation
187         span, _ = utils.retrieve_span(task, task_id)
188         if span is None:
189             return
190 
191         status_kwargs = {"canonical_code": StatusCanonicalCode.UNKNOWN}
192 
193         ex = kwargs.get("einfo")
194 
195         if (
196             hasattr(task, "throws")
197             and ex is not None
198             and isinstance(ex.exception, task.throws)
199         ):
200             return
201 
202         if ex is not None:
203             status_kwargs["description"] = str(ex)
204 
205         span.set_status(Status(**status_kwargs))
206 
207     @staticmethod
208     def _trace_retry(*args, **kwargs):
209         task = utils.retrieve_task_from_sender(kwargs)
210         task_id = utils.retrieve_task_id_from_request(kwargs)
211         reason = utils.retrieve_reason(kwargs)
212 
213         if task is None or task_id is None or reason is None:
214             return
215 
216         span, _ = utils.retrieve_span(task, task_id)
217         if span is None:
218             return
219 
220         # Add retry reason metadata to span
221         # Use `str(reason)` instead of `reason.message` in case we get
222         # something that isn't an `Exception`
223         span.set_attribute(_TASK_RETRY_REASON_KEY, str(reason))
224
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py b/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py
--- a/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py
@@ -30,11 +30,20 @@
 
 .. code:: python
 
+    from opentelemetry import trace
+    from opentelemetry.sdk.trace import TracerProvider
+    from opentelemetry.sdk.trace.export import BatchExportSpanProcessor
     from opentelemetry.instrumentation.celery import CeleryInstrumentor
 
-    CeleryInstrumentor().instrument()
-
     from celery import Celery
+    from celery.signals import worker_process_init
+
+    @worker_process_init.connect(weak=False)
+    def init_celery_tracing(*args, **kwargs):
+        trace.set_tracer_provider(TracerProvider())
+        span_processor = BatchExportSpanProcessor(ConsoleSpanExporter())
+        trace.get_tracer_provider().add_span_processor(span_processor)
+        CeleryInstrumentor().instrument()
 
     app = Celery("tasks", broker="amqp://localhost")
 
@@ -50,13 +59,15 @@
 
 import logging
 import signal
+from collections.abc import Iterable
 
 from celery import signals  # pylint: disable=no-name-in-module
 
-from opentelemetry import trace
+from opentelemetry import propagators, trace
 from opentelemetry.instrumentation.celery import utils
 from opentelemetry.instrumentation.celery.version import __version__
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
+from opentelemetry.trace.propagation import get_current_span
 from opentelemetry.trace.status import Status, StatusCanonicalCode
 
 logger = logging.getLogger(__name__)
@@ -106,9 +117,16 @@
         if task is None or task_id is None:
             return
 
+        request = task.request
+        tracectx = propagators.extract(carrier_extractor, request) or {}
+        parent = get_current_span(tracectx)
+
         logger.debug("prerun signal start task_id=%s", task_id)
 
-        span = self._tracer.start_span(task.name, kind=trace.SpanKind.CONSUMER)
+        operation_name = "{0}/{1}".format(_TASK_RUN, task.name)
+        span = self._tracer.start_span(
+            operation_name, parent=parent, kind=trace.SpanKind.CONSUMER
+        )
 
         activation = self._tracer.use_span(span, end_on_exit=True)
         activation.__enter__()
@@ -146,7 +164,10 @@
         if task is None or task_id is None:
             return
 
-        span = self._tracer.start_span(task.name, kind=trace.SpanKind.PRODUCER)
+        operation_name = "{0}/{1}".format(_TASK_APPLY_ASYNC, task.name)
+        span = self._tracer.start_span(
+            operation_name, kind=trace.SpanKind.PRODUCER
+        )
 
         # apply some attributes here because most of the data is not available
         span.set_attribute(_TASK_TAG_KEY, _TASK_APPLY_ASYNC)
@@ -158,6 +179,10 @@
         activation.__enter__()
         utils.attach_span(task, task_id, (span, activation), is_publish=True)
 
+        headers = kwargs.get("headers")
+        if headers:
+            propagators.inject(type(headers).__setitem__, headers)
+
     @staticmethod
     def _trace_after_publish(*args, **kwargs):
         task = utils.retrieve_task_from_sender(kwargs)
@@ -221,3 +246,10 @@
         # Use `str(reason)` instead of `reason.message` in case we get
         # something that isn't an `Exception`
         span.set_attribute(_TASK_RETRY_REASON_KEY, str(reason))
+
+
+def carrier_extractor(carrier, key):
+    value = getattr(carrier, key, [])
+    if isinstance(value, str) or not isinstance(value, Iterable):
+        value = (value,)
+    return value

{"golden_diff": "diff --git a/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py b/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py\n--- a/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py\n+++ b/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py\n@@ -30,11 +30,20 @@\n \n .. code:: python\n \n+    from opentelemetry import trace\n+    from opentelemetry.sdk.trace import TracerProvider\n+    from opentelemetry.sdk.trace.export import BatchExportSpanProcessor\n     from opentelemetry.instrumentation.celery import CeleryInstrumentor\n \n-    CeleryInstrumentor().instrument()\n-\n     from celery import Celery\n+    from celery.signals import worker_process_init\n+\n+    @worker_process_init.connect(weak=False)\n+    def init_celery_tracing(*args, **kwargs):\n+        trace.set_tracer_provider(TracerProvider())\n+        span_processor = BatchExportSpanProcessor(ConsoleSpanExporter())\n+        trace.get_tracer_provider().add_span_processor(span_processor)\n+        CeleryInstrumentor().instrument()\n \n     app = Celery(\"tasks\", broker=\"amqp://localhost\")\n \n@@ -50,13 +59,15 @@\n \n import logging\n import signal\n+from collections.abc import Iterable\n \n from celery import signals  # pylint: disable=no-name-in-module\n \n-from opentelemetry import trace\n+from opentelemetry import propagators, trace\n from opentelemetry.instrumentation.celery import utils\n from opentelemetry.instrumentation.celery.version import __version__\n from opentelemetry.instrumentation.instrumentor import BaseInstrumentor\n+from opentelemetry.trace.propagation import get_current_span\n from opentelemetry.trace.status import Status, StatusCanonicalCode\n \n logger = logging.getLogger(__name__)\n@@ -106,9 +117,16 @@\n         if task is None or task_id is None:\n             return\n \n+        request = task.request\n+        tracectx = propagators.extract(carrier_extractor, request) or {}\n+        parent = get_current_span(tracectx)\n+\n         logger.debug(\"prerun signal start task_id=%s\", task_id)\n \n-        span = self._tracer.start_span(task.name, kind=trace.SpanKind.CONSUMER)\n+        operation_name = \"{0}/{1}\".format(_TASK_RUN, task.name)\n+        span = self._tracer.start_span(\n+            operation_name, parent=parent, kind=trace.SpanKind.CONSUMER\n+        )\n \n         activation = self._tracer.use_span(span, end_on_exit=True)\n         activation.__enter__()\n@@ -146,7 +164,10 @@\n         if task is None or task_id is None:\n             return\n \n-        span = self._tracer.start_span(task.name, kind=trace.SpanKind.PRODUCER)\n+        operation_name = \"{0}/{1}\".format(_TASK_APPLY_ASYNC, task.name)\n+        span = self._tracer.start_span(\n+            operation_name, kind=trace.SpanKind.PRODUCER\n+        )\n \n         # apply some attributes here because most of the data is not available\n         span.set_attribute(_TASK_TAG_KEY, _TASK_APPLY_ASYNC)\n@@ -158,6 +179,10 @@\n         activation.__enter__()\n         utils.attach_span(task, task_id, (span, activation), is_publish=True)\n \n+        headers = kwargs.get(\"headers\")\n+        if headers:\n+            propagators.inject(type(headers).__setitem__, headers)\n+\n     @staticmethod\n     def _trace_after_publish(*args, **kwargs):\n         task = utils.retrieve_task_from_sender(kwargs)\n@@ -221,3 +246,10 @@\n         # Use `str(reason)` instead of `reason.message` in case we get\n         # something that isn't an `Exception`\n         span.set_attribute(_TASK_RETRY_REASON_KEY, str(reason))\n+\n+\n+def carrier_extractor(carrier, key):\n+    value = getattr(carrier, key, [])\n+    if isinstance(value, str) or not isinstance(value, Iterable):\n+        value = (value,)\n+    return value\n", "issue": "Celery Instrumentation with apply_async function\nI have a basic django application which uses celery to call a method using apply_async and delay functions. I am trying to integrate Otel into my django application. I am using  0.10b0 version of all Otel libraries. I have used celery-auto instrumentation for produce the spans/traces. But I see that the celery-auto instrumentation isnt working. \r\n\r\n**Steps to reproduce**\r\nI have attached my code in order to reproduce the issue. \r\n1. Run the python application using \r\n`python manage.py runserver`\r\n<img width=\"684\" alt=\"Screen Shot 2020-07-01 at 1 54 05 PM\" src=\"https://user-images.githubusercontent.com/11060386/86276134-5bf37780-bba2-11ea-951e-75cca763ee55.png\">\r\n\r\n2. Then also start the celery worker using\r\ncelery -A django_email_celery worker --loglevel=info\r\n<img width=\"1352\" alt=\"Screen Shot 2020-07-01 at 1 55 11 PM\" src=\"https://user-images.githubusercontent.com/11060386/86276219-82b1ae00-bba2-11ea-9a8d-96454683a694.png\">\r\n\r\n\r\n**What is the expected behavior?**\r\nOnce you hit the `localhost` url, I see the trace and spans created by the Otel. But I was expecting the `traceid` to continue from method `index()` in `views.py` to `send_log` in `tasks.py`. \r\n\r\n**What is the actual behavior?**\r\nBut then the a new trace is being created at the `send_log`. \r\n<img width=\"748\" alt=\"Screen Shot 2020-07-01 at 1 59 29 PM\" src=\"https://user-images.githubusercontent.com/11060386/86276630-331fb200-bba3-11ea-9eab-5df16112a275.png\">\r\n<img width=\"1336\" alt=\"Screen Shot 2020-07-01 at 2 01 44 PM\" src=\"https://user-images.githubusercontent.com/11060386/86276746-737f3000-bba3-11ea-8750-be0e416399f0.png\">\r\n\r\n**Other Info**\r\nI have added celery instrumentation in `celery.py` file\r\n\r\n[django_email_celery.zip](https://github.com/open-telemetry/opentelemetry-python/files/4859611/django_email_celery.zip)\r\n\r\n@ocelotl @codeboten \n", "before_files": [{"content": "# Copyright The OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"\nInstrument `celery`_ to trace Celery applications.\n\n.. _celery: https://pypi.org/project/celery/\n\nUsage\n-----\n\n* Start broker backend\n\n.. code::\n\n    docker run -p 5672:5672 rabbitmq\n\n\n* Run instrumented task\n\n.. code:: python\n\n    from opentelemetry.instrumentation.celery import CeleryInstrumentor\n\n    CeleryInstrumentor().instrument()\n\n    from celery import Celery\n\n    app = Celery(\"tasks\", broker=\"amqp://localhost\")\n\n    @app.task\n    def add(x, y):\n        return x + y\n\n    add.delay(42, 50)\n\nAPI\n---\n\"\"\"\n\nimport logging\nimport signal\n\nfrom celery import signals  # pylint: disable=no-name-in-module\n\nfrom opentelemetry import trace\nfrom opentelemetry.instrumentation.celery import utils\nfrom opentelemetry.instrumentation.celery.version import __version__\nfrom opentelemetry.instrumentation.instrumentor import BaseInstrumentor\nfrom opentelemetry.trace.status import Status, StatusCanonicalCode\n\nlogger = logging.getLogger(__name__)\n\n# Task operations\n_TASK_TAG_KEY = \"celery.action\"\n_TASK_APPLY_ASYNC = \"apply_async\"\n_TASK_RUN = \"run\"\n\n_TASK_RETRY_REASON_KEY = \"celery.retry.reason\"\n_TASK_REVOKED_REASON_KEY = \"celery.revoked.reason\"\n_TASK_REVOKED_TERMINATED_SIGNAL_KEY = \"celery.terminated.signal\"\n_TASK_NAME_KEY = \"celery.task_name\"\n_MESSAGE_ID_ATTRIBUTE_NAME = \"messaging.message_id\"\n\n\nclass CeleryInstrumentor(BaseInstrumentor):\n    def _instrument(self, **kwargs):\n        tracer_provider = kwargs.get(\"tracer_provider\")\n\n        # pylint: disable=attribute-defined-outside-init\n        self._tracer = trace.get_tracer(__name__, __version__, tracer_provider)\n\n        signals.task_prerun.connect(self._trace_prerun, weak=False)\n        signals.task_postrun.connect(self._trace_postrun, weak=False)\n        signals.before_task_publish.connect(\n            self._trace_before_publish, weak=False\n        )\n        signals.after_task_publish.connect(\n            self._trace_after_publish, weak=False\n        )\n        signals.task_failure.connect(self._trace_failure, weak=False)\n        signals.task_retry.connect(self._trace_retry, weak=False)\n\n    def _uninstrument(self, **kwargs):\n        signals.task_prerun.disconnect(self._trace_prerun)\n        signals.task_postrun.disconnect(self._trace_postrun)\n        signals.before_task_publish.disconnect(self._trace_before_publish)\n        signals.after_task_publish.disconnect(self._trace_after_publish)\n        signals.task_failure.disconnect(self._trace_failure)\n        signals.task_retry.disconnect(self._trace_retry)\n\n    def _trace_prerun(self, *args, **kwargs):\n        task = utils.retrieve_task(kwargs)\n        task_id = utils.retrieve_task_id(kwargs)\n\n        if task is None or task_id is None:\n            return\n\n        logger.debug(\"prerun signal start task_id=%s\", task_id)\n\n        span = self._tracer.start_span(task.name, kind=trace.SpanKind.CONSUMER)\n\n        activation = self._tracer.use_span(span, end_on_exit=True)\n        activation.__enter__()\n        utils.attach_span(task, task_id, (span, activation))\n\n    @staticmethod\n    def _trace_postrun(*args, **kwargs):\n        task = utils.retrieve_task(kwargs)\n        task_id = utils.retrieve_task_id(kwargs)\n\n        if task is None or task_id is None:\n            return\n\n        logger.debug(\"postrun signal task_id=%s\", task_id)\n\n        # retrieve and finish the Span\n        span, activation = utils.retrieve_span(task, task_id)\n        if span is None:\n            logger.warning(\"no existing span found for task_id=%s\", task_id)\n            return\n\n        # request context tags\n        span.set_attribute(_TASK_TAG_KEY, _TASK_RUN)\n        utils.set_attributes_from_context(span, kwargs)\n        utils.set_attributes_from_context(span, task.request)\n        span.set_attribute(_TASK_NAME_KEY, task.name)\n\n        activation.__exit__(None, None, None)\n        utils.detach_span(task, task_id)\n\n    def _trace_before_publish(self, *args, **kwargs):\n        task = utils.retrieve_task_from_sender(kwargs)\n        task_id = utils.retrieve_task_id_from_message(kwargs)\n\n        if task is None or task_id is None:\n            return\n\n        span = self._tracer.start_span(task.name, kind=trace.SpanKind.PRODUCER)\n\n        # apply some attributes here because most of the data is not available\n        span.set_attribute(_TASK_TAG_KEY, _TASK_APPLY_ASYNC)\n        span.set_attribute(_MESSAGE_ID_ATTRIBUTE_NAME, task_id)\n        span.set_attribute(_TASK_NAME_KEY, task.name)\n        utils.set_attributes_from_context(span, kwargs)\n\n        activation = self._tracer.use_span(span, end_on_exit=True)\n        activation.__enter__()\n        utils.attach_span(task, task_id, (span, activation), is_publish=True)\n\n    @staticmethod\n    def _trace_after_publish(*args, **kwargs):\n        task = utils.retrieve_task_from_sender(kwargs)\n        task_id = utils.retrieve_task_id_from_message(kwargs)\n\n        if task is None or task_id is None:\n            return\n\n        # retrieve and finish the Span\n        _, activation = utils.retrieve_span(task, task_id, is_publish=True)\n        if activation is None:\n            logger.warning(\"no existing span found for task_id=%s\", task_id)\n            return\n\n        activation.__exit__(None, None, None)\n        utils.detach_span(task, task_id, is_publish=True)\n\n    @staticmethod\n    def _trace_failure(*args, **kwargs):\n        task = utils.retrieve_task_from_sender(kwargs)\n        task_id = utils.retrieve_task_id(kwargs)\n\n        if task is None or task_id is None:\n            return\n\n        # retrieve and pass exception info to activation\n        span, _ = utils.retrieve_span(task, task_id)\n        if span is None:\n            return\n\n        status_kwargs = {\"canonical_code\": StatusCanonicalCode.UNKNOWN}\n\n        ex = kwargs.get(\"einfo\")\n\n        if (\n            hasattr(task, \"throws\")\n            and ex is not None\n            and isinstance(ex.exception, task.throws)\n        ):\n            return\n\n        if ex is not None:\n            status_kwargs[\"description\"] = str(ex)\n\n        span.set_status(Status(**status_kwargs))\n\n    @staticmethod\n    def _trace_retry(*args, **kwargs):\n        task = utils.retrieve_task_from_sender(kwargs)\n        task_id = utils.retrieve_task_id_from_request(kwargs)\n        reason = utils.retrieve_reason(kwargs)\n\n        if task is None or task_id is None or reason is None:\n            return\n\n        span, _ = utils.retrieve_span(task, task_id)\n        if span is None:\n            return\n\n        # Add retry reason metadata to span\n        # Use `str(reason)` instead of `reason.message` in case we get\n        # something that isn't an `Exception`\n        span.set_attribute(_TASK_RETRY_REASON_KEY, str(reason))\n", "path": "instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py"}], "after_files": [{"content": "# Copyright The OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"\nInstrument `celery`_ to trace Celery applications.\n\n.. _celery: https://pypi.org/project/celery/\n\nUsage\n-----\n\n* Start broker backend\n\n.. code::\n\n    docker run -p 5672:5672 rabbitmq\n\n\n* Run instrumented task\n\n.. code:: python\n\n    from opentelemetry import trace\n    from opentelemetry.sdk.trace import TracerProvider\n    from opentelemetry.sdk.trace.export import BatchExportSpanProcessor\n    from opentelemetry.instrumentation.celery import CeleryInstrumentor\n\n    from celery import Celery\n    from celery.signals import worker_process_init\n\n    @worker_process_init.connect(weak=False)\n    def init_celery_tracing(*args, **kwargs):\n        trace.set_tracer_provider(TracerProvider())\n        span_processor = BatchExportSpanProcessor(ConsoleSpanExporter())\n        trace.get_tracer_provider().add_span_processor(span_processor)\n        CeleryInstrumentor().instrument()\n\n    app = Celery(\"tasks\", broker=\"amqp://localhost\")\n\n    @app.task\n    def add(x, y):\n        return x + y\n\n    add.delay(42, 50)\n\nAPI\n---\n\"\"\"\n\nimport logging\nimport signal\nfrom collections.abc import Iterable\n\nfrom celery import signals  # pylint: disable=no-name-in-module\n\nfrom opentelemetry import propagators, trace\nfrom opentelemetry.instrumentation.celery import utils\nfrom opentelemetry.instrumentation.celery.version import __version__\nfrom opentelemetry.instrumentation.instrumentor import BaseInstrumentor\nfrom opentelemetry.trace.propagation import get_current_span\nfrom opentelemetry.trace.status import Status, StatusCanonicalCode\n\nlogger = logging.getLogger(__name__)\n\n# Task operations\n_TASK_TAG_KEY = \"celery.action\"\n_TASK_APPLY_ASYNC = \"apply_async\"\n_TASK_RUN = \"run\"\n\n_TASK_RETRY_REASON_KEY = \"celery.retry.reason\"\n_TASK_REVOKED_REASON_KEY = \"celery.revoked.reason\"\n_TASK_REVOKED_TERMINATED_SIGNAL_KEY = \"celery.terminated.signal\"\n_TASK_NAME_KEY = \"celery.task_name\"\n_MESSAGE_ID_ATTRIBUTE_NAME = \"messaging.message_id\"\n\n\nclass CeleryInstrumentor(BaseInstrumentor):\n    def _instrument(self, **kwargs):\n        tracer_provider = kwargs.get(\"tracer_provider\")\n\n        # pylint: disable=attribute-defined-outside-init\n        self._tracer = trace.get_tracer(__name__, __version__, tracer_provider)\n\n        signals.task_prerun.connect(self._trace_prerun, weak=False)\n        signals.task_postrun.connect(self._trace_postrun, weak=False)\n        signals.before_task_publish.connect(\n            self._trace_before_publish, weak=False\n        )\n        signals.after_task_publish.connect(\n            self._trace_after_publish, weak=False\n        )\n        signals.task_failure.connect(self._trace_failure, weak=False)\n        signals.task_retry.connect(self._trace_retry, weak=False)\n\n    def _uninstrument(self, **kwargs):\n        signals.task_prerun.disconnect(self._trace_prerun)\n        signals.task_postrun.disconnect(self._trace_postrun)\n        signals.before_task_publish.disconnect(self._trace_before_publish)\n        signals.after_task_publish.disconnect(self._trace_after_publish)\n        signals.task_failure.disconnect(self._trace_failure)\n        signals.task_retry.disconnect(self._trace_retry)\n\n    def _trace_prerun(self, *args, **kwargs):\n        task = utils.retrieve_task(kwargs)\n        task_id = utils.retrieve_task_id(kwargs)\n\n        if task is None or task_id is None:\n            return\n\n        request = task.request\n        tracectx = propagators.extract(carrier_extractor, request) or {}\n        parent = get_current_span(tracectx)\n\n        logger.debug(\"prerun signal start task_id=%s\", task_id)\n\n        operation_name = \"{0}/{1}\".format(_TASK_RUN, task.name)\n        span = self._tracer.start_span(\n            operation_name, parent=parent, kind=trace.SpanKind.CONSUMER\n        )\n\n        activation = self._tracer.use_span(span, end_on_exit=True)\n        activation.__enter__()\n        utils.attach_span(task, task_id, (span, activation))\n\n    @staticmethod\n    def _trace_postrun(*args, **kwargs):\n        task = utils.retrieve_task(kwargs)\n        task_id = utils.retrieve_task_id(kwargs)\n\n        if task is None or task_id is None:\n            return\n\n        logger.debug(\"postrun signal task_id=%s\", task_id)\n\n        # retrieve and finish the Span\n        span, activation = utils.retrieve_span(task, task_id)\n        if span is None:\n            logger.warning(\"no existing span found for task_id=%s\", task_id)\n            return\n\n        # request context tags\n        span.set_attribute(_TASK_TAG_KEY, _TASK_RUN)\n        utils.set_attributes_from_context(span, kwargs)\n        utils.set_attributes_from_context(span, task.request)\n        span.set_attribute(_TASK_NAME_KEY, task.name)\n\n        activation.__exit__(None, None, None)\n        utils.detach_span(task, task_id)\n\n    def _trace_before_publish(self, *args, **kwargs):\n        task = utils.retrieve_task_from_sender(kwargs)\n        task_id = utils.retrieve_task_id_from_message(kwargs)\n\n        if task is None or task_id is None:\n            return\n\n        operation_name = \"{0}/{1}\".format(_TASK_APPLY_ASYNC, task.name)\n        span = self._tracer.start_span(\n            operation_name, kind=trace.SpanKind.PRODUCER\n        )\n\n        # apply some attributes here because most of the data is not available\n        span.set_attribute(_TASK_TAG_KEY, _TASK_APPLY_ASYNC)\n        span.set_attribute(_MESSAGE_ID_ATTRIBUTE_NAME, task_id)\n        span.set_attribute(_TASK_NAME_KEY, task.name)\n        utils.set_attributes_from_context(span, kwargs)\n\n        activation = self._tracer.use_span(span, end_on_exit=True)\n        activation.__enter__()\n        utils.attach_span(task, task_id, (span, activation), is_publish=True)\n\n        headers = kwargs.get(\"headers\")\n        if headers:\n            propagators.inject(type(headers).__setitem__, headers)\n\n    @staticmethod\n    def _trace_after_publish(*args, **kwargs):\n        task = utils.retrieve_task_from_sender(kwargs)\n        task_id = utils.retrieve_task_id_from_message(kwargs)\n\n        if task is None or task_id is None:\n            return\n\n        # retrieve and finish the Span\n        _, activation = utils.retrieve_span(task, task_id, is_publish=True)\n        if activation is None:\n            logger.warning(\"no existing span found for task_id=%s\", task_id)\n            return\n\n        activation.__exit__(None, None, None)\n        utils.detach_span(task, task_id, is_publish=True)\n\n    @staticmethod\n    def _trace_failure(*args, **kwargs):\n        task = utils.retrieve_task_from_sender(kwargs)\n        task_id = utils.retrieve_task_id(kwargs)\n\n        if task is None or task_id is None:\n            return\n\n        # retrieve and pass exception info to activation\n        span, _ = utils.retrieve_span(task, task_id)\n        if span is None:\n            return\n\n        status_kwargs = {\"canonical_code\": StatusCanonicalCode.UNKNOWN}\n\n        ex = kwargs.get(\"einfo\")\n\n        if (\n            hasattr(task, \"throws\")\n            and ex is not None\n            and isinstance(ex.exception, task.throws)\n        ):\n            return\n\n        if ex is not None:\n            status_kwargs[\"description\"] = str(ex)\n\n        span.set_status(Status(**status_kwargs))\n\n    @staticmethod\n    def _trace_retry(*args, **kwargs):\n        task = utils.retrieve_task_from_sender(kwargs)\n        task_id = utils.retrieve_task_id_from_request(kwargs)\n        reason = utils.retrieve_reason(kwargs)\n\n        if task is None or task_id is None or reason is None:\n            return\n\n        span, _ = utils.retrieve_span(task, task_id)\n        if span is None:\n            return\n\n        # Add retry reason metadata to span\n        # Use `str(reason)` instead of `reason.message` in case we get\n        # something that isn't an `Exception`\n        span.set_attribute(_TASK_RETRY_REASON_KEY, str(reason))\n\n\ndef carrier_extractor(carrier, key):\n    value = getattr(carrier, key, [])\n    if isinstance(value, str) or not isinstance(value, Iterable):\n        value = (value,)\n    return value\n", "path": "instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py"}]}