We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Only collect `db.statement` if there is sanitization
Spec https://github.com/open-telemetry/opentelemetry-specification/pull/3127
- [ ] [aiopg](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-aiopg)
- [ ] [asyncpg](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-asyncpg)
- [ ] [dbapi](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-dbapi)
- [ ] [elasticsearch](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-elasticsearch)
- [ ] [mysql](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-mysql)
- [ ] [pymemcache](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-pymemcache)
- [ ] [pymongo](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-pymongo)
- [ ] [pymysql](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-pymysql)
- [ ] [redis](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-redis)
- [ ] [sqlalchemy](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-sqlalchemy)
- [ ] [sqlite3](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-sqlite3)

--- 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         if isinstance(v, dict):
33             items.extend(_flatten_dict(v, new_key).items())
34         else:
35             items.append((new_key, v))
36     return dict(items)
37 
38 
39 def _unflatten_dict(d):
40     res = {}
41     for k, v in d.items():
42         keys = k.split(".")
43         d = res
44         for key in keys[:-1]:
45             if key not in d:
46                 d[key] = {}
47             d = d[key]
48         d[keys[-1]] = v
49     return res
50 
51 
52 def sanitize_body(body) -> str:
53     flatten_body = _flatten_dict(body)
54 
55     for key in flatten_body:
56         if key.endswith(sanitized_keys):
57             flatten_body[key] = sanitized_value
58 
59     return str(_unflatten_dict(flatten_body))
60
```

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 sanitize_query (bool) - an optional query sanitization flag
48 request_hook (Callable) - a function with extra user-defined logic to be performed before performing the request
49 this function signature is:
50 def request_hook(span: Span, method: str, url: str, kwargs)
51 
52 response_hook (Callable) - a function with extra user-defined logic to be performed after performing the request
53 this function signature is:
54 def response_hook(span: Span, response: dict)
55 
56 for example:
57 
58 .. code: python
59 
60     from opentelemetry.instrumentation.elasticsearch import ElasticsearchInstrumentor
61     import elasticsearch
62 
63     def request_hook(span, method, url, kwargs):
64         if span and span.is_recording():
65             span.set_attribute("custom_user_attribute_from_request_hook", "some-value")
66 
67     def response_hook(span, response):
68         if span and span.is_recording():
69             span.set_attribute("custom_user_attribute_from_response_hook", "some-value")
70 
71     # instrument elasticsearch with request and response hooks
72     ElasticsearchInstrumentor().instrument(request_hook=request_hook, response_hook=response_hook)
73 
74     # Using elasticsearch as normal now will automatically generate spans,
75     # including user custom attributes added from the hooks
76     es = elasticsearch.Elasticsearch()
77     es.index(index='my-index', doc_type='my-type', id=1, body={'my': 'data', 'timestamp': datetime.now()})
78     es.get(index='my-index', doc_type='my-type', id=1)
79 
80 API
81 ---
82 """
83 
84 import re
85 from logging import getLogger
86 from os import environ
87 from typing import Collection
88 
89 import elasticsearch
90 import elasticsearch.exceptions
91 from wrapt import wrap_function_wrapper as _wrap
92 
93 from opentelemetry.instrumentation.elasticsearch.package import _instruments
94 from opentelemetry.instrumentation.elasticsearch.version import __version__
95 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
96 from opentelemetry.instrumentation.utils import unwrap
97 from opentelemetry.semconv.trace import SpanAttributes
98 from opentelemetry.trace import SpanKind, get_tracer
99 
100 from .utils import sanitize_body
101 
102 logger = getLogger(__name__)
103 
104 
105 # Values to add as tags from the actual
106 # payload returned by Elasticsearch, if any.
107 _ATTRIBUTES_FROM_RESULT = [
108     "found",
109     "timed_out",
110     "took",
111 ]
112 
113 _DEFAULT_OP_NAME = "request"
114 
115 
116 class ElasticsearchInstrumentor(BaseInstrumentor):
117     """An instrumentor for elasticsearch
118     See `BaseInstrumentor`
119     """
120 
121     def __init__(self, span_name_prefix=None):
122         if not span_name_prefix:
123             span_name_prefix = environ.get(
124                 "OTEL_PYTHON_ELASTICSEARCH_NAME_PREFIX",
125                 "Elasticsearch",
126             )
127         self._span_name_prefix = span_name_prefix.strip()
128         super().__init__()
129 
130     def instrumentation_dependencies(self) -> Collection[str]:
131         return _instruments
132 
133     def _instrument(self, **kwargs):
134         """
135         Instruments Elasticsearch module
136         """
137         tracer_provider = kwargs.get("tracer_provider")
138         tracer = get_tracer(__name__, __version__, tracer_provider)
139         request_hook = kwargs.get("request_hook")
140         response_hook = kwargs.get("response_hook")
141         sanitize_query = kwargs.get("sanitize_query", False)
142         _wrap(
143             elasticsearch,
144             "Transport.perform_request",
145             _wrap_perform_request(
146                 tracer,
147                 sanitize_query,
148                 self._span_name_prefix,
149                 request_hook,
150                 response_hook,
151             ),
152         )
153 
154     def _uninstrument(self, **kwargs):
155         unwrap(elasticsearch.Transport, "perform_request")
156 
157 
158 _regex_doc_url = re.compile(r"/_doc/([^/]+)")
159 
160 # search api https://www.elastic.co/guide/en/elasticsearch/reference/current/search-search.html
161 _regex_search_url = re.compile(r"/([^/]+)/_search[/]?")
162 
163 
164 def _wrap_perform_request(
165     tracer,
166     sanitize_query,
167     span_name_prefix,
168     request_hook=None,
169     response_hook=None,
170 ):
171     # pylint: disable=R0912,R0914
172     def wrapper(wrapped, _, args, kwargs):
173         method = url = None
174         try:
175             method, url, *_ = args
176         except IndexError:
177             logger.warning(
178                 "expected perform_request to receive two positional arguments. "
179                 "Got %d",
180                 len(args),
181             )
182 
183         op_name = span_name_prefix + (url or method or _DEFAULT_OP_NAME)
184 
185         doc_id = None
186         search_target = None
187 
188         if url:
189             # 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()
190             # A limitation of the regex is that only the '_doc' mapping type is supported. Mapping types are deprecated since Elasticsearch 7
191             # https://github.com/open-telemetry/opentelemetry-python-contrib/issues/708
192             match = _regex_doc_url.search(url)
193             if match is not None:
194                 # Remove the full document ID from the URL
195                 doc_span = match.span()
196                 op_name = (
197                     span_name_prefix
198                     + url[: doc_span[0]]
199                     + "/_doc/:id"
200                     + url[doc_span[1] :]
201                 )
202                 # Put the document ID in attributes
203                 doc_id = match.group(1)
204             match = _regex_search_url.search(url)
205             if match is not None:
206                 op_name = span_name_prefix + "/<target>/_search"
207                 search_target = match.group(1)
208 
209         params = kwargs.get("params", {})
210         body = kwargs.get("body", None)
211 
212         with tracer.start_as_current_span(
213             op_name,
214             kind=SpanKind.CLIENT,
215         ) as span:
216             if callable(request_hook):
217                 request_hook(span, method, url, kwargs)
218 
219             if span.is_recording():
220                 attributes = {
221                     SpanAttributes.DB_SYSTEM: "elasticsearch",
222                 }
223                 if url:
224                     attributes["elasticsearch.url"] = url
225                 if method:
226                     attributes["elasticsearch.method"] = method
227                 if body:
228                     statement = str(body)
229                     if sanitize_query:
230                         statement = sanitize_body(body)
231                     attributes[SpanAttributes.DB_STATEMENT] = statement
232                 if params:
233                     attributes["elasticsearch.params"] = str(params)
234                 if doc_id:
235                     attributes["elasticsearch.id"] = doc_id
236                 if search_target:
237                     attributes["elasticsearch.target"] = search_target
238                 for key, value in attributes.items():
239                     span.set_attribute(key, value)
240 
241             rv = wrapped(*args, **kwargs)
242             if isinstance(rv, dict) and span.is_recording():
243                 for member in _ATTRIBUTES_FROM_RESULT:
244                     if member in rv:
245                         span.set_attribute(
246                             f"elasticsearch.{member}",
247                             str(rv[member]),
248                         )
249 
250             if callable(response_hook):
251                 response_hook(span, rv)
252             return rv
253 
254     return wrapper
255
```
--- 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
@@ -44,7 +44,6 @@
 
 The instrument() method accepts the following keyword args:
 tracer_provider (TracerProvider) - an optional tracer provider
-sanitize_query (bool) - an optional query sanitization flag
 request_hook (Callable) - a function with extra user-defined logic to be performed before performing the request
 this function signature is:
 def request_hook(span: Span, method: str, url: str, kwargs)
@@ -138,13 +137,11 @@
         tracer = get_tracer(__name__, __version__, tracer_provider)
         request_hook = kwargs.get("request_hook")
         response_hook = kwargs.get("response_hook")
-        sanitize_query = kwargs.get("sanitize_query", False)
         _wrap(
             elasticsearch,
             "Transport.perform_request",
             _wrap_perform_request(
                 tracer,
-                sanitize_query,
                 self._span_name_prefix,
                 request_hook,
                 response_hook,
@@ -163,7 +160,6 @@
 
 def _wrap_perform_request(
     tracer,
-    sanitize_query,
     span_name_prefix,
     request_hook=None,
     response_hook=None,
@@ -225,10 +221,9 @@
                 if method:
                     attributes["elasticsearch.method"] = method
                 if body:
-                    statement = str(body)
-                    if sanitize_query:
-                        statement = sanitize_body(body)
-                    attributes[SpanAttributes.DB_STATEMENT] = statement
+                    attributes[SpanAttributes.DB_STATEMENT] = sanitize_body(
+                        body
+                    )
                 if params:
                     attributes["elasticsearch.params"] = str(params)
                 if doc_id:
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
@@ -29,7 +29,8 @@
     items = []
     for k, v in d.items():
         new_key = parent_key + "." + k if parent_key else k
-        if isinstance(v, dict):
+        # recursive call _flatten_dict for a non-empty dict value
+        if isinstance(v, dict) and v:
             items.extend(_flatten_dict(v, new_key).items())
         else:
             items.append((new_key, v))

{"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@@ -44,7 +44,6 @@\n \n The instrument() method accepts the following keyword args:\n tracer_provider (TracerProvider) - an optional tracer provider\n-sanitize_query (bool) - an optional query sanitization flag\n request_hook (Callable) - a function with extra user-defined logic to be performed before performing the request\n this function signature is:\n def request_hook(span: Span, method: str, url: str, kwargs)\n@@ -138,13 +137,11 @@\n         tracer = get_tracer(__name__, __version__, tracer_provider)\n         request_hook = kwargs.get(\"request_hook\")\n         response_hook = kwargs.get(\"response_hook\")\n-        sanitize_query = kwargs.get(\"sanitize_query\", False)\n         _wrap(\n             elasticsearch,\n             \"Transport.perform_request\",\n             _wrap_perform_request(\n                 tracer,\n-                sanitize_query,\n                 self._span_name_prefix,\n                 request_hook,\n                 response_hook,\n@@ -163,7 +160,6 @@\n \n def _wrap_perform_request(\n     tracer,\n-    sanitize_query,\n     span_name_prefix,\n     request_hook=None,\n     response_hook=None,\n@@ -225,10 +221,9 @@\n                 if method:\n                     attributes[\"elasticsearch.method\"] = method\n                 if body:\n-                    statement = str(body)\n-                    if sanitize_query:\n-                        statement = sanitize_body(body)\n-                    attributes[SpanAttributes.DB_STATEMENT] = statement\n+                    attributes[SpanAttributes.DB_STATEMENT] = sanitize_body(\n+                        body\n+                    )\n                 if params:\n                     attributes[\"elasticsearch.params\"] = str(params)\n                 if doc_id:\ndiff --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@@ -29,7 +29,8 @@\n     items = []\n     for k, v in d.items():\n         new_key = parent_key + \".\" + k if parent_key else k\n-        if isinstance(v, dict):\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", "issue": "Only collect `db.statement` if there is sanitization\nSpec https://github.com/open-telemetry/opentelemetry-specification/pull/3127\r\n- [ ] [aiopg](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-aiopg)\r\n- [ ] [asyncpg](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-asyncpg)\r\n- [ ] [dbapi](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-dbapi)\r\n- [ ] [elasticsearch](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-elasticsearch)\r\n- [ ] [mysql](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-mysql)\r\n- [ ] [pymemcache](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-pymemcache)\r\n- [ ] [pymongo](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-pymongo)\r\n- [ ] [pymysql](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-pymysql)\r\n- [ ] [redis](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-redis)\r\n- [ ] [sqlalchemy](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-sqlalchemy)\r\n- [ ] [sqlite3](https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/instrumentation/opentelemetry-instrumentation-sqlite3)\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        if isinstance(v, dict):\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"}, {"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\nsanitize_query (bool) - an optional query sanitization flag\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\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(__name__, __version__, tracer_provider)\n        request_hook = kwargs.get(\"request_hook\")\n        response_hook = kwargs.get(\"response_hook\")\n        sanitize_query = kwargs.get(\"sanitize_query\", False)\n        _wrap(\n            elasticsearch,\n            \"Transport.perform_request\",\n            _wrap_perform_request(\n                tracer,\n                sanitize_query,\n                self._span_name_prefix,\n                request_hook,\n                response_hook,\n            ),\n        )\n\n    def _uninstrument(self, **kwargs):\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    sanitize_query,\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                    statement = str(body)\n                    if sanitize_query:\n                        statement = sanitize_body(body)\n                    attributes[SpanAttributes.DB_STATEMENT] = statement\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\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"}, {"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\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(__name__, __version__, tracer_provider)\n        request_hook = kwargs.get(\"request_hook\")\n        response_hook = kwargs.get(\"response_hook\")\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        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"}]}