We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[META 659] Backend dependencies granularity for NoSQL and Messaging
See meta issue for the description and details:
- Meta issue: https://github.com/elastic/apm/issues/659

This is done for everything except for Elasticsearch.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `elasticapm/instrumentation/packages/elasticsearch.py`
Content:
```
1 #  BSD 3-Clause License
2 #
3 #  Copyright (c) 2019, Elasticsearch BV
4 #  All rights reserved.
5 #
6 #  Redistribution and use in source and binary forms, with or without
7 #  modification, are permitted provided that the following conditions are met:
8 #
9 #  * Redistributions of source code must retain the above copyright notice, this
10 #    list of conditions and the following disclaimer.
11 #
12 #  * Redistributions in binary form must reproduce the above copyright notice,
13 #    this list of conditions and the following disclaimer in the documentation
14 #    and/or other materials provided with the distribution.
15 #
16 #  * Neither the name of the copyright holder nor the names of its
17 #    contributors may be used to endorse or promote products derived from
18 #    this software without specific prior written permission.
19 #
20 #  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
21 #  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
22 #  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
23 #  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
24 #  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
25 #  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
26 #  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
27 #  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
28 #  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
29 #  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
30 
31 from __future__ import absolute_import
32 
33 import re
34 from typing import Optional
35 from urllib.parse import parse_qs, urlparse
36 
37 import elasticapm
38 from elasticapm.instrumentation.packages.base import AbstractInstrumentedModule
39 from elasticapm.traces import DroppedSpan, execution_context
40 from elasticapm.utils.logging import get_logger
41 
42 logger = get_logger("elasticapm.instrument")
43 
44 should_capture_body_re = re.compile("/(_search|_msearch|_count|_async_search|_sql|_eql)(/|$)")
45 
46 
47 class ElasticsearchConnectionInstrumentation(AbstractInstrumentedModule):
48     name = "elasticsearch_connection"
49 
50     def get_instrument_list(self):
51         try:
52             import elastic_transport  # noqa: F401
53 
54             return [
55                 ("elastic_transport._node._http_urllib3", "Urllib3HttpNode.perform_request"),
56                 ("elastic_transport._node._http_requests", "RequestsHttpNode.perform_request"),
57             ]
58         except ImportError:
59             return [
60                 ("elasticsearch.connection.http_urllib3", "Urllib3HttpConnection.perform_request"),
61                 ("elasticsearch.connection.http_requests", "RequestsHttpConnection.perform_request"),
62             ]
63 
64     def call(self, module, method, wrapped, instance, args, kwargs):
65         span = execution_context.get_span()
66         if not span or isinstance(span, DroppedSpan):
67             return wrapped(*args, **kwargs)
68 
69         self._update_context_by_request_data(span.context, instance, args, kwargs)
70 
71         result = wrapped(*args, **kwargs)
72         if hasattr(result, "meta"):  # elasticsearch-py 8.x+
73             status_code = result.meta.status
74         else:
75             status_code = result[0]
76         span.context["http"] = {"status_code": status_code}
77 
78         return result
79 
80     def _update_context_by_request_data(self, context, instance, args, kwargs):
81         args_len = len(args)
82         url = args[1] if args_len > 1 else kwargs.get("url")
83         params = args[2] if args_len > 2 else kwargs.get("params")
84         body_serialized = args[3] if args_len > 3 else kwargs.get("body")
85 
86         if "?" in url and not params:
87             url, qs = url.split("?", 1)
88             params = {k: v[0] for k, v in parse_qs(qs).items()}
89 
90         should_capture_body = bool(should_capture_body_re.search(url))
91 
92         context["db"] = {"type": "elasticsearch"}
93         if should_capture_body:
94             query = []
95             # using both q AND body is allowed in some API endpoints / ES versions,
96             # but not in others. We simply capture both if they are there so the
97             # user can see it.
98             if params and "q" in params:
99                 # 'q' may already be encoded to a byte string at this point.
100                 # We assume utf8, which is the default
101                 q = params["q"]
102                 if isinstance(q, bytes):
103                     q = q.decode("utf-8", errors="replace")
104                 query.append("q=" + q)
105             if body_serialized:
106                 if isinstance(body_serialized, bytes):
107                     query.append(body_serialized.decode("utf-8", errors="replace"))
108                 else:
109                     query.append(body_serialized)
110             if query:
111                 context["db"]["statement"] = "\n\n".join(query)
112 
113         # ES5: `host` is URL, no `port` attribute
114         # ES6, ES7: `host` URL, `hostname` is host, `port` is port
115         # ES8: `host` is hostname, no `hostname` attribute, `port` is `port`
116         if not hasattr(instance, "port"):
117             # ES5, parse hostname and port from URL stored in `host`
118             parsed_url = urlparse(instance.host)
119             host = parsed_url.hostname
120             port = parsed_url.port
121         elif not hasattr(instance, "hostname"):
122             # ES8 (and up, one can hope)
123             host = instance.host
124             port = instance.port
125         else:
126             # ES6, ES7
127             host = instance.hostname
128             port = instance.port
129 
130         context["destination"] = {"address": host, "port": port}
131 
132 
133 class ElasticsearchTransportInstrumentation(AbstractInstrumentedModule):
134     name = "elasticsearch_connection"
135 
136     def get_instrument_list(self):
137         try:
138             import elastic_transport  # noqa: F401
139 
140             return [
141                 ("elastic_transport", "Transport.perform_request"),
142             ]
143         except ImportError:
144             return [
145                 ("elasticsearch.transport", "Transport.perform_request"),
146             ]
147 
148     def call(self, module, method, wrapped, instance, args, kwargs):
149         with elasticapm.capture_span(
150             self._get_signature(args, kwargs),
151             span_type="db",
152             span_subtype="elasticsearch",
153             span_action="query",
154             extra={},
155             skip_frames=2,
156             leaf=True,
157         ) as span:
158             result_data = wrapped(*args, **kwargs)
159 
160             hits = self._get_hits(result_data)
161             if hits:
162                 span.context["db"]["rows_affected"] = hits
163 
164             return result_data
165 
166     def _get_signature(self, args, kwargs):
167         args_len = len(args)
168         http_method = args[0] if args_len else kwargs.get("method")
169         http_path = args[1] if args_len > 1 else kwargs.get("url")
170         http_path = http_path.split("?", 1)[0]  # we don't want to capture a potential query string in the span name
171 
172         return "ES %s %s" % (http_method, http_path)
173 
174     def _get_hits(self, result) -> Optional[int]:
175         if getattr(result, "body", None) and "hits" in result.body:  # ES >= 8
176             return result.body["hits"]["total"]["value"]
177         elif isinstance(result, dict) and "hits" in result:
178             return (
179                 result["hits"]["total"]["value"]
180                 if isinstance(result["hits"]["total"], dict)
181                 else result["hits"]["total"]
182             )
183
```
--- 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/elasticapm/instrumentation/packages/elasticsearch.py b/elasticapm/instrumentation/packages/elasticsearch.py
--- a/elasticapm/instrumentation/packages/elasticsearch.py
+++ b/elasticapm/instrumentation/packages/elasticsearch.py
@@ -71,9 +71,13 @@
         result = wrapped(*args, **kwargs)
         if hasattr(result, "meta"):  # elasticsearch-py 8.x+
             status_code = result.meta.status
+            cluster = result.meta.headers.get("x-found-handling-cluster")
         else:
             status_code = result[0]
+            cluster = result[1].get("x-found-handling-cluster")
         span.context["http"] = {"status_code": status_code}
+        if cluster:
+            span.context["db"] = {"instance": cluster}
 
         return result

{"golden_diff": "diff --git a/elasticapm/instrumentation/packages/elasticsearch.py b/elasticapm/instrumentation/packages/elasticsearch.py\n--- a/elasticapm/instrumentation/packages/elasticsearch.py\n+++ b/elasticapm/instrumentation/packages/elasticsearch.py\n@@ -71,9 +71,13 @@\n         result = wrapped(*args, **kwargs)\n         if hasattr(result, \"meta\"):  # elasticsearch-py 8.x+\n             status_code = result.meta.status\n+            cluster = result.meta.headers.get(\"x-found-handling-cluster\")\n         else:\n             status_code = result[0]\n+            cluster = result[1].get(\"x-found-handling-cluster\")\n         span.context[\"http\"] = {\"status_code\": status_code}\n+        if cluster:\n+            span.context[\"db\"] = {\"instance\": cluster}\n \n         return result\n", "issue": "[META 659] Backend dependencies granularity for NoSQL and Messaging\nSee meta issue for the description and details:\r\n- Meta issue: https://github.com/elastic/apm/issues/659\r\n\r\nThis is done for everything except for Elasticsearch.\r\n\n", "before_files": [{"content": "#  BSD 3-Clause License\n#\n#  Copyright (c) 2019, Elasticsearch BV\n#  All rights reserved.\n#\n#  Redistribution and use in source and binary forms, with or without\n#  modification, are permitted provided that the following conditions are met:\n#\n#  * Redistributions of source code must retain the above copyright notice, this\n#    list of conditions and the following disclaimer.\n#\n#  * Redistributions in binary form must reproduce the above copyright notice,\n#    this list of conditions and the following disclaimer in the documentation\n#    and/or other materials provided with the distribution.\n#\n#  * Neither the name of the copyright holder nor the names of its\n#    contributors may be used to endorse or promote products derived from\n#    this software without specific prior written permission.\n#\n#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE\n#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE\n#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL\n#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR\n#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER\n#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,\n#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n#  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n\nfrom __future__ import absolute_import\n\nimport re\nfrom typing import Optional\nfrom urllib.parse import parse_qs, urlparse\n\nimport elasticapm\nfrom elasticapm.instrumentation.packages.base import AbstractInstrumentedModule\nfrom elasticapm.traces import DroppedSpan, execution_context\nfrom elasticapm.utils.logging import get_logger\n\nlogger = get_logger(\"elasticapm.instrument\")\n\nshould_capture_body_re = re.compile(\"/(_search|_msearch|_count|_async_search|_sql|_eql)(/|$)\")\n\n\nclass ElasticsearchConnectionInstrumentation(AbstractInstrumentedModule):\n    name = \"elasticsearch_connection\"\n\n    def get_instrument_list(self):\n        try:\n            import elastic_transport  # noqa: F401\n\n            return [\n                (\"elastic_transport._node._http_urllib3\", \"Urllib3HttpNode.perform_request\"),\n                (\"elastic_transport._node._http_requests\", \"RequestsHttpNode.perform_request\"),\n            ]\n        except ImportError:\n            return [\n                (\"elasticsearch.connection.http_urllib3\", \"Urllib3HttpConnection.perform_request\"),\n                (\"elasticsearch.connection.http_requests\", \"RequestsHttpConnection.perform_request\"),\n            ]\n\n    def call(self, module, method, wrapped, instance, args, kwargs):\n        span = execution_context.get_span()\n        if not span or isinstance(span, DroppedSpan):\n            return wrapped(*args, **kwargs)\n\n        self._update_context_by_request_data(span.context, instance, args, kwargs)\n\n        result = wrapped(*args, **kwargs)\n        if hasattr(result, \"meta\"):  # elasticsearch-py 8.x+\n            status_code = result.meta.status\n        else:\n            status_code = result[0]\n        span.context[\"http\"] = {\"status_code\": status_code}\n\n        return result\n\n    def _update_context_by_request_data(self, context, instance, args, kwargs):\n        args_len = len(args)\n        url = args[1] if args_len > 1 else kwargs.get(\"url\")\n        params = args[2] if args_len > 2 else kwargs.get(\"params\")\n        body_serialized = args[3] if args_len > 3 else kwargs.get(\"body\")\n\n        if \"?\" in url and not params:\n            url, qs = url.split(\"?\", 1)\n            params = {k: v[0] for k, v in parse_qs(qs).items()}\n\n        should_capture_body = bool(should_capture_body_re.search(url))\n\n        context[\"db\"] = {\"type\": \"elasticsearch\"}\n        if should_capture_body:\n            query = []\n            # using both q AND body is allowed in some API endpoints / ES versions,\n            # but not in others. We simply capture both if they are there so the\n            # user can see it.\n            if params and \"q\" in params:\n                # 'q' may already be encoded to a byte string at this point.\n                # We assume utf8, which is the default\n                q = params[\"q\"]\n                if isinstance(q, bytes):\n                    q = q.decode(\"utf-8\", errors=\"replace\")\n                query.append(\"q=\" + q)\n            if body_serialized:\n                if isinstance(body_serialized, bytes):\n                    query.append(body_serialized.decode(\"utf-8\", errors=\"replace\"))\n                else:\n                    query.append(body_serialized)\n            if query:\n                context[\"db\"][\"statement\"] = \"\\n\\n\".join(query)\n\n        # ES5: `host` is URL, no `port` attribute\n        # ES6, ES7: `host` URL, `hostname` is host, `port` is port\n        # ES8: `host` is hostname, no `hostname` attribute, `port` is `port`\n        if not hasattr(instance, \"port\"):\n            # ES5, parse hostname and port from URL stored in `host`\n            parsed_url = urlparse(instance.host)\n            host = parsed_url.hostname\n            port = parsed_url.port\n        elif not hasattr(instance, \"hostname\"):\n            # ES8 (and up, one can hope)\n            host = instance.host\n            port = instance.port\n        else:\n            # ES6, ES7\n            host = instance.hostname\n            port = instance.port\n\n        context[\"destination\"] = {\"address\": host, \"port\": port}\n\n\nclass ElasticsearchTransportInstrumentation(AbstractInstrumentedModule):\n    name = \"elasticsearch_connection\"\n\n    def get_instrument_list(self):\n        try:\n            import elastic_transport  # noqa: F401\n\n            return [\n                (\"elastic_transport\", \"Transport.perform_request\"),\n            ]\n        except ImportError:\n            return [\n                (\"elasticsearch.transport\", \"Transport.perform_request\"),\n            ]\n\n    def call(self, module, method, wrapped, instance, args, kwargs):\n        with elasticapm.capture_span(\n            self._get_signature(args, kwargs),\n            span_type=\"db\",\n            span_subtype=\"elasticsearch\",\n            span_action=\"query\",\n            extra={},\n            skip_frames=2,\n            leaf=True,\n        ) as span:\n            result_data = wrapped(*args, **kwargs)\n\n            hits = self._get_hits(result_data)\n            if hits:\n                span.context[\"db\"][\"rows_affected\"] = hits\n\n            return result_data\n\n    def _get_signature(self, args, kwargs):\n        args_len = len(args)\n        http_method = args[0] if args_len else kwargs.get(\"method\")\n        http_path = args[1] if args_len > 1 else kwargs.get(\"url\")\n        http_path = http_path.split(\"?\", 1)[0]  # we don't want to capture a potential query string in the span name\n\n        return \"ES %s %s\" % (http_method, http_path)\n\n    def _get_hits(self, result) -> Optional[int]:\n        if getattr(result, \"body\", None) and \"hits\" in result.body:  # ES >= 8\n            return result.body[\"hits\"][\"total\"][\"value\"]\n        elif isinstance(result, dict) and \"hits\" in result:\n            return (\n                result[\"hits\"][\"total\"][\"value\"]\n                if isinstance(result[\"hits\"][\"total\"], dict)\n                else result[\"hits\"][\"total\"]\n            )\n", "path": "elasticapm/instrumentation/packages/elasticsearch.py"}], "after_files": [{"content": "#  BSD 3-Clause License\n#\n#  Copyright (c) 2019, Elasticsearch BV\n#  All rights reserved.\n#\n#  Redistribution and use in source and binary forms, with or without\n#  modification, are permitted provided that the following conditions are met:\n#\n#  * Redistributions of source code must retain the above copyright notice, this\n#    list of conditions and the following disclaimer.\n#\n#  * Redistributions in binary form must reproduce the above copyright notice,\n#    this list of conditions and the following disclaimer in the documentation\n#    and/or other materials provided with the distribution.\n#\n#  * Neither the name of the copyright holder nor the names of its\n#    contributors may be used to endorse or promote products derived from\n#    this software without specific prior written permission.\n#\n#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE\n#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE\n#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL\n#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR\n#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER\n#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,\n#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n#  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n\nfrom __future__ import absolute_import\n\nimport re\nfrom typing import Optional\nfrom urllib.parse import parse_qs, urlparse\n\nimport elasticapm\nfrom elasticapm.instrumentation.packages.base import AbstractInstrumentedModule\nfrom elasticapm.traces import DroppedSpan, execution_context\nfrom elasticapm.utils.logging import get_logger\n\nlogger = get_logger(\"elasticapm.instrument\")\n\nshould_capture_body_re = re.compile(\"/(_search|_msearch|_count|_async_search|_sql|_eql)(/|$)\")\n\n\nclass ElasticsearchConnectionInstrumentation(AbstractInstrumentedModule):\n    name = \"elasticsearch_connection\"\n\n    def get_instrument_list(self):\n        try:\n            import elastic_transport  # noqa: F401\n\n            return [\n                (\"elastic_transport._node._http_urllib3\", \"Urllib3HttpNode.perform_request\"),\n                (\"elastic_transport._node._http_requests\", \"RequestsHttpNode.perform_request\"),\n            ]\n        except ImportError:\n            return [\n                (\"elasticsearch.connection.http_urllib3\", \"Urllib3HttpConnection.perform_request\"),\n                (\"elasticsearch.connection.http_requests\", \"RequestsHttpConnection.perform_request\"),\n            ]\n\n    def call(self, module, method, wrapped, instance, args, kwargs):\n        span = execution_context.get_span()\n        if not span or isinstance(span, DroppedSpan):\n            return wrapped(*args, **kwargs)\n\n        self._update_context_by_request_data(span.context, instance, args, kwargs)\n\n        result = wrapped(*args, **kwargs)\n        if hasattr(result, \"meta\"):  # elasticsearch-py 8.x+\n            status_code = result.meta.status\n            cluster = result.meta.headers.get(\"x-found-handling-cluster\")\n        else:\n            status_code = result[0]\n            cluster = result[1].get(\"x-found-handling-cluster\")\n        span.context[\"http\"] = {\"status_code\": status_code}\n        if cluster:\n            span.context[\"db\"] = {\"instance\": cluster}\n\n        return result\n\n    def _update_context_by_request_data(self, context, instance, args, kwargs):\n        args_len = len(args)\n        url = args[1] if args_len > 1 else kwargs.get(\"url\")\n        params = args[2] if args_len > 2 else kwargs.get(\"params\")\n        body_serialized = args[3] if args_len > 3 else kwargs.get(\"body\")\n\n        if \"?\" in url and not params:\n            url, qs = url.split(\"?\", 1)\n            params = {k: v[0] for k, v in parse_qs(qs).items()}\n\n        should_capture_body = bool(should_capture_body_re.search(url))\n\n        context[\"db\"] = {\"type\": \"elasticsearch\"}\n        if should_capture_body:\n            query = []\n            # using both q AND body is allowed in some API endpoints / ES versions,\n            # but not in others. We simply capture both if they are there so the\n            # user can see it.\n            if params and \"q\" in params:\n                # 'q' may already be encoded to a byte string at this point.\n                # We assume utf8, which is the default\n                q = params[\"q\"]\n                if isinstance(q, bytes):\n                    q = q.decode(\"utf-8\", errors=\"replace\")\n                query.append(\"q=\" + q)\n            if body_serialized:\n                if isinstance(body_serialized, bytes):\n                    query.append(body_serialized.decode(\"utf-8\", errors=\"replace\"))\n                else:\n                    query.append(body_serialized)\n            if query:\n                context[\"db\"][\"statement\"] = \"\\n\\n\".join(query)\n\n        # ES5: `host` is URL, no `port` attribute\n        # ES6, ES7: `host` URL, `hostname` is host, `port` is port\n        # ES8: `host` is hostname, no `hostname` attribute, `port` is `port`\n        if not hasattr(instance, \"port\"):\n            # ES5, parse hostname and port from URL stored in `host`\n            parsed_url = urlparse(instance.host)\n            host = parsed_url.hostname\n            port = parsed_url.port\n        elif not hasattr(instance, \"hostname\"):\n            # ES8 (and up, one can hope)\n            host = instance.host\n            port = instance.port\n        else:\n            # ES6, ES7\n            host = instance.hostname\n            port = instance.port\n\n        context[\"destination\"] = {\"address\": host, \"port\": port}\n\n\nclass ElasticsearchTransportInstrumentation(AbstractInstrumentedModule):\n    name = \"elasticsearch_connection\"\n\n    def get_instrument_list(self):\n        try:\n            import elastic_transport  # noqa: F401\n\n            return [\n                (\"elastic_transport\", \"Transport.perform_request\"),\n            ]\n        except ImportError:\n            return [\n                (\"elasticsearch.transport\", \"Transport.perform_request\"),\n            ]\n\n    def call(self, module, method, wrapped, instance, args, kwargs):\n        with elasticapm.capture_span(\n            self._get_signature(args, kwargs),\n            span_type=\"db\",\n            span_subtype=\"elasticsearch\",\n            span_action=\"query\",\n            extra={},\n            skip_frames=2,\n            leaf=True,\n        ) as span:\n            result_data = wrapped(*args, **kwargs)\n\n            hits = self._get_hits(result_data)\n            if hits:\n                span.context[\"db\"][\"rows_affected\"] = hits\n\n            return result_data\n\n    def _get_signature(self, args, kwargs):\n        args_len = len(args)\n        http_method = args[0] if args_len else kwargs.get(\"method\")\n        http_path = args[1] if args_len > 1 else kwargs.get(\"url\")\n        http_path = http_path.split(\"?\", 1)[0]  # we don't want to capture a potential query string in the span name\n\n        return \"ES %s %s\" % (http_method, http_path)\n\n    def _get_hits(self, result) -> Optional[int]:\n        if getattr(result, \"body\", None) and \"hits\" in result.body:  # ES >= 8\n            return result.body[\"hits\"][\"total\"][\"value\"]\n        elif isinstance(result, dict) and \"hits\" in result:\n            return (\n                result[\"hits\"][\"total\"][\"value\"]\n                if isinstance(result[\"hits\"][\"total\"], dict)\n                else result[\"hits\"][\"total\"]\n            )\n", "path": "elasticapm/instrumentation/packages/elasticsearch.py"}]}