prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  Elasticsearch sanitization does not work for bulk queries
  **Describe your environment**
  
  Discovered in elasticsearch 5.5.3 and elasticsearch-dsl 5.4.0 and caused by moving to the default sanitization in #1758.
  
  The issue is illustrated here where `body` comes in as a string, not as a dictionary:
  
  <img width="1355" alt="image" src="https://github.com/open-telemetry/opentelemetry-python-contrib/assets/684275/0f49ce7d-9558-4148-a2fd-28dc4bd70983">
  
  This is caseud by the bulk flow specifically as the body gets translated to a string here:
  
  <img width="1174" alt="image" src="https://github.com/open-telemetry/opentelemetry-python-contrib/assets/684275/2106e7aa-2bde-4579-b562-a8dc23007a1c">
  
  which looks like this:
  
  <img width="1144" alt="image" src="https://github.com/open-telemetry/opentelemetry-python-contrib/assets/684275/ae72d60a-5047-4559-ad05-6fc1ddb2658d">
  
  
  **Steps to reproduce**
  
  I don't have a super straightforward way to reproduce other than to use the bulk API from elasticsearch.
  
  
  **What is the expected behavior?**
  What did you expect to see?
  
  **What is the actual behavior?**
  
  The below stacktrace:
  
  ```
    File "/Users/phillip/Library/Caches/pypoetry/virtualenvs/someenv/lib/python3.11/site-packages/elasticsearch/helpers/__init__.py", line 95, in _process_bulk_chunk
      resp = client.bulk('\n'.join(bulk_actions) + '\n', **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    File "/Users/phillip/Library/Caches/pypoetry/virtualenvs/someenv/lib/python3.11/site-packages/elasticsearch/client/utils.py", line 73, in _wrapped
      return func(*args, params=params, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    File "/Users/phillip/Library/Caches/pypoetry/virtualenvs/someenv/lib/python3.11/site-packages/elasticsearch/client/__init__.py", line 1173, in bulk
      return self.transport.perform_request('POST', _make_path(index,
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    File "/Users/phillip/Library/Caches/pypoetry/virtualenvs/someenv/lib/python3.11/site-packages/opentelemetry/instrumentation/elasticsearch/__init__.py", line 224, in wrapper
      attributes[SpanAttributes.DB_STATEMENT] = sanitize_body(
                                                ^^^^^^^^^^^^^^
    File "/Users/phillip/Library/Caches/pypoetry/virtualenvs/someenv/lib/python3.11/site-packages/opentelemetry/instrumentation/elasticsearch/utils.py", line 54, in sanitize_body
      flatten_body = _flatten_dict(body)
                     ^^^^^^^^^^^^^^^^^^^
    File "/Users/phillip/Library/Caches/pypoetry/virtualenvs/someenv/lib/python3.11/site-packages/opentelemetry/instrumentation/elasticsearch/utils.py", line 30, in _flatten_dict
      for k, v in d.items():
                  ^^^^^^^
  AttributeError: 'str' object has no attribute 'items'
  
  ```
  
  **Additional context**
  Add any other context about the problem here.
  
  
  --- 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-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/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 sanitized_keys = (
  16     "message",
  17     "should",
  18     "filter",
  19     "query",
  20     "queries",
  21     "intervals",
  22     "match",
  23 )
  24 sanitized_value = "?"
  25 
  26 
  27 # pylint: disable=C0103
  28 def _flatten_dict(d, parent_key=""):
  29     items = []
  30     for k, v in d.items():
  31         new_key = parent_key + "." + k if parent_key else k
  32         # recursive call _flatten_dict for a non-empty dict value
  33         if isinstance(v, dict) and v:
  34             items.extend(_flatten_dict(v, new_key).items())
  35         else:
  36             items.append((new_key, v))
  37     return dict(items)
  38 
  39 
  40 def _unflatten_dict(d):
  41     res = {}
  42     for k, v in d.items():
  43         keys = k.split(".")
  44         d = res
  45         for key in keys[:-1]:
  46             if key not in d:
  47                 d[key] = {}
  48             d = d[key]
  49         d[keys[-1]] = v
  50     return res
  51 
  52 
  53 def sanitize_body(body) -> str:
  54     flatten_body = _flatten_dict(body)
  55 
  56     for key in flatten_body:
  57         if key.endswith(sanitized_keys):
  58             flatten_body[key] = sanitized_value
  59 
  60     return str(_unflatten_dict(flatten_body))
  61
  ```
  --- 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()
  
  ```
  
golden_diff: diff --git a/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/utils.py b/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/utils.py
  --- a/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/utils.py
  +++ b/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/utils.py
  @@ -11,6 +11,7 @@
   # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   # See the License for the specific language governing permissions and
   # limitations under the License.
  +import json
   
   sanitized_keys = (
       "message",
  @@ -51,6 +52,9 @@
   
   
   def sanitize_body(body) -> str:
  +    if isinstance(body, str):
  +        body = json.loads(body)
  +
       flatten_body = _flatten_dict(body)
   
       for key in flatten_body:
  
verification_info: {"golden_diff": "diff --git a/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/utils.py b/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/utils.py\n--- a/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/utils.py\n+++ b/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/utils.py\n@@ -11,6 +11,7 @@\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+import json\n \n sanitized_keys = (\n     \"message\",\n@@ -51,6 +52,9 @@\n \n \n def sanitize_body(body) -> str:\n+    if isinstance(body, str):\n+        body = json.loads(body)\n+\n     flatten_body = _flatten_dict(body)\n \n     for key in flatten_body:\n", "issue": "Elasticsearch sanitization does not work for bulk queries\n**Describe your environment**\r\n\r\nDiscovered in elasticsearch 5.5.3 and elasticsearch-dsl 5.4.0 and caused by moving to the default sanitization in #1758.\r\n\r\nThe issue is illustrated here where `body` comes in as a string, not as a dictionary:\r\n\r\n<img width=\"1355\" alt=\"image\" src=\"https://github.com/open-telemetry/opentelemetry-python-contrib/assets/684275/0f49ce7d-9558-4148-a2fd-28dc4bd70983\">\r\n\r\nThis is caseud by the bulk flow specifically as the body gets translated to a string here:\r\n\r\n<img width=\"1174\" alt=\"image\" src=\"https://github.com/open-telemetry/opentelemetry-python-contrib/assets/684275/2106e7aa-2bde-4579-b562-a8dc23007a1c\">\r\n\r\nwhich looks like this:\r\n\r\n<img width=\"1144\" alt=\"image\" src=\"https://github.com/open-telemetry/opentelemetry-python-contrib/assets/684275/ae72d60a-5047-4559-ad05-6fc1ddb2658d\">\r\n\r\n\r\n**Steps to reproduce**\r\n\r\nI don't have a super straightforward way to reproduce other than to use the bulk API from elasticsearch.\r\n\r\n\r\n**What is the expected behavior?**\r\nWhat did you expect to see?\r\n\r\n**What is the actual behavior?**\r\n\r\nThe below stacktrace:\r\n\r\n```\r\n  File \"/Users/phillip/Library/Caches/pypoetry/virtualenvs/someenv/lib/python3.11/site-packages/elasticsearch/helpers/__init__.py\", line 95, in _process_bulk_chunk\r\n    resp = client.bulk('\\n'.join(bulk_actions) + '\\n', **kwargs)\r\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/Users/phillip/Library/Caches/pypoetry/virtualenvs/someenv/lib/python3.11/site-packages/elasticsearch/client/utils.py\", line 73, in _wrapped\r\n    return func(*args, params=params, **kwargs)\r\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/Users/phillip/Library/Caches/pypoetry/virtualenvs/someenv/lib/python3.11/site-packages/elasticsearch/client/__init__.py\", line 1173, in bulk\r\n    return self.transport.perform_request('POST', _make_path(index,\r\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\r\n  File \"/Users/phillip/Library/Caches/pypoetry/virtualenvs/someenv/lib/python3.11/site-packages/opentelemetry/instrumentation/elasticsearch/__init__.py\", line 224, in wrapper\r\n    attributes[SpanAttributes.DB_STATEMENT] = sanitize_body(\r\n                                              ^^^^^^^^^^^^^^\r\n  File \"/Users/phillip/Library/Caches/pypoetry/virtualenvs/someenv/lib/python3.11/site-packages/opentelemetry/instrumentation/elasticsearch/utils.py\", line 54, in sanitize_body\r\n    flatten_body = _flatten_dict(body)\r\n                   ^^^^^^^^^^^^^^^^^^^\r\n  File \"/Users/phillip/Library/Caches/pypoetry/virtualenvs/someenv/lib/python3.11/site-packages/opentelemetry/instrumentation/elasticsearch/utils.py\", line 30, in _flatten_dict\r\n    for k, v in d.items():\r\n                ^^^^^^^\r\nAttributeError: 'str' object has no attribute 'items'\r\n\r\n```\r\n\r\n**Additional context**\r\nAdd any other context about the problem here.\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\nsanitized_keys = (\n    \"message\",\n    \"should\",\n    \"filter\",\n    \"query\",\n    \"queries\",\n    \"intervals\",\n    \"match\",\n)\nsanitized_value = \"?\"\n\n\n# pylint: disable=C0103\ndef _flatten_dict(d, parent_key=\"\"):\n    items = []\n    for k, v in d.items():\n        new_key = parent_key + \".\" + k if parent_key else k\n        # recursive call _flatten_dict for a non-empty dict value\n        if isinstance(v, dict) and v:\n            items.extend(_flatten_dict(v, new_key).items())\n        else:\n            items.append((new_key, v))\n    return dict(items)\n\n\ndef _unflatten_dict(d):\n    res = {}\n    for k, v in d.items():\n        keys = k.split(\".\")\n        d = res\n        for key in keys[:-1]:\n            if key not in d:\n                d[key] = {}\n            d = d[key]\n        d[keys[-1]] = v\n    return res\n\n\ndef sanitize_body(body) -> str:\n    flatten_body = _flatten_dict(body)\n\n    for key in flatten_body:\n        if key.endswith(sanitized_keys):\n            flatten_body[key] = sanitized_value\n\n    return str(_unflatten_dict(flatten_body))\n", "path": "instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/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.\nimport json\n\nsanitized_keys = (\n    \"message\",\n    \"should\",\n    \"filter\",\n    \"query\",\n    \"queries\",\n    \"intervals\",\n    \"match\",\n)\nsanitized_value = \"?\"\n\n\n# pylint: disable=C0103\ndef _flatten_dict(d, parent_key=\"\"):\n    items = []\n    for k, v in d.items():\n        new_key = parent_key + \".\" + k if parent_key else k\n        # recursive call _flatten_dict for a non-empty dict value\n        if isinstance(v, dict) and v:\n            items.extend(_flatten_dict(v, new_key).items())\n        else:\n            items.append((new_key, v))\n    return dict(items)\n\n\ndef _unflatten_dict(d):\n    res = {}\n    for k, v in d.items():\n        keys = k.split(\".\")\n        d = res\n        for key in keys[:-1]:\n            if key not in d:\n                d[key] = {}\n            d = d[key]\n        d[keys[-1]] = v\n    return res\n\n\ndef sanitize_body(body) -> str:\n    if isinstance(body, str):\n        body = json.loads(body)\n\n    flatten_body = _flatten_dict(body)\n\n    for key in flatten_body:\n        if key.endswith(sanitized_keys):\n            flatten_body[key] = sanitized_value\n\n    return str(_unflatten_dict(flatten_body))\n", "path": "instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/utils.py"}]}