We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Partial celery task time limit is not support
Python 3.8

**Steps to reproduce**
- Create celery task with only one of the time limit soft/hard
Or use this test to reproduce:

`instrumentation/opentelemetry-instrumentation-celery/tests/test_utils.py`

```python
    def test_set_attributes_partial_timelimit_hard_limit(self):
        context = {
            "correlation_id": "44b7f305",
            "delivery_info": {"eager": True},
            "eta": "soon",
            "expires": "later",
            "hostname": "localhost",
            "id": "44b7f305",
            "reply_to": "44b7f305",
            "retries": 4,
            "timelimit": ("now", None),
            "custom_meta": "custom_value",
            "routing_key": "celery",
        }
        span = trace._Span("name", mock.Mock(spec=trace_api.SpanContext))
        utils.set_attributes_from_context(span, context)
        self.assertEqual(span.attributes.get("celery.timelimit"), ("now", ""))
```
**What is the expected behavior?**
The time limit that was specify should be instrumented 

**What is the actual behavior?**
Exception is raised and no time limit is instrumented
```
[__init__.py:_translate_attributes:164] _translate_key_values(key, value)
exception.trace.1	
[exporter.py:_translate_key_values:126] return KeyValue(key=key, value=_translate_value(value))
exception.trace.2	
[ exporter.py:_translate_value:104] array_value=ArrayValue(values=[_translate_value(v) for v in value])
exception.trace.3	
[ exporter.py:<listcomp>:104] array_value=ArrayValue(values=[_translate_value(v) for v in value])
exception.trace.4	
[ exporter.py:_translate_value:119] "Invalid type {} of value {}".format(type(value), value)
```
 


--- 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/utils.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 import logging
16 
17 from celery import registry  # pylint: disable=no-name-in-module
18 
19 from opentelemetry.semconv.trace import SpanAttributes
20 
21 logger = logging.getLogger(__name__)
22 
23 # Celery Context key
24 CTX_KEY = "__otel_task_span"
25 
26 # Celery Context attributes
27 CELERY_CONTEXT_ATTRIBUTES = (
28     "compression",
29     "correlation_id",
30     "countdown",
31     "delivery_info",
32     "declare",
33     "eta",
34     "exchange",
35     "expires",
36     "hostname",
37     "id",
38     "priority",
39     "queue",
40     "reply_to",
41     "retries",
42     "routing_key",
43     "serializer",
44     "timelimit",
45     "origin",
46     "state",
47 )
48 
49 
50 # pylint:disable=too-many-branches
51 def set_attributes_from_context(span, context):
52     """Helper to extract meta values from a Celery Context"""
53     if not span.is_recording():
54         return
55     for key in CELERY_CONTEXT_ATTRIBUTES:
56         value = context.get(key)
57 
58         # Skip this key if it is not set
59         if value is None or value == "":
60             continue
61 
62         # Skip `timelimit` if it is not set (it's default/unset value is a
63         # tuple or a list of `None` values
64         if key == "timelimit" and value in [(None, None), [None, None]]:
65             continue
66 
67         # Skip `retries` if it's value is `0`
68         if key == "retries" and value == 0:
69             continue
70 
71         attribute_name = None
72 
73         # Celery 4.0 uses `origin` instead of `hostname`; this change preserves
74         # the same name for the tag despite Celery version
75         if key == "origin":
76             key = "hostname"
77 
78         elif key == "delivery_info":
79             # Get also destination from this
80             routing_key = value.get("routing_key")
81             if routing_key is not None:
82                 span.set_attribute(
83                     SpanAttributes.MESSAGING_DESTINATION, routing_key
84                 )
85             value = str(value)
86 
87         elif key == "id":
88             attribute_name = SpanAttributes.MESSAGING_MESSAGE_ID
89 
90         elif key == "correlation_id":
91             attribute_name = SpanAttributes.MESSAGING_CONVERSATION_ID
92 
93         elif key == "routing_key":
94             attribute_name = SpanAttributes.MESSAGING_DESTINATION
95 
96         # according to https://docs.celeryproject.org/en/stable/userguide/routing.html#exchange-types
97         elif key == "declare":
98             attribute_name = SpanAttributes.MESSAGING_DESTINATION_KIND
99             for declare in value:
100                 if declare.exchange.type == "direct":
101                     value = "queue"
102                     break
103                 if declare.exchange.type == "topic":
104                     value = "topic"
105                     break
106 
107         # set attribute name if not set specially for a key
108         if attribute_name is None:
109             attribute_name = f"celery.{key}"
110 
111         span.set_attribute(attribute_name, value)
112 
113 
114 def attach_span(task, task_id, span, is_publish=False):
115     """Helper to propagate a `Span` for the given `Task` instance. This
116     function uses a `dict` that stores the Span using the
117     `(task_id, is_publish)` as a key. This is useful when information must be
118     propagated from one Celery signal to another.
119 
120     We use (task_id, is_publish) for the key to ensure that publishing a
121     task from within another task does not cause any conflicts.
122 
123     This mostly happens when either a task fails and a retry policy is in place,
124     or when a task is manually retries (e.g. `task.retry()`), we end up trying
125     to publish a task with the same id as the task currently running.
126 
127     Previously publishing the new task would overwrite the existing `celery.run` span
128     in the `dict` causing that span to be forgotten and never finished
129     NOTE: We cannot test for this well yet, because we do not run a celery worker,
130     and cannot run `task.apply_async()`
131     """
132     span_dict = getattr(task, CTX_KEY, None)
133     if span_dict is None:
134         span_dict = {}
135         setattr(task, CTX_KEY, span_dict)
136 
137     span_dict[(task_id, is_publish)] = span
138 
139 
140 def detach_span(task, task_id, is_publish=False):
141     """Helper to remove a `Span` in a Celery task when it's propagated.
142     This function handles tasks where the `Span` is not attached.
143     """
144     span_dict = getattr(task, CTX_KEY, None)
145     if span_dict is None:
146         return
147 
148     # See note in `attach_span` for key info
149     span_dict.pop((task_id, is_publish), (None, None))
150 
151 
152 def retrieve_span(task, task_id, is_publish=False):
153     """Helper to retrieve an active `Span` stored in a `Task`
154     instance
155     """
156     span_dict = getattr(task, CTX_KEY, None)
157     if span_dict is None:
158         return (None, None)
159 
160     # See note in `attach_span` for key info
161     return span_dict.get((task_id, is_publish), (None, None))
162 
163 
164 def retrieve_task(kwargs):
165     task = kwargs.get("task")
166     if task is None:
167         logger.debug("Unable to retrieve task from signal arguments")
168     return task
169 
170 
171 def retrieve_task_from_sender(kwargs):
172     sender = kwargs.get("sender")
173     if sender is None:
174         logger.debug("Unable to retrieve the sender from signal arguments")
175 
176     # before and after publish signals sender is the task name
177     # for retry and failure signals sender is the task object
178     if isinstance(sender, str):
179         sender = registry.tasks.get(sender)
180         if sender is None:
181             logger.debug("Unable to retrieve the task from sender=%s", sender)
182 
183     return sender
184 
185 
186 def retrieve_task_id(kwargs):
187     task_id = kwargs.get("task_id")
188     if task_id is None:
189         logger.debug("Unable to retrieve task_id from signal arguments")
190     return task_id
191 
192 
193 def retrieve_task_id_from_request(kwargs):
194     # retry signal does not include task_id as argument so use request argument
195     request = kwargs.get("request")
196     if request is None:
197         logger.debug("Unable to retrieve the request from signal arguments")
198 
199     task_id = getattr(request, "id")
200     if task_id is None:
201         logger.debug("Unable to retrieve the task_id from the request")
202 
203     return task_id
204 
205 
206 def retrieve_task_id_from_message(kwargs):
207     """Helper to retrieve the `Task` identifier from the message `body`.
208     This helper supports Protocol Version 1 and 2. The Protocol is well
209     detailed in the official documentation:
210     http://docs.celeryproject.org/en/latest/internals/protocol.html
211     """
212     headers = kwargs.get("headers")
213     body = kwargs.get("body")
214     if headers is not None and len(headers) > 0:
215         # Protocol Version 2 (default from Celery 4.0)
216         return headers.get("id")
217     # Protocol Version 1
218     return body.get("id")
219 
220 
221 def retrieve_reason(kwargs):
222     reason = kwargs.get("reason")
223     if not reason:
224         logger.debug("Unable to retrieve the retry reason")
225     return reason
226
```
--- 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/utils.py b/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/utils.py
--- a/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/utils.py
+++ b/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/utils.py
@@ -61,8 +61,11 @@
 
         # Skip `timelimit` if it is not set (it's default/unset value is a
         # tuple or a list of `None` values
-        if key == "timelimit" and value in [(None, None), [None, None]]:
-            continue
+        if key == "timelimit":
+            if value in [(None, None), [None, None]]:
+                continue
+            if None in value:
+                value = ["" if tl is None else tl for tl in value]
 
         # Skip `retries` if it's value is `0`
         if key == "retries" and value == 0:

{"golden_diff": "diff --git a/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/utils.py b/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/utils.py\n--- a/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/utils.py\n+++ b/instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/utils.py\n@@ -61,8 +61,11 @@\n \n         # Skip `timelimit` if it is not set (it's default/unset value is a\n         # tuple or a list of `None` values\n-        if key == \"timelimit\" and value in [(None, None), [None, None]]:\n-            continue\n+        if key == \"timelimit\":\n+            if value in [(None, None), [None, None]]:\n+                continue\n+            if None in value:\n+                value = [\"\" if tl is None else tl for tl in value]\n \n         # Skip `retries` if it's value is `0`\n         if key == \"retries\" and value == 0:\n", "issue": "Partial celery task time limit is not support\nPython 3.8\r\n\r\n**Steps to reproduce**\r\n- Create celery task with only one of the time limit soft/hard\r\nOr use this test to reproduce:\r\n\r\n`instrumentation/opentelemetry-instrumentation-celery/tests/test_utils.py`\r\n\r\n```python\r\n    def test_set_attributes_partial_timelimit_hard_limit(self):\r\n        context = {\r\n            \"correlation_id\": \"44b7f305\",\r\n            \"delivery_info\": {\"eager\": True},\r\n            \"eta\": \"soon\",\r\n            \"expires\": \"later\",\r\n            \"hostname\": \"localhost\",\r\n            \"id\": \"44b7f305\",\r\n            \"reply_to\": \"44b7f305\",\r\n            \"retries\": 4,\r\n            \"timelimit\": (\"now\", None),\r\n            \"custom_meta\": \"custom_value\",\r\n            \"routing_key\": \"celery\",\r\n        }\r\n        span = trace._Span(\"name\", mock.Mock(spec=trace_api.SpanContext))\r\n        utils.set_attributes_from_context(span, context)\r\n        self.assertEqual(span.attributes.get(\"celery.timelimit\"), (\"now\", \"\"))\r\n```\r\n**What is the expected behavior?**\r\nThe time limit that was specify should be instrumented \r\n\r\n**What is the actual behavior?**\r\nException is raised and no time limit is instrumented\r\n```\r\n[__init__.py:_translate_attributes:164] _translate_key_values(key, value)\r\nexception.trace.1\t\r\n[exporter.py:_translate_key_values:126] return KeyValue(key=key, value=_translate_value(value))\r\nexception.trace.2\t\r\n[ exporter.py:_translate_value:104] array_value=ArrayValue(values=[_translate_value(v) for v in value])\r\nexception.trace.3\t\r\n[ exporter.py:<listcomp>:104] array_value=ArrayValue(values=[_translate_value(v) for v in value])\r\nexception.trace.4\t\r\n[ exporter.py:_translate_value:119] \"Invalid type {} of value {}\".format(type(value), value)\r\n```\r\n \r\n\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\nimport logging\n\nfrom celery import registry  # pylint: disable=no-name-in-module\n\nfrom opentelemetry.semconv.trace import SpanAttributes\n\nlogger = logging.getLogger(__name__)\n\n# Celery Context key\nCTX_KEY = \"__otel_task_span\"\n\n# Celery Context attributes\nCELERY_CONTEXT_ATTRIBUTES = (\n    \"compression\",\n    \"correlation_id\",\n    \"countdown\",\n    \"delivery_info\",\n    \"declare\",\n    \"eta\",\n    \"exchange\",\n    \"expires\",\n    \"hostname\",\n    \"id\",\n    \"priority\",\n    \"queue\",\n    \"reply_to\",\n    \"retries\",\n    \"routing_key\",\n    \"serializer\",\n    \"timelimit\",\n    \"origin\",\n    \"state\",\n)\n\n\n# pylint:disable=too-many-branches\ndef set_attributes_from_context(span, context):\n    \"\"\"Helper to extract meta values from a Celery Context\"\"\"\n    if not span.is_recording():\n        return\n    for key in CELERY_CONTEXT_ATTRIBUTES:\n        value = context.get(key)\n\n        # Skip this key if it is not set\n        if value is None or value == \"\":\n            continue\n\n        # Skip `timelimit` if it is not set (it's default/unset value is a\n        # tuple or a list of `None` values\n        if key == \"timelimit\" and value in [(None, None), [None, None]]:\n            continue\n\n        # Skip `retries` if it's value is `0`\n        if key == \"retries\" and value == 0:\n            continue\n\n        attribute_name = None\n\n        # Celery 4.0 uses `origin` instead of `hostname`; this change preserves\n        # the same name for the tag despite Celery version\n        if key == \"origin\":\n            key = \"hostname\"\n\n        elif key == \"delivery_info\":\n            # Get also destination from this\n            routing_key = value.get(\"routing_key\")\n            if routing_key is not None:\n                span.set_attribute(\n                    SpanAttributes.MESSAGING_DESTINATION, routing_key\n                )\n            value = str(value)\n\n        elif key == \"id\":\n            attribute_name = SpanAttributes.MESSAGING_MESSAGE_ID\n\n        elif key == \"correlation_id\":\n            attribute_name = SpanAttributes.MESSAGING_CONVERSATION_ID\n\n        elif key == \"routing_key\":\n            attribute_name = SpanAttributes.MESSAGING_DESTINATION\n\n        # according to https://docs.celeryproject.org/en/stable/userguide/routing.html#exchange-types\n        elif key == \"declare\":\n            attribute_name = SpanAttributes.MESSAGING_DESTINATION_KIND\n            for declare in value:\n                if declare.exchange.type == \"direct\":\n                    value = \"queue\"\n                    break\n                if declare.exchange.type == \"topic\":\n                    value = \"topic\"\n                    break\n\n        # set attribute name if not set specially for a key\n        if attribute_name is None:\n            attribute_name = f\"celery.{key}\"\n\n        span.set_attribute(attribute_name, value)\n\n\ndef attach_span(task, task_id, span, is_publish=False):\n    \"\"\"Helper to propagate a `Span` for the given `Task` instance. This\n    function uses a `dict` that stores the Span using the\n    `(task_id, is_publish)` as a key. This is useful when information must be\n    propagated from one Celery signal to another.\n\n    We use (task_id, is_publish) for the key to ensure that publishing a\n    task from within another task does not cause any conflicts.\n\n    This mostly happens when either a task fails and a retry policy is in place,\n    or when a task is manually retries (e.g. `task.retry()`), we end up trying\n    to publish a task with the same id as the task currently running.\n\n    Previously publishing the new task would overwrite the existing `celery.run` span\n    in the `dict` causing that span to be forgotten and never finished\n    NOTE: We cannot test for this well yet, because we do not run a celery worker,\n    and cannot run `task.apply_async()`\n    \"\"\"\n    span_dict = getattr(task, CTX_KEY, None)\n    if span_dict is None:\n        span_dict = {}\n        setattr(task, CTX_KEY, span_dict)\n\n    span_dict[(task_id, is_publish)] = span\n\n\ndef detach_span(task, task_id, is_publish=False):\n    \"\"\"Helper to remove a `Span` in a Celery task when it's propagated.\n    This function handles tasks where the `Span` is not attached.\n    \"\"\"\n    span_dict = getattr(task, CTX_KEY, None)\n    if span_dict is None:\n        return\n\n    # See note in `attach_span` for key info\n    span_dict.pop((task_id, is_publish), (None, None))\n\n\ndef retrieve_span(task, task_id, is_publish=False):\n    \"\"\"Helper to retrieve an active `Span` stored in a `Task`\n    instance\n    \"\"\"\n    span_dict = getattr(task, CTX_KEY, None)\n    if span_dict is None:\n        return (None, None)\n\n    # See note in `attach_span` for key info\n    return span_dict.get((task_id, is_publish), (None, None))\n\n\ndef retrieve_task(kwargs):\n    task = kwargs.get(\"task\")\n    if task is None:\n        logger.debug(\"Unable to retrieve task from signal arguments\")\n    return task\n\n\ndef retrieve_task_from_sender(kwargs):\n    sender = kwargs.get(\"sender\")\n    if sender is None:\n        logger.debug(\"Unable to retrieve the sender from signal arguments\")\n\n    # before and after publish signals sender is the task name\n    # for retry and failure signals sender is the task object\n    if isinstance(sender, str):\n        sender = registry.tasks.get(sender)\n        if sender is None:\n            logger.debug(\"Unable to retrieve the task from sender=%s\", sender)\n\n    return sender\n\n\ndef retrieve_task_id(kwargs):\n    task_id = kwargs.get(\"task_id\")\n    if task_id is None:\n        logger.debug(\"Unable to retrieve task_id from signal arguments\")\n    return task_id\n\n\ndef retrieve_task_id_from_request(kwargs):\n    # retry signal does not include task_id as argument so use request argument\n    request = kwargs.get(\"request\")\n    if request is None:\n        logger.debug(\"Unable to retrieve the request from signal arguments\")\n\n    task_id = getattr(request, \"id\")\n    if task_id is None:\n        logger.debug(\"Unable to retrieve the task_id from the request\")\n\n    return task_id\n\n\ndef retrieve_task_id_from_message(kwargs):\n    \"\"\"Helper to retrieve the `Task` identifier from the message `body`.\n    This helper supports Protocol Version 1 and 2. The Protocol is well\n    detailed in the official documentation:\n    http://docs.celeryproject.org/en/latest/internals/protocol.html\n    \"\"\"\n    headers = kwargs.get(\"headers\")\n    body = kwargs.get(\"body\")\n    if headers is not None and len(headers) > 0:\n        # Protocol Version 2 (default from Celery 4.0)\n        return headers.get(\"id\")\n    # Protocol Version 1\n    return body.get(\"id\")\n\n\ndef retrieve_reason(kwargs):\n    reason = kwargs.get(\"reason\")\n    if not reason:\n        logger.debug(\"Unable to retrieve the retry reason\")\n    return reason\n", "path": "instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/utils.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\nimport logging\n\nfrom celery import registry  # pylint: disable=no-name-in-module\n\nfrom opentelemetry.semconv.trace import SpanAttributes\n\nlogger = logging.getLogger(__name__)\n\n# Celery Context key\nCTX_KEY = \"__otel_task_span\"\n\n# Celery Context attributes\nCELERY_CONTEXT_ATTRIBUTES = (\n    \"compression\",\n    \"correlation_id\",\n    \"countdown\",\n    \"delivery_info\",\n    \"declare\",\n    \"eta\",\n    \"exchange\",\n    \"expires\",\n    \"hostname\",\n    \"id\",\n    \"priority\",\n    \"queue\",\n    \"reply_to\",\n    \"retries\",\n    \"routing_key\",\n    \"serializer\",\n    \"timelimit\",\n    \"origin\",\n    \"state\",\n)\n\n\n# pylint:disable=too-many-branches\ndef set_attributes_from_context(span, context):\n    \"\"\"Helper to extract meta values from a Celery Context\"\"\"\n    if not span.is_recording():\n        return\n    for key in CELERY_CONTEXT_ATTRIBUTES:\n        value = context.get(key)\n\n        # Skip this key if it is not set\n        if value is None or value == \"\":\n            continue\n\n        # Skip `timelimit` if it is not set (it's default/unset value is a\n        # tuple or a list of `None` values\n        if key == \"timelimit\":\n            if value in [(None, None), [None, None]]:\n                continue\n            if None in value:\n                value = [\"\" if tl is None else tl for tl in value]\n\n        # Skip `retries` if it's value is `0`\n        if key == \"retries\" and value == 0:\n            continue\n\n        attribute_name = None\n\n        # Celery 4.0 uses `origin` instead of `hostname`; this change preserves\n        # the same name for the tag despite Celery version\n        if key == \"origin\":\n            key = \"hostname\"\n\n        elif key == \"delivery_info\":\n            # Get also destination from this\n            routing_key = value.get(\"routing_key\")\n            if routing_key is not None:\n                span.set_attribute(\n                    SpanAttributes.MESSAGING_DESTINATION, routing_key\n                )\n            value = str(value)\n\n        elif key == \"id\":\n            attribute_name = SpanAttributes.MESSAGING_MESSAGE_ID\n\n        elif key == \"correlation_id\":\n            attribute_name = SpanAttributes.MESSAGING_CONVERSATION_ID\n\n        elif key == \"routing_key\":\n            attribute_name = SpanAttributes.MESSAGING_DESTINATION\n\n        # according to https://docs.celeryproject.org/en/stable/userguide/routing.html#exchange-types\n        elif key == \"declare\":\n            attribute_name = SpanAttributes.MESSAGING_DESTINATION_KIND\n            for declare in value:\n                if declare.exchange.type == \"direct\":\n                    value = \"queue\"\n                    break\n                if declare.exchange.type == \"topic\":\n                    value = \"topic\"\n                    break\n\n        # set attribute name if not set specially for a key\n        if attribute_name is None:\n            attribute_name = f\"celery.{key}\"\n\n        span.set_attribute(attribute_name, value)\n\n\ndef attach_span(task, task_id, span, is_publish=False):\n    \"\"\"Helper to propagate a `Span` for the given `Task` instance. This\n    function uses a `dict` that stores the Span using the\n    `(task_id, is_publish)` as a key. This is useful when information must be\n    propagated from one Celery signal to another.\n\n    We use (task_id, is_publish) for the key to ensure that publishing a\n    task from within another task does not cause any conflicts.\n\n    This mostly happens when either a task fails and a retry policy is in place,\n    or when a task is manually retries (e.g. `task.retry()`), we end up trying\n    to publish a task with the same id as the task currently running.\n\n    Previously publishing the new task would overwrite the existing `celery.run` span\n    in the `dict` causing that span to be forgotten and never finished\n    NOTE: We cannot test for this well yet, because we do not run a celery worker,\n    and cannot run `task.apply_async()`\n    \"\"\"\n    span_dict = getattr(task, CTX_KEY, None)\n    if span_dict is None:\n        span_dict = {}\n        setattr(task, CTX_KEY, span_dict)\n\n    span_dict[(task_id, is_publish)] = span\n\n\ndef detach_span(task, task_id, is_publish=False):\n    \"\"\"Helper to remove a `Span` in a Celery task when it's propagated.\n    This function handles tasks where the `Span` is not attached.\n    \"\"\"\n    span_dict = getattr(task, CTX_KEY, None)\n    if span_dict is None:\n        return\n\n    # See note in `attach_span` for key info\n    span_dict.pop((task_id, is_publish), (None, None))\n\n\ndef retrieve_span(task, task_id, is_publish=False):\n    \"\"\"Helper to retrieve an active `Span` stored in a `Task`\n    instance\n    \"\"\"\n    span_dict = getattr(task, CTX_KEY, None)\n    if span_dict is None:\n        return (None, None)\n\n    # See note in `attach_span` for key info\n    return span_dict.get((task_id, is_publish), (None, None))\n\n\ndef retrieve_task(kwargs):\n    task = kwargs.get(\"task\")\n    if task is None:\n        logger.debug(\"Unable to retrieve task from signal arguments\")\n    return task\n\n\ndef retrieve_task_from_sender(kwargs):\n    sender = kwargs.get(\"sender\")\n    if sender is None:\n        logger.debug(\"Unable to retrieve the sender from signal arguments\")\n\n    # before and after publish signals sender is the task name\n    # for retry and failure signals sender is the task object\n    if isinstance(sender, str):\n        sender = registry.tasks.get(sender)\n        if sender is None:\n            logger.debug(\"Unable to retrieve the task from sender=%s\", sender)\n\n    return sender\n\n\ndef retrieve_task_id(kwargs):\n    task_id = kwargs.get(\"task_id\")\n    if task_id is None:\n        logger.debug(\"Unable to retrieve task_id from signal arguments\")\n    return task_id\n\n\ndef retrieve_task_id_from_request(kwargs):\n    # retry signal does not include task_id as argument so use request argument\n    request = kwargs.get(\"request\")\n    if request is None:\n        logger.debug(\"Unable to retrieve the request from signal arguments\")\n\n    task_id = getattr(request, \"id\")\n    if task_id is None:\n        logger.debug(\"Unable to retrieve the task_id from the request\")\n\n    return task_id\n\n\ndef retrieve_task_id_from_message(kwargs):\n    \"\"\"Helper to retrieve the `Task` identifier from the message `body`.\n    This helper supports Protocol Version 1 and 2. The Protocol is well\n    detailed in the official documentation:\n    http://docs.celeryproject.org/en/latest/internals/protocol.html\n    \"\"\"\n    headers = kwargs.get(\"headers\")\n    body = kwargs.get(\"body\")\n    if headers is not None and len(headers) > 0:\n        # Protocol Version 2 (default from Celery 4.0)\n        return headers.get(\"id\")\n    # Protocol Version 1\n    return body.get(\"id\")\n\n\ndef retrieve_reason(kwargs):\n    reason = kwargs.get(\"reason\")\n    if not reason:\n        logger.debug(\"Unable to retrieve the retry reason\")\n    return reason\n", "path": "instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/utils.py"}]}