We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
fix: elasticsearch `bulk` API produces lists
# Description

Fixes an issue seen in production where the elasticsearch bulk API would end up sending a list through this code path and not a dictionary.

For example:

```
File \"/code/platform_be/core/logic/ingest.py\", line 378, in update_source_doc_version_history
        bulk(es_client, bulk_es_updates)
      File \"/opt/python/lib/python3.10/site-packages/elasticsearch/helpers/actions.py\", line 521, in bulk
        for ok, item in streaming_bulk(
      File \"/opt/python/lib/python3.10/site-packages/elasticsearch/helpers/actions.py\", line 436, in streaming_bulk
        for data, (ok, info) in zip(
      File \"/opt/python/lib/python3.10/site-packages/elasticsearch/helpers/actions.py\", line 339, in _process_bulk_chunk
        resp = client.bulk(*args, operations=bulk_actions, **kwargs)  # type: ignore[arg-type]
      File \"/opt/python/lib/python3.10/site-packages/elasticsearch/_sync/client/utils.py\", line 414, in wrapped
        return api(*args, **kwargs)
      File \"/opt/python/lib/python3.10/site-packages/elasticsearch/_sync/client/__init__.py\", line 704, in bulk
        return self.perform_request(  # type: ignore[return-value]
      File \"/opt/python/lib/python3.10/site-packages/elasticsearch/_sync/client/_base.py\", line 285, in perform_request
        meta, resp_body = self.transport.perform_request(
      File \"/opt/python/lib/python3.10/site-packages/opentelemetry/instrumentation/elasticsearch/__init__.py\", line 242, in wrapper
        attributes[SpanAttributes.DB_STATEMENT] = sanitize_body(
      File \"/opt/python/lib/python3.10/site-packages/opentelemetry/instrumentation/elasticsearch/utils.py\", line 58, in sanitize_body
        flatten_body = _flatten_dict(body)
      File \"/opt/python/lib/python3.10/site-packages/opentelemetry/instrumentation/elasticsearch/utils.py\", line 31, in _flatten_dict
        for k, v in d.items():
    AttributeError: 'list' object has no attribute 'items'"
```

## Type of change

Please delete options that are not relevant.

- [x] Bug fix (non-breaking change which fixes an issue)

# How Has This Been Tested?

Please describe the tests that you ran to verify your changes. Provide instructions so we can reproduce. Please also list any relevant details for your test configuration

- [x] Unit tests

# Does This PR Require a Core Repo Change?

- [x] No.

# Checklist:

See [contributing.md](https://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/CONTRIBUTING.md) for styleguide, changelog guidelines, and more.

- [ ] Followed the style guidelines of this project
- [ ] Changelogs have been updated
- [ ] Unit tests have been added
- [ ] Documentation has been updated


--- 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/__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 """
16 This library allows tracing HTTP elasticsearch made by the
17 `elasticsearch <https://elasticsearch-py.readthedocs.io/en/master/>`_ library.
18 
19 Usage
20 -----
21 
22 .. code-block:: python
23 
24     from opentelemetry.instrumentation.elasticsearch import ElasticsearchInstrumentor
25     import elasticsearch
26 
27 
28     # instrument elasticsearch
29     ElasticsearchInstrumentor().instrument()
30 
31     # Using elasticsearch as normal now will automatically generate spans
32     es = elasticsearch.Elasticsearch()
33     es.index(index='my-index', doc_type='my-type', id=1, body={'my': 'data', 'timestamp': datetime.now()})
34     es.get(index='my-index', doc_type='my-type', id=1)
35 
36 Elasticsearch instrumentation prefixes operation names with the string "Elasticsearch". This
37 can be changed to a different string by either setting the OTEL_PYTHON_ELASTICSEARCH_NAME_PREFIX
38 environment variable or by passing the prefix as an argument to the instrumentor. For example,
39 
40 
41 .. code-block:: python
42 
43     ElasticsearchInstrumentor("my-custom-prefix").instrument()
44 
45 The instrument() method accepts the following keyword args:
46 tracer_provider (TracerProvider) - an optional tracer provider
47 request_hook (Callable) - a function with extra user-defined logic to be performed before performing the request
48 this function signature is:
49 def request_hook(span: Span, method: str, url: str, kwargs)
50 
51 response_hook (Callable) - a function with extra user-defined logic to be performed after performing the request
52 this function signature is:
53 def response_hook(span: Span, response: dict)
54 
55 for example:
56 
57 .. code: python
58 
59     from opentelemetry.instrumentation.elasticsearch import ElasticsearchInstrumentor
60     import elasticsearch
61 
62     def request_hook(span, method, url, kwargs):
63         if span and span.is_recording():
64             span.set_attribute("custom_user_attribute_from_request_hook", "some-value")
65 
66     def response_hook(span, response):
67         if span and span.is_recording():
68             span.set_attribute("custom_user_attribute_from_response_hook", "some-value")
69 
70     # instrument elasticsearch with request and response hooks
71     ElasticsearchInstrumentor().instrument(request_hook=request_hook, response_hook=response_hook)
72 
73     # Using elasticsearch as normal now will automatically generate spans,
74     # including user custom attributes added from the hooks
75     es = elasticsearch.Elasticsearch()
76     es.index(index='my-index', doc_type='my-type', id=1, body={'my': 'data', 'timestamp': datetime.now()})
77     es.get(index='my-index', doc_type='my-type', id=1)
78 
79 API
80 ---
81 """
82 
83 import re
84 from logging import getLogger
85 from os import environ
86 from typing import Collection
87 
88 import elasticsearch
89 import elasticsearch.exceptions
90 from wrapt import wrap_function_wrapper as _wrap
91 
92 from opentelemetry.instrumentation.elasticsearch.package import _instruments
93 from opentelemetry.instrumentation.elasticsearch.version import __version__
94 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
95 from opentelemetry.instrumentation.utils import unwrap
96 from opentelemetry.semconv.trace import SpanAttributes
97 from opentelemetry.trace import SpanKind, get_tracer
98 
99 from .utils import sanitize_body
100 
101 # Split of elasticsearch and elastic_transport in 8.0.0+
102 # https://www.elastic.co/guide/en/elasticsearch/client/python-api/master/release-notes.html#rn-8-0-0
103 es_transport_split = elasticsearch.VERSION[0] > 7
104 if es_transport_split:
105     import elastic_transport
106 
107 logger = getLogger(__name__)
108 
109 
110 # Values to add as tags from the actual
111 # payload returned by Elasticsearch, if any.
112 _ATTRIBUTES_FROM_RESULT = [
113     "found",
114     "timed_out",
115     "took",
116 ]
117 
118 _DEFAULT_OP_NAME = "request"
119 
120 
121 class ElasticsearchInstrumentor(BaseInstrumentor):
122     """An instrumentor for elasticsearch
123     See `BaseInstrumentor`
124     """
125 
126     def __init__(self, span_name_prefix=None):
127         if not span_name_prefix:
128             span_name_prefix = environ.get(
129                 "OTEL_PYTHON_ELASTICSEARCH_NAME_PREFIX",
130                 "Elasticsearch",
131             )
132         self._span_name_prefix = span_name_prefix.strip()
133         super().__init__()
134 
135     def instrumentation_dependencies(self) -> Collection[str]:
136         return _instruments
137 
138     def _instrument(self, **kwargs):
139         """
140         Instruments Elasticsearch module
141         """
142         tracer_provider = kwargs.get("tracer_provider")
143         tracer = get_tracer(
144             __name__,
145             __version__,
146             tracer_provider,
147             schema_url="https://opentelemetry.io/schemas/1.11.0",
148         )
149         request_hook = kwargs.get("request_hook")
150         response_hook = kwargs.get("response_hook")
151         if es_transport_split:
152             _wrap(
153                 elastic_transport,
154                 "Transport.perform_request",
155                 _wrap_perform_request(
156                     tracer,
157                     self._span_name_prefix,
158                     request_hook,
159                     response_hook,
160                 ),
161             )
162         else:
163             _wrap(
164                 elasticsearch,
165                 "Transport.perform_request",
166                 _wrap_perform_request(
167                     tracer,
168                     self._span_name_prefix,
169                     request_hook,
170                     response_hook,
171                 ),
172             )
173 
174     def _uninstrument(self, **kwargs):
175         # pylint: disable=no-member
176         unwrap(elasticsearch.Transport, "perform_request")
177 
178 
179 _regex_doc_url = re.compile(r"/_doc/([^/]+)")
180 
181 # search api https://www.elastic.co/guide/en/elasticsearch/reference/current/search-search.html
182 _regex_search_url = re.compile(r"/([^/]+)/_search[/]?")
183 
184 
185 def _wrap_perform_request(
186     tracer,
187     span_name_prefix,
188     request_hook=None,
189     response_hook=None,
190 ):
191     # pylint: disable=R0912,R0914
192     def wrapper(wrapped, _, args, kwargs):
193         method = url = None
194         try:
195             method, url, *_ = args
196         except IndexError:
197             logger.warning(
198                 "expected perform_request to receive two positional arguments. "
199                 "Got %d",
200                 len(args),
201             )
202 
203         op_name = span_name_prefix + (url or method or _DEFAULT_OP_NAME)
204 
205         doc_id = None
206         search_target = None
207 
208         if url:
209             # TODO: This regex-based solution avoids creating an unbounded number of span names, but should be replaced by instrumenting individual Elasticsearch methods instead of Transport.perform_request()
210             # A limitation of the regex is that only the '_doc' mapping type is supported. Mapping types are deprecated since Elasticsearch 7
211             # https://github.com/open-telemetry/opentelemetry-python-contrib/issues/708
212             match = _regex_doc_url.search(url)
213             if match is not None:
214                 # Remove the full document ID from the URL
215                 doc_span = match.span()
216                 op_name = (
217                     span_name_prefix
218                     + url[: doc_span[0]]
219                     + "/_doc/:id"
220                     + url[doc_span[1] :]
221                 )
222                 # Put the document ID in attributes
223                 doc_id = match.group(1)
224             match = _regex_search_url.search(url)
225             if match is not None:
226                 op_name = span_name_prefix + "/<target>/_search"
227                 search_target = match.group(1)
228 
229         params = kwargs.get("params", {})
230         body = kwargs.get("body", None)
231 
232         with tracer.start_as_current_span(
233             op_name,
234             kind=SpanKind.CLIENT,
235         ) as span:
236             if callable(request_hook):
237                 request_hook(span, method, url, kwargs)
238 
239             if span.is_recording():
240                 attributes = {
241                     SpanAttributes.DB_SYSTEM: "elasticsearch",
242                 }
243                 if url:
244                     attributes["elasticsearch.url"] = url
245                 if method:
246                     attributes["elasticsearch.method"] = method
247                 if body:
248                     attributes[SpanAttributes.DB_STATEMENT] = sanitize_body(
249                         body
250                     )
251                 if params:
252                     attributes["elasticsearch.params"] = str(params)
253                 if doc_id:
254                     attributes["elasticsearch.id"] = doc_id
255                 if search_target:
256                     attributes["elasticsearch.target"] = search_target
257                 for key, value in attributes.items():
258                     span.set_attribute(key, value)
259 
260             rv = wrapped(*args, **kwargs)
261             if isinstance(rv, dict) and span.is_recording():
262                 for member in _ATTRIBUTES_FROM_RESULT:
263                     if member in rv:
264                         span.set_attribute(
265                             f"elasticsearch.{member}",
266                             str(rv[member]),
267                         )
268 
269             if callable(response_hook):
270                 response_hook(span, rv)
271             return rv
272 
273     return wrapper
274
```
--- 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-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/__init__.py b/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/__init__.py
--- a/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/__init__.py
@@ -245,9 +245,11 @@
                 if method:
                     attributes["elasticsearch.method"] = method
                 if body:
-                    attributes[SpanAttributes.DB_STATEMENT] = sanitize_body(
-                        body
-                    )
+                    # Don't set db.statement for bulk requests, as it can be very large
+                    if isinstance(body, dict):
+                        attributes[
+                            SpanAttributes.DB_STATEMENT
+                        ] = sanitize_body(body)
                 if params:
                     attributes["elasticsearch.params"] = str(params)
                 if doc_id:

{"golden_diff": "diff --git a/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/__init__.py b/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/__init__.py\n--- a/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/__init__.py\n+++ b/instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/__init__.py\n@@ -245,9 +245,11 @@\n                 if method:\n                     attributes[\"elasticsearch.method\"] = method\n                 if body:\n-                    attributes[SpanAttributes.DB_STATEMENT] = sanitize_body(\n-                        body\n-                    )\n+                    # Don't set db.statement for bulk requests, as it can be very large\n+                    if isinstance(body, dict):\n+                        attributes[\n+                            SpanAttributes.DB_STATEMENT\n+                        ] = sanitize_body(body)\n                 if params:\n                     attributes[\"elasticsearch.params\"] = str(params)\n                 if doc_id:\n", "issue": "fix: elasticsearch `bulk` API produces lists\n# Description\r\n\r\nFixes an issue seen in production where the elasticsearch bulk API would end up sending a list through this code path and not a dictionary.\r\n\r\nFor example:\r\n\r\n```\r\nFile \\\"/code/platform_be/core/logic/ingest.py\\\", line 378, in update_source_doc_version_history\r\n        bulk(es_client, bulk_es_updates)\r\n      File \\\"/opt/python/lib/python3.10/site-packages/elasticsearch/helpers/actions.py\\\", line 521, in bulk\r\n        for ok, item in streaming_bulk(\r\n      File \\\"/opt/python/lib/python3.10/site-packages/elasticsearch/helpers/actions.py\\\", line 436, in streaming_bulk\r\n        for data, (ok, info) in zip(\r\n      File \\\"/opt/python/lib/python3.10/site-packages/elasticsearch/helpers/actions.py\\\", line 339, in _process_bulk_chunk\r\n        resp = client.bulk(*args, operations=bulk_actions, **kwargs)  # type: ignore[arg-type]\r\n      File \\\"/opt/python/lib/python3.10/site-packages/elasticsearch/_sync/client/utils.py\\\", line 414, in wrapped\r\n        return api(*args, **kwargs)\r\n      File \\\"/opt/python/lib/python3.10/site-packages/elasticsearch/_sync/client/__init__.py\\\", line 704, in bulk\r\n        return self.perform_request(  # type: ignore[return-value]\r\n      File \\\"/opt/python/lib/python3.10/site-packages/elasticsearch/_sync/client/_base.py\\\", line 285, in perform_request\r\n        meta, resp_body = self.transport.perform_request(\r\n      File \\\"/opt/python/lib/python3.10/site-packages/opentelemetry/instrumentation/elasticsearch/__init__.py\\\", line 242, in wrapper\r\n        attributes[SpanAttributes.DB_STATEMENT] = sanitize_body(\r\n      File \\\"/opt/python/lib/python3.10/site-packages/opentelemetry/instrumentation/elasticsearch/utils.py\\\", line 58, in sanitize_body\r\n        flatten_body = _flatten_dict(body)\r\n      File \\\"/opt/python/lib/python3.10/site-packages/opentelemetry/instrumentation/elasticsearch/utils.py\\\", line 31, in _flatten_dict\r\n        for k, v in d.items():\r\n    AttributeError: 'list' object has no attribute 'items'\"\r\n```\r\n\r\n## Type of change\r\n\r\nPlease delete options that are not relevant.\r\n\r\n- [x] Bug fix (non-breaking change which fixes an issue)\r\n\r\n# How Has This Been Tested?\r\n\r\nPlease describe the tests that you ran to verify your changes. Provide instructions so we can reproduce. Please also list any relevant details for your test configuration\r\n\r\n- [x] Unit tests\r\n\r\n# Does This PR Require a Core Repo Change?\r\n\r\n- [x] No.\r\n\r\n# Checklist:\r\n\r\nSee [contributing.md](https://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/CONTRIBUTING.md) for styleguide, changelog guidelines, and more.\r\n\r\n- [ ] Followed the style guidelines of this project\r\n- [ ] Changelogs have been updated\r\n- [ ] Unit tests have been added\r\n- [ ] Documentation has been updated\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\n\"\"\"\nThis library allows tracing HTTP elasticsearch made by the\n`elasticsearch <https://elasticsearch-py.readthedocs.io/en/master/>`_ library.\n\nUsage\n-----\n\n.. code-block:: python\n\n    from opentelemetry.instrumentation.elasticsearch import ElasticsearchInstrumentor\n    import elasticsearch\n\n\n    # instrument elasticsearch\n    ElasticsearchInstrumentor().instrument()\n\n    # Using elasticsearch as normal now will automatically generate spans\n    es = elasticsearch.Elasticsearch()\n    es.index(index='my-index', doc_type='my-type', id=1, body={'my': 'data', 'timestamp': datetime.now()})\n    es.get(index='my-index', doc_type='my-type', id=1)\n\nElasticsearch instrumentation prefixes operation names with the string \"Elasticsearch\". This\ncan be changed to a different string by either setting the OTEL_PYTHON_ELASTICSEARCH_NAME_PREFIX\nenvironment variable or by passing the prefix as an argument to the instrumentor. For example,\n\n\n.. code-block:: python\n\n    ElasticsearchInstrumentor(\"my-custom-prefix\").instrument()\n\nThe instrument() method accepts the following keyword args:\ntracer_provider (TracerProvider) - an optional tracer provider\nrequest_hook (Callable) - a function with extra user-defined logic to be performed before performing the request\nthis function signature is:\ndef request_hook(span: Span, method: str, url: str, kwargs)\n\nresponse_hook (Callable) - a function with extra user-defined logic to be performed after performing the request\nthis function signature is:\ndef response_hook(span: Span, response: dict)\n\nfor example:\n\n.. code: python\n\n    from opentelemetry.instrumentation.elasticsearch import ElasticsearchInstrumentor\n    import elasticsearch\n\n    def request_hook(span, method, url, kwargs):\n        if span and span.is_recording():\n            span.set_attribute(\"custom_user_attribute_from_request_hook\", \"some-value\")\n\n    def response_hook(span, response):\n        if span and span.is_recording():\n            span.set_attribute(\"custom_user_attribute_from_response_hook\", \"some-value\")\n\n    # instrument elasticsearch with request and response hooks\n    ElasticsearchInstrumentor().instrument(request_hook=request_hook, response_hook=response_hook)\n\n    # Using elasticsearch as normal now will automatically generate spans,\n    # including user custom attributes added from the hooks\n    es = elasticsearch.Elasticsearch()\n    es.index(index='my-index', doc_type='my-type', id=1, body={'my': 'data', 'timestamp': datetime.now()})\n    es.get(index='my-index', doc_type='my-type', id=1)\n\nAPI\n---\n\"\"\"\n\nimport re\nfrom logging import getLogger\nfrom os import environ\nfrom typing import Collection\n\nimport elasticsearch\nimport elasticsearch.exceptions\nfrom wrapt import wrap_function_wrapper as _wrap\n\nfrom opentelemetry.instrumentation.elasticsearch.package import _instruments\nfrom opentelemetry.instrumentation.elasticsearch.version import __version__\nfrom opentelemetry.instrumentation.instrumentor import BaseInstrumentor\nfrom opentelemetry.instrumentation.utils import unwrap\nfrom opentelemetry.semconv.trace import SpanAttributes\nfrom opentelemetry.trace import SpanKind, get_tracer\n\nfrom .utils import sanitize_body\n\n# Split of elasticsearch and elastic_transport in 8.0.0+\n# https://www.elastic.co/guide/en/elasticsearch/client/python-api/master/release-notes.html#rn-8-0-0\nes_transport_split = elasticsearch.VERSION[0] > 7\nif es_transport_split:\n    import elastic_transport\n\nlogger = getLogger(__name__)\n\n\n# Values to add as tags from the actual\n# payload returned by Elasticsearch, if any.\n_ATTRIBUTES_FROM_RESULT = [\n    \"found\",\n    \"timed_out\",\n    \"took\",\n]\n\n_DEFAULT_OP_NAME = \"request\"\n\n\nclass ElasticsearchInstrumentor(BaseInstrumentor):\n    \"\"\"An instrumentor for elasticsearch\n    See `BaseInstrumentor`\n    \"\"\"\n\n    def __init__(self, span_name_prefix=None):\n        if not span_name_prefix:\n            span_name_prefix = environ.get(\n                \"OTEL_PYTHON_ELASTICSEARCH_NAME_PREFIX\",\n                \"Elasticsearch\",\n            )\n        self._span_name_prefix = span_name_prefix.strip()\n        super().__init__()\n\n    def instrumentation_dependencies(self) -> Collection[str]:\n        return _instruments\n\n    def _instrument(self, **kwargs):\n        \"\"\"\n        Instruments Elasticsearch module\n        \"\"\"\n        tracer_provider = kwargs.get(\"tracer_provider\")\n        tracer = get_tracer(\n            __name__,\n            __version__,\n            tracer_provider,\n            schema_url=\"https://opentelemetry.io/schemas/1.11.0\",\n        )\n        request_hook = kwargs.get(\"request_hook\")\n        response_hook = kwargs.get(\"response_hook\")\n        if es_transport_split:\n            _wrap(\n                elastic_transport,\n                \"Transport.perform_request\",\n                _wrap_perform_request(\n                    tracer,\n                    self._span_name_prefix,\n                    request_hook,\n                    response_hook,\n                ),\n            )\n        else:\n            _wrap(\n                elasticsearch,\n                \"Transport.perform_request\",\n                _wrap_perform_request(\n                    tracer,\n                    self._span_name_prefix,\n                    request_hook,\n                    response_hook,\n                ),\n            )\n\n    def _uninstrument(self, **kwargs):\n        # pylint: disable=no-member\n        unwrap(elasticsearch.Transport, \"perform_request\")\n\n\n_regex_doc_url = re.compile(r\"/_doc/([^/]+)\")\n\n# search api https://www.elastic.co/guide/en/elasticsearch/reference/current/search-search.html\n_regex_search_url = re.compile(r\"/([^/]+)/_search[/]?\")\n\n\ndef _wrap_perform_request(\n    tracer,\n    span_name_prefix,\n    request_hook=None,\n    response_hook=None,\n):\n    # pylint: disable=R0912,R0914\n    def wrapper(wrapped, _, args, kwargs):\n        method = url = None\n        try:\n            method, url, *_ = args\n        except IndexError:\n            logger.warning(\n                \"expected perform_request to receive two positional arguments. \"\n                \"Got %d\",\n                len(args),\n            )\n\n        op_name = span_name_prefix + (url or method or _DEFAULT_OP_NAME)\n\n        doc_id = None\n        search_target = None\n\n        if url:\n            # TODO: This regex-based solution avoids creating an unbounded number of span names, but should be replaced by instrumenting individual Elasticsearch methods instead of Transport.perform_request()\n            # A limitation of the regex is that only the '_doc' mapping type is supported. Mapping types are deprecated since Elasticsearch 7\n            # https://github.com/open-telemetry/opentelemetry-python-contrib/issues/708\n            match = _regex_doc_url.search(url)\n            if match is not None:\n                # Remove the full document ID from the URL\n                doc_span = match.span()\n                op_name = (\n                    span_name_prefix\n                    + url[: doc_span[0]]\n                    + \"/_doc/:id\"\n                    + url[doc_span[1] :]\n                )\n                # Put the document ID in attributes\n                doc_id = match.group(1)\n            match = _regex_search_url.search(url)\n            if match is not None:\n                op_name = span_name_prefix + \"/<target>/_search\"\n                search_target = match.group(1)\n\n        params = kwargs.get(\"params\", {})\n        body = kwargs.get(\"body\", None)\n\n        with tracer.start_as_current_span(\n            op_name,\n            kind=SpanKind.CLIENT,\n        ) as span:\n            if callable(request_hook):\n                request_hook(span, method, url, kwargs)\n\n            if span.is_recording():\n                attributes = {\n                    SpanAttributes.DB_SYSTEM: \"elasticsearch\",\n                }\n                if url:\n                    attributes[\"elasticsearch.url\"] = url\n                if method:\n                    attributes[\"elasticsearch.method\"] = method\n                if body:\n                    attributes[SpanAttributes.DB_STATEMENT] = sanitize_body(\n                        body\n                    )\n                if params:\n                    attributes[\"elasticsearch.params\"] = str(params)\n                if doc_id:\n                    attributes[\"elasticsearch.id\"] = doc_id\n                if search_target:\n                    attributes[\"elasticsearch.target\"] = search_target\n                for key, value in attributes.items():\n                    span.set_attribute(key, value)\n\n            rv = wrapped(*args, **kwargs)\n            if isinstance(rv, dict) and span.is_recording():\n                for member in _ATTRIBUTES_FROM_RESULT:\n                    if member in rv:\n                        span.set_attribute(\n                            f\"elasticsearch.{member}\",\n                            str(rv[member]),\n                        )\n\n            if callable(response_hook):\n                response_hook(span, rv)\n            return rv\n\n    return wrapper\n", "path": "instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/__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\n\"\"\"\nThis library allows tracing HTTP elasticsearch made by the\n`elasticsearch <https://elasticsearch-py.readthedocs.io/en/master/>`_ library.\n\nUsage\n-----\n\n.. code-block:: python\n\n    from opentelemetry.instrumentation.elasticsearch import ElasticsearchInstrumentor\n    import elasticsearch\n\n\n    # instrument elasticsearch\n    ElasticsearchInstrumentor().instrument()\n\n    # Using elasticsearch as normal now will automatically generate spans\n    es = elasticsearch.Elasticsearch()\n    es.index(index='my-index', doc_type='my-type', id=1, body={'my': 'data', 'timestamp': datetime.now()})\n    es.get(index='my-index', doc_type='my-type', id=1)\n\nElasticsearch instrumentation prefixes operation names with the string \"Elasticsearch\". This\ncan be changed to a different string by either setting the OTEL_PYTHON_ELASTICSEARCH_NAME_PREFIX\nenvironment variable or by passing the prefix as an argument to the instrumentor. For example,\n\n\n.. code-block:: python\n\n    ElasticsearchInstrumentor(\"my-custom-prefix\").instrument()\n\nThe instrument() method accepts the following keyword args:\ntracer_provider (TracerProvider) - an optional tracer provider\nrequest_hook (Callable) - a function with extra user-defined logic to be performed before performing the request\nthis function signature is:\ndef request_hook(span: Span, method: str, url: str, kwargs)\n\nresponse_hook (Callable) - a function with extra user-defined logic to be performed after performing the request\nthis function signature is:\ndef response_hook(span: Span, response: dict)\n\nfor example:\n\n.. code: python\n\n    from opentelemetry.instrumentation.elasticsearch import ElasticsearchInstrumentor\n    import elasticsearch\n\n    def request_hook(span, method, url, kwargs):\n        if span and span.is_recording():\n            span.set_attribute(\"custom_user_attribute_from_request_hook\", \"some-value\")\n\n    def response_hook(span, response):\n        if span and span.is_recording():\n            span.set_attribute(\"custom_user_attribute_from_response_hook\", \"some-value\")\n\n    # instrument elasticsearch with request and response hooks\n    ElasticsearchInstrumentor().instrument(request_hook=request_hook, response_hook=response_hook)\n\n    # Using elasticsearch as normal now will automatically generate spans,\n    # including user custom attributes added from the hooks\n    es = elasticsearch.Elasticsearch()\n    es.index(index='my-index', doc_type='my-type', id=1, body={'my': 'data', 'timestamp': datetime.now()})\n    es.get(index='my-index', doc_type='my-type', id=1)\n\nAPI\n---\n\"\"\"\n\nimport re\nfrom logging import getLogger\nfrom os import environ\nfrom typing import Collection\n\nimport elasticsearch\nimport elasticsearch.exceptions\nfrom wrapt import wrap_function_wrapper as _wrap\n\nfrom opentelemetry.instrumentation.elasticsearch.package import _instruments\nfrom opentelemetry.instrumentation.elasticsearch.version import __version__\nfrom opentelemetry.instrumentation.instrumentor import BaseInstrumentor\nfrom opentelemetry.instrumentation.utils import unwrap\nfrom opentelemetry.semconv.trace import SpanAttributes\nfrom opentelemetry.trace import SpanKind, get_tracer\n\nfrom .utils import sanitize_body\n\n# Split of elasticsearch and elastic_transport in 8.0.0+\n# https://www.elastic.co/guide/en/elasticsearch/client/python-api/master/release-notes.html#rn-8-0-0\nes_transport_split = elasticsearch.VERSION[0] > 7\nif es_transport_split:\n    import elastic_transport\n\nlogger = getLogger(__name__)\n\n\n# Values to add as tags from the actual\n# payload returned by Elasticsearch, if any.\n_ATTRIBUTES_FROM_RESULT = [\n    \"found\",\n    \"timed_out\",\n    \"took\",\n]\n\n_DEFAULT_OP_NAME = \"request\"\n\n\nclass ElasticsearchInstrumentor(BaseInstrumentor):\n    \"\"\"An instrumentor for elasticsearch\n    See `BaseInstrumentor`\n    \"\"\"\n\n    def __init__(self, span_name_prefix=None):\n        if not span_name_prefix:\n            span_name_prefix = environ.get(\n                \"OTEL_PYTHON_ELASTICSEARCH_NAME_PREFIX\",\n                \"Elasticsearch\",\n            )\n        self._span_name_prefix = span_name_prefix.strip()\n        super().__init__()\n\n    def instrumentation_dependencies(self) -> Collection[str]:\n        return _instruments\n\n    def _instrument(self, **kwargs):\n        \"\"\"\n        Instruments Elasticsearch module\n        \"\"\"\n        tracer_provider = kwargs.get(\"tracer_provider\")\n        tracer = get_tracer(\n            __name__,\n            __version__,\n            tracer_provider,\n            schema_url=\"https://opentelemetry.io/schemas/1.11.0\",\n        )\n        request_hook = kwargs.get(\"request_hook\")\n        response_hook = kwargs.get(\"response_hook\")\n        if es_transport_split:\n            _wrap(\n                elastic_transport,\n                \"Transport.perform_request\",\n                _wrap_perform_request(\n                    tracer,\n                    self._span_name_prefix,\n                    request_hook,\n                    response_hook,\n                ),\n            )\n        else:\n            _wrap(\n                elasticsearch,\n                \"Transport.perform_request\",\n                _wrap_perform_request(\n                    tracer,\n                    self._span_name_prefix,\n                    request_hook,\n                    response_hook,\n                ),\n            )\n\n    def _uninstrument(self, **kwargs):\n        # pylint: disable=no-member\n        unwrap(elasticsearch.Transport, \"perform_request\")\n\n\n_regex_doc_url = re.compile(r\"/_doc/([^/]+)\")\n\n# search api https://www.elastic.co/guide/en/elasticsearch/reference/current/search-search.html\n_regex_search_url = re.compile(r\"/([^/]+)/_search[/]?\")\n\n\ndef _wrap_perform_request(\n    tracer,\n    span_name_prefix,\n    request_hook=None,\n    response_hook=None,\n):\n    # pylint: disable=R0912,R0914\n    def wrapper(wrapped, _, args, kwargs):\n        method = url = None\n        try:\n            method, url, *_ = args\n        except IndexError:\n            logger.warning(\n                \"expected perform_request to receive two positional arguments. \"\n                \"Got %d\",\n                len(args),\n            )\n\n        op_name = span_name_prefix + (url or method or _DEFAULT_OP_NAME)\n\n        doc_id = None\n        search_target = None\n\n        if url:\n            # TODO: This regex-based solution avoids creating an unbounded number of span names, but should be replaced by instrumenting individual Elasticsearch methods instead of Transport.perform_request()\n            # A limitation of the regex is that only the '_doc' mapping type is supported. Mapping types are deprecated since Elasticsearch 7\n            # https://github.com/open-telemetry/opentelemetry-python-contrib/issues/708\n            match = _regex_doc_url.search(url)\n            if match is not None:\n                # Remove the full document ID from the URL\n                doc_span = match.span()\n                op_name = (\n                    span_name_prefix\n                    + url[: doc_span[0]]\n                    + \"/_doc/:id\"\n                    + url[doc_span[1] :]\n                )\n                # Put the document ID in attributes\n                doc_id = match.group(1)\n            match = _regex_search_url.search(url)\n            if match is not None:\n                op_name = span_name_prefix + \"/<target>/_search\"\n                search_target = match.group(1)\n\n        params = kwargs.get(\"params\", {})\n        body = kwargs.get(\"body\", None)\n\n        with tracer.start_as_current_span(\n            op_name,\n            kind=SpanKind.CLIENT,\n        ) as span:\n            if callable(request_hook):\n                request_hook(span, method, url, kwargs)\n\n            if span.is_recording():\n                attributes = {\n                    SpanAttributes.DB_SYSTEM: \"elasticsearch\",\n                }\n                if url:\n                    attributes[\"elasticsearch.url\"] = url\n                if method:\n                    attributes[\"elasticsearch.method\"] = method\n                if body:\n                    # Don't set db.statement for bulk requests, as it can be very large\n                    if isinstance(body, dict):\n                        attributes[\n                            SpanAttributes.DB_STATEMENT\n                        ] = sanitize_body(body)\n                if params:\n                    attributes[\"elasticsearch.params\"] = str(params)\n                if doc_id:\n                    attributes[\"elasticsearch.id\"] = doc_id\n                if search_target:\n                    attributes[\"elasticsearch.target\"] = search_target\n                for key, value in attributes.items():\n                    span.set_attribute(key, value)\n\n            rv = wrapped(*args, **kwargs)\n            if isinstance(rv, dict) and span.is_recording():\n                for member in _ATTRIBUTES_FROM_RESULT:\n                    if member in rv:\n                        span.set_attribute(\n                            f\"elasticsearch.{member}\",\n                            str(rv[member]),\n                        )\n\n            if callable(response_hook):\n                response_hook(span, rv)\n            return rv\n\n    return wrapper\n", "path": "instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/__init__.py"}]}