We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Metrics instrumentation celery
Relevant semconv: https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/semantic_conventions/instrumentation/kafka.md

--- 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     from celery import Celery
36     from celery.signals import worker_process_init
37 
38     @worker_process_init.connect(weak=False)
39     def init_celery_tracing(*args, **kwargs):
40         CeleryInstrumentor().instrument()
41 
42     app = Celery("tasks", broker="amqp://localhost")
43 
44     @app.task
45     def add(x, y):
46         return x + y
47 
48     add.delay(42, 50)
49 
50 Setting up tracing
51 ------------------
52 
53 When tracing a celery worker process, tracing and instrumention both must be initialized after the celery worker
54 process is initialized. This is required for any tracing components that might use threading to work correctly
55 such as the BatchSpanProcessor. Celery provides a signal called ``worker_process_init`` that can be used to
56 accomplish this as shown in the example above.
57 
58 API
59 ---
60 """
61 
62 import logging
63 from typing import Collection, Iterable
64 
65 from celery import signals  # pylint: disable=no-name-in-module
66 
67 from opentelemetry import trace
68 from opentelemetry.instrumentation.celery import utils
69 from opentelemetry.instrumentation.celery.package import _instruments
70 from opentelemetry.instrumentation.celery.version import __version__
71 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
72 from opentelemetry.propagate import extract, inject
73 from opentelemetry.propagators.textmap import Getter
74 from opentelemetry.semconv.trace import SpanAttributes
75 from opentelemetry.trace.status import Status, StatusCode
76 
77 logger = logging.getLogger(__name__)
78 
79 # Task operations
80 _TASK_TAG_KEY = "celery.action"
81 _TASK_APPLY_ASYNC = "apply_async"
82 _TASK_RUN = "run"
83 
84 _TASK_RETRY_REASON_KEY = "celery.retry.reason"
85 _TASK_REVOKED_REASON_KEY = "celery.revoked.reason"
86 _TASK_REVOKED_TERMINATED_SIGNAL_KEY = "celery.terminated.signal"
87 _TASK_NAME_KEY = "celery.task_name"
88 
89 
90 class CeleryGetter(Getter):
91     def get(self, carrier, key):
92         value = getattr(carrier, key, None)
93         if value is None:
94             return None
95         if isinstance(value, str) or not isinstance(value, Iterable):
96             value = (value,)
97         return value
98 
99     def keys(self, carrier):
100         return []
101 
102 
103 celery_getter = CeleryGetter()
104 
105 
106 class CeleryInstrumentor(BaseInstrumentor):
107     def instrumentation_dependencies(self) -> Collection[str]:
108         return _instruments
109 
110     def _instrument(self, **kwargs):
111         tracer_provider = kwargs.get("tracer_provider")
112 
113         # pylint: disable=attribute-defined-outside-init
114         self._tracer = trace.get_tracer(__name__, __version__, tracer_provider)
115 
116         signals.task_prerun.connect(self._trace_prerun, weak=False)
117         signals.task_postrun.connect(self._trace_postrun, weak=False)
118         signals.before_task_publish.connect(
119             self._trace_before_publish, weak=False
120         )
121         signals.after_task_publish.connect(
122             self._trace_after_publish, weak=False
123         )
124         signals.task_failure.connect(self._trace_failure, weak=False)
125         signals.task_retry.connect(self._trace_retry, weak=False)
126 
127     def _uninstrument(self, **kwargs):
128         signals.task_prerun.disconnect(self._trace_prerun)
129         signals.task_postrun.disconnect(self._trace_postrun)
130         signals.before_task_publish.disconnect(self._trace_before_publish)
131         signals.after_task_publish.disconnect(self._trace_after_publish)
132         signals.task_failure.disconnect(self._trace_failure)
133         signals.task_retry.disconnect(self._trace_retry)
134 
135     def _trace_prerun(self, *args, **kwargs):
136         task = utils.retrieve_task(kwargs)
137         task_id = utils.retrieve_task_id(kwargs)
138 
139         if task is None or task_id is None:
140             return
141 
142         request = task.request
143         tracectx = extract(request, getter=celery_getter) or None
144 
145         logger.debug("prerun signal start task_id=%s", task_id)
146 
147         operation_name = f"{_TASK_RUN}/{task.name}"
148         span = self._tracer.start_span(
149             operation_name, context=tracectx, kind=trace.SpanKind.CONSUMER
150         )
151 
152         activation = trace.use_span(span, end_on_exit=True)
153         activation.__enter__()  # pylint: disable=E1101
154         utils.attach_span(task, task_id, (span, activation))
155 
156     @staticmethod
157     def _trace_postrun(*args, **kwargs):
158         task = utils.retrieve_task(kwargs)
159         task_id = utils.retrieve_task_id(kwargs)
160 
161         if task is None or task_id is None:
162             return
163 
164         logger.debug("postrun signal task_id=%s", task_id)
165 
166         # retrieve and finish the Span
167         span, activation = utils.retrieve_span(task, task_id)
168         if span is None:
169             logger.warning("no existing span found for task_id=%s", task_id)
170             return
171 
172         # request context tags
173         if span.is_recording():
174             span.set_attribute(_TASK_TAG_KEY, _TASK_RUN)
175             utils.set_attributes_from_context(span, kwargs)
176             utils.set_attributes_from_context(span, task.request)
177             span.set_attribute(_TASK_NAME_KEY, task.name)
178 
179         activation.__exit__(None, None, None)
180         utils.detach_span(task, task_id)
181 
182     def _trace_before_publish(self, *args, **kwargs):
183         task = utils.retrieve_task_from_sender(kwargs)
184         task_id = utils.retrieve_task_id_from_message(kwargs)
185 
186         if task_id is None:
187             return
188 
189         if task is None:
190             # task is an anonymous task send using send_task or using canvas workflow
191             # Signatures() to send to a task not in the current processes dependency
192             # tree
193             task_name = kwargs.get("sender", "unknown")
194         else:
195             task_name = task.name
196         operation_name = f"{_TASK_APPLY_ASYNC}/{task_name}"
197         span = self._tracer.start_span(
198             operation_name, kind=trace.SpanKind.PRODUCER
199         )
200 
201         # apply some attributes here because most of the data is not available
202         if span.is_recording():
203             span.set_attribute(_TASK_TAG_KEY, _TASK_APPLY_ASYNC)
204             span.set_attribute(SpanAttributes.MESSAGING_MESSAGE_ID, task_id)
205             span.set_attribute(_TASK_NAME_KEY, task_name)
206             utils.set_attributes_from_context(span, kwargs)
207 
208         activation = trace.use_span(span, end_on_exit=True)
209         activation.__enter__()  # pylint: disable=E1101
210 
211         utils.attach_span(task, task_id, (span, activation), is_publish=True)
212 
213         headers = kwargs.get("headers")
214         if headers:
215             inject(headers)
216 
217     @staticmethod
218     def _trace_after_publish(*args, **kwargs):
219         task = utils.retrieve_task_from_sender(kwargs)
220         task_id = utils.retrieve_task_id_from_message(kwargs)
221 
222         if task is None or task_id is None:
223             return
224 
225         # retrieve and finish the Span
226         _, activation = utils.retrieve_span(task, task_id, is_publish=True)
227         if activation is None:
228             logger.warning("no existing span found for task_id=%s", task_id)
229             return
230 
231         activation.__exit__(None, None, None)  # pylint: disable=E1101
232         utils.detach_span(task, task_id, is_publish=True)
233 
234     @staticmethod
235     def _trace_failure(*args, **kwargs):
236         task = utils.retrieve_task_from_sender(kwargs)
237         task_id = utils.retrieve_task_id(kwargs)
238 
239         if task is None or task_id is None:
240             return
241 
242         # retrieve and pass exception info to activation
243         span, _ = utils.retrieve_span(task, task_id)
244         if span is None or not span.is_recording():
245             return
246 
247         status_kwargs = {"status_code": StatusCode.ERROR}
248 
249         ex = kwargs.get("einfo")
250 
251         if (
252             hasattr(task, "throws")
253             and ex is not None
254             and isinstance(ex.exception, task.throws)
255         ):
256             return
257 
258         if ex is not None:
259             status_kwargs["description"] = str(ex)
260             span.record_exception(ex)
261         span.set_status(Status(**status_kwargs))
262 
263     @staticmethod
264     def _trace_retry(*args, **kwargs):
265         task = utils.retrieve_task_from_sender(kwargs)
266         task_id = utils.retrieve_task_id_from_request(kwargs)
267         reason = utils.retrieve_reason(kwargs)
268 
269         if task is None or task_id is None or reason is None:
270             return
271 
272         span, _ = utils.retrieve_span(task, task_id)
273         if span is None or not span.is_recording():
274             return
275 
276         # Add retry reason metadata to span
277         # Use `str(reason)` instead of `reason.message` in case we get
278         # something that isn't an `Exception`
279         span.set_attribute(_TASK_RETRY_REASON_KEY, str(reason))
280
```
--- 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
@@ -60,6 +60,7 @@
 """
 
 import logging
+from timeit import default_timer
 from typing import Collection, Iterable
 
 from celery import signals  # pylint: disable=no-name-in-module
@@ -69,6 +70,7 @@
 from opentelemetry.instrumentation.celery.package import _instruments
 from opentelemetry.instrumentation.celery.version import __version__
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
+from opentelemetry.metrics import get_meter
 from opentelemetry.propagate import extract, inject
 from opentelemetry.propagators.textmap import Getter
 from opentelemetry.semconv.trace import SpanAttributes
@@ -104,6 +106,11 @@
 
 
 class CeleryInstrumentor(BaseInstrumentor):
+    def __init__(self):
+        super().__init__()
+        self.metrics = None
+        self.task_id_to_start_time = {}
+
     def instrumentation_dependencies(self) -> Collection[str]:
         return _instruments
 
@@ -113,6 +120,11 @@
         # pylint: disable=attribute-defined-outside-init
         self._tracer = trace.get_tracer(__name__, __version__, tracer_provider)
 
+        meter_provider = kwargs.get("meter_provider")
+        meter = get_meter(__name__, __version__, meter_provider)
+
+        self.create_celery_metrics(meter)
+
         signals.task_prerun.connect(self._trace_prerun, weak=False)
         signals.task_postrun.connect(self._trace_postrun, weak=False)
         signals.before_task_publish.connect(
@@ -139,6 +151,7 @@
         if task is None or task_id is None:
             return
 
+        self.update_task_duration_time(task_id)
         request = task.request
         tracectx = extract(request, getter=celery_getter) or None
 
@@ -153,8 +166,7 @@
         activation.__enter__()  # pylint: disable=E1101
         utils.attach_span(task, task_id, (span, activation))
 
-    @staticmethod
-    def _trace_postrun(*args, **kwargs):
+    def _trace_postrun(self, *args, **kwargs):
         task = utils.retrieve_task(kwargs)
         task_id = utils.retrieve_task_id(kwargs)
 
@@ -178,6 +190,9 @@
 
         activation.__exit__(None, None, None)
         utils.detach_span(task, task_id)
+        self.update_task_duration_time(task_id)
+        labels = {"task": task.name, "worker": task.request.hostname}
+        self._record_histograms(task_id, labels)
 
     def _trace_before_publish(self, *args, **kwargs):
         task = utils.retrieve_task_from_sender(kwargs)
@@ -277,3 +292,30 @@
         # 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 update_task_duration_time(self, task_id):
+        cur_time = default_timer()
+        task_duration_time_until_now = (
+            cur_time - self.task_id_to_start_time[task_id]
+            if task_id in self.task_id_to_start_time
+            else cur_time
+        )
+        self.task_id_to_start_time[task_id] = task_duration_time_until_now
+
+    def _record_histograms(self, task_id, metric_attributes):
+        if task_id is None:
+            return
+
+        self.metrics["flower.task.runtime.seconds"].record(
+            self.task_id_to_start_time.get(task_id),
+            attributes=metric_attributes,
+        )
+
+    def create_celery_metrics(self, meter) -> None:
+        self.metrics = {
+            "flower.task.runtime.seconds": meter.create_histogram(
+                name="flower.task.runtime.seconds",
+                unit="seconds",
+                description="The time it took to run the task.",
+            )
+        }

{"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@@ -60,6 +60,7 @@\n \"\"\"\n \n import logging\n+from timeit import default_timer\n from typing import Collection, Iterable\n \n from celery import signals  # pylint: disable=no-name-in-module\n@@ -69,6 +70,7 @@\n from opentelemetry.instrumentation.celery.package import _instruments\n from opentelemetry.instrumentation.celery.version import __version__\n from opentelemetry.instrumentation.instrumentor import BaseInstrumentor\n+from opentelemetry.metrics import get_meter\n from opentelemetry.propagate import extract, inject\n from opentelemetry.propagators.textmap import Getter\n from opentelemetry.semconv.trace import SpanAttributes\n@@ -104,6 +106,11 @@\n \n \n class CeleryInstrumentor(BaseInstrumentor):\n+    def __init__(self):\n+        super().__init__()\n+        self.metrics = None\n+        self.task_id_to_start_time = {}\n+\n     def instrumentation_dependencies(self) -> Collection[str]:\n         return _instruments\n \n@@ -113,6 +120,11 @@\n         # pylint: disable=attribute-defined-outside-init\n         self._tracer = trace.get_tracer(__name__, __version__, tracer_provider)\n \n+        meter_provider = kwargs.get(\"meter_provider\")\n+        meter = get_meter(__name__, __version__, meter_provider)\n+\n+        self.create_celery_metrics(meter)\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@@ -139,6 +151,7 @@\n         if task is None or task_id is None:\n             return\n \n+        self.update_task_duration_time(task_id)\n         request = task.request\n         tracectx = extract(request, getter=celery_getter) or None\n \n@@ -153,8 +166,7 @@\n         activation.__enter__()  # pylint: disable=E1101\n         utils.attach_span(task, task_id, (span, activation))\n \n-    @staticmethod\n-    def _trace_postrun(*args, **kwargs):\n+    def _trace_postrun(self, *args, **kwargs):\n         task = utils.retrieve_task(kwargs)\n         task_id = utils.retrieve_task_id(kwargs)\n \n@@ -178,6 +190,9 @@\n \n         activation.__exit__(None, None, None)\n         utils.detach_span(task, task_id)\n+        self.update_task_duration_time(task_id)\n+        labels = {\"task\": task.name, \"worker\": task.request.hostname}\n+        self._record_histograms(task_id, labels)\n \n     def _trace_before_publish(self, *args, **kwargs):\n         task = utils.retrieve_task_from_sender(kwargs)\n@@ -277,3 +292,30 @@\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+    def update_task_duration_time(self, task_id):\n+        cur_time = default_timer()\n+        task_duration_time_until_now = (\n+            cur_time - self.task_id_to_start_time[task_id]\n+            if task_id in self.task_id_to_start_time\n+            else cur_time\n+        )\n+        self.task_id_to_start_time[task_id] = task_duration_time_until_now\n+\n+    def _record_histograms(self, task_id, metric_attributes):\n+        if task_id is None:\n+            return\n+\n+        self.metrics[\"flower.task.runtime.seconds\"].record(\n+            self.task_id_to_start_time.get(task_id),\n+            attributes=metric_attributes,\n+        )\n+\n+    def create_celery_metrics(self, meter) -> None:\n+        self.metrics = {\n+            \"flower.task.runtime.seconds\": meter.create_histogram(\n+                name=\"flower.task.runtime.seconds\",\n+                unit=\"seconds\",\n+                description=\"The time it took to run the task.\",\n+            )\n+        }\n", "issue": "Metrics instrumentation celery\nRelevant semconv: https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/semantic_conventions/instrumentation/kafka.md\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    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        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\nSetting up tracing\n------------------\n\nWhen tracing a celery worker process, tracing and instrumention both must be initialized after the celery worker\nprocess is initialized. This is required for any tracing components that might use threading to work correctly\nsuch as the BatchSpanProcessor. Celery provides a signal called ``worker_process_init`` that can be used to\naccomplish this as shown in the example above.\n\nAPI\n---\n\"\"\"\n\nimport logging\nfrom typing import Collection, Iterable\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.package import _instruments\nfrom opentelemetry.instrumentation.celery.version import __version__\nfrom opentelemetry.instrumentation.instrumentor import BaseInstrumentor\nfrom opentelemetry.propagate import extract, inject\nfrom opentelemetry.propagators.textmap import Getter\nfrom opentelemetry.semconv.trace import SpanAttributes\nfrom opentelemetry.trace.status import Status, StatusCode\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\n\nclass CeleryGetter(Getter):\n    def get(self, carrier, key):\n        value = getattr(carrier, key, None)\n        if value is None:\n            return None\n        if isinstance(value, str) or not isinstance(value, Iterable):\n            value = (value,)\n        return value\n\n    def keys(self, carrier):\n        return []\n\n\ncelery_getter = CeleryGetter()\n\n\nclass CeleryInstrumentor(BaseInstrumentor):\n    def instrumentation_dependencies(self) -> Collection[str]:\n        return _instruments\n\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 = extract(request, getter=celery_getter) or None\n\n        logger.debug(\"prerun signal start task_id=%s\", task_id)\n\n        operation_name = f\"{_TASK_RUN}/{task.name}\"\n        span = self._tracer.start_span(\n            operation_name, context=tracectx, kind=trace.SpanKind.CONSUMER\n        )\n\n        activation = trace.use_span(span, end_on_exit=True)\n        activation.__enter__()  # pylint: disable=E1101\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        if span.is_recording():\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_id is None:\n            return\n\n        if task is None:\n            # task is an anonymous task send using send_task or using canvas workflow\n            # Signatures() to send to a task not in the current processes dependency\n            # tree\n            task_name = kwargs.get(\"sender\", \"unknown\")\n        else:\n            task_name = task.name\n        operation_name = f\"{_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        if span.is_recording():\n            span.set_attribute(_TASK_TAG_KEY, _TASK_APPLY_ASYNC)\n            span.set_attribute(SpanAttributes.MESSAGING_MESSAGE_ID, task_id)\n            span.set_attribute(_TASK_NAME_KEY, task_name)\n            utils.set_attributes_from_context(span, kwargs)\n\n        activation = trace.use_span(span, end_on_exit=True)\n        activation.__enter__()  # pylint: disable=E1101\n\n        utils.attach_span(task, task_id, (span, activation), is_publish=True)\n\n        headers = kwargs.get(\"headers\")\n        if headers:\n            inject(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)  # pylint: disable=E1101\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 or not span.is_recording():\n            return\n\n        status_kwargs = {\"status_code\": StatusCode.ERROR}\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            span.record_exception(ex)\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 or not span.is_recording():\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.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        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\nSetting up tracing\n------------------\n\nWhen tracing a celery worker process, tracing and instrumention both must be initialized after the celery worker\nprocess is initialized. This is required for any tracing components that might use threading to work correctly\nsuch as the BatchSpanProcessor. Celery provides a signal called ``worker_process_init`` that can be used to\naccomplish this as shown in the example above.\n\nAPI\n---\n\"\"\"\n\nimport logging\nfrom timeit import default_timer\nfrom typing import Collection, Iterable\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.package import _instruments\nfrom opentelemetry.instrumentation.celery.version import __version__\nfrom opentelemetry.instrumentation.instrumentor import BaseInstrumentor\nfrom opentelemetry.metrics import get_meter\nfrom opentelemetry.propagate import extract, inject\nfrom opentelemetry.propagators.textmap import Getter\nfrom opentelemetry.semconv.trace import SpanAttributes\nfrom opentelemetry.trace.status import Status, StatusCode\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\n\nclass CeleryGetter(Getter):\n    def get(self, carrier, key):\n        value = getattr(carrier, key, None)\n        if value is None:\n            return None\n        if isinstance(value, str) or not isinstance(value, Iterable):\n            value = (value,)\n        return value\n\n    def keys(self, carrier):\n        return []\n\n\ncelery_getter = CeleryGetter()\n\n\nclass CeleryInstrumentor(BaseInstrumentor):\n    def __init__(self):\n        super().__init__()\n        self.metrics = None\n        self.task_id_to_start_time = {}\n\n    def instrumentation_dependencies(self) -> Collection[str]:\n        return _instruments\n\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        meter_provider = kwargs.get(\"meter_provider\")\n        meter = get_meter(__name__, __version__, meter_provider)\n\n        self.create_celery_metrics(meter)\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        self.update_task_duration_time(task_id)\n        request = task.request\n        tracectx = extract(request, getter=celery_getter) or None\n\n        logger.debug(\"prerun signal start task_id=%s\", task_id)\n\n        operation_name = f\"{_TASK_RUN}/{task.name}\"\n        span = self._tracer.start_span(\n            operation_name, context=tracectx, kind=trace.SpanKind.CONSUMER\n        )\n\n        activation = trace.use_span(span, end_on_exit=True)\n        activation.__enter__()  # pylint: disable=E1101\n        utils.attach_span(task, task_id, (span, activation))\n\n    def _trace_postrun(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(\"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        if span.is_recording():\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        self.update_task_duration_time(task_id)\n        labels = {\"task\": task.name, \"worker\": task.request.hostname}\n        self._record_histograms(task_id, labels)\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_id is None:\n            return\n\n        if task is None:\n            # task is an anonymous task send using send_task or using canvas workflow\n            # Signatures() to send to a task not in the current processes dependency\n            # tree\n            task_name = kwargs.get(\"sender\", \"unknown\")\n        else:\n            task_name = task.name\n        operation_name = f\"{_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        if span.is_recording():\n            span.set_attribute(_TASK_TAG_KEY, _TASK_APPLY_ASYNC)\n            span.set_attribute(SpanAttributes.MESSAGING_MESSAGE_ID, task_id)\n            span.set_attribute(_TASK_NAME_KEY, task_name)\n            utils.set_attributes_from_context(span, kwargs)\n\n        activation = trace.use_span(span, end_on_exit=True)\n        activation.__enter__()  # pylint: disable=E1101\n\n        utils.attach_span(task, task_id, (span, activation), is_publish=True)\n\n        headers = kwargs.get(\"headers\")\n        if headers:\n            inject(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)  # pylint: disable=E1101\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 or not span.is_recording():\n            return\n\n        status_kwargs = {\"status_code\": StatusCode.ERROR}\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            span.record_exception(ex)\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 or not span.is_recording():\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    def update_task_duration_time(self, task_id):\n        cur_time = default_timer()\n        task_duration_time_until_now = (\n            cur_time - self.task_id_to_start_time[task_id]\n            if task_id in self.task_id_to_start_time\n            else cur_time\n        )\n        self.task_id_to_start_time[task_id] = task_duration_time_until_now\n\n    def _record_histograms(self, task_id, metric_attributes):\n        if task_id is None:\n            return\n\n        self.metrics[\"flower.task.runtime.seconds\"].record(\n            self.task_id_to_start_time.get(task_id),\n            attributes=metric_attributes,\n        )\n\n    def create_celery_metrics(self, meter) -> None:\n        self.metrics = {\n            \"flower.task.runtime.seconds\": meter.create_histogram(\n                name=\"flower.task.runtime.seconds\",\n                unit=\"seconds\",\n                description=\"The time it took to run the task.\",\n            )\n        }\n", "path": "instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/__init__.py"}]}