We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
KeyError: 'db' for DroppedSpan when running scan query on Elasticsearch
**Describe the bug**: ...

Elastic APM fails with a `KeyError: db key not found`.

The application where this happens is a Django project that stores/reads data from Elasticsearch. I have APM enable (APM server and Elasticsearch cluster are both running on Elastic Cloud). The library fails with the aforementioned error (shown in the snippet and screenshots below) while running a scan query on Elasticsearch. It looks like it's dropping some spans, which ends up in this case:

```
            hits = self._get_hits(result_data)
            if hits:
                span.context["db"]["rows_affected"] = hits
 
            return result_data
```

here's a screenshot of what I see on the APM Error page:

![image](https://user-images.githubusercontent.com/13816/200343191-fcff09cb-17f0-4f95-90db-f49fd1ab632d.png)

Few variables from the context:

![image](https://user-images.githubusercontent.com/13816/200343396-b95e0f90-a62e-4a4d-ad5d-87554b2a1d53.png)


**To Reproduce**

Unfortunately, I don't have a reproducible snippet. 

**Environment (please complete the following information)**
- OS: Linux (containerized)
- Python version: 3.9.15
- Framework and version [e.g. Django 2.1]:
- APM Server version: 7.17.4
- Agent version: 

```
$ pip freeze | grep elastic
django-elasticsearch-dsl==7.2.2
elastic-apm==6.13.1
elasticsearch==7.17.4
elasticsearch-dsl==7.4.0
```


**Additional context**

Add any other context about the problem here.



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `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             cluster = result.meta.headers.get("x-found-handling-cluster")
75         else:
76             status_code = result[0]
77             cluster = result[1].get("x-found-handling-cluster")
78         span.context["http"] = {"status_code": status_code}
79         if cluster:
80             span.context["db"] = {"instance": cluster}
81 
82         return result
83 
84     def _update_context_by_request_data(self, context, instance, args, kwargs):
85         args_len = len(args)
86         url = args[1] if args_len > 1 else kwargs.get("url")
87         params = args[2] if args_len > 2 else kwargs.get("params")
88         body_serialized = args[3] if args_len > 3 else kwargs.get("body")
89 
90         if "?" in url and not params:
91             url, qs = url.split("?", 1)
92             params = {k: v[0] for k, v in parse_qs(qs).items()}
93 
94         should_capture_body = bool(should_capture_body_re.search(url))
95 
96         context["db"] = {"type": "elasticsearch"}
97         if should_capture_body:
98             query = []
99             # using both q AND body is allowed in some API endpoints / ES versions,
100             # but not in others. We simply capture both if they are there so the
101             # user can see it.
102             if params and "q" in params:
103                 # 'q' may already be encoded to a byte string at this point.
104                 # We assume utf8, which is the default
105                 q = params["q"]
106                 if isinstance(q, bytes):
107                     q = q.decode("utf-8", errors="replace")
108                 query.append("q=" + q)
109             if body_serialized:
110                 if isinstance(body_serialized, bytes):
111                     query.append(body_serialized.decode("utf-8", errors="replace"))
112                 else:
113                     query.append(body_serialized)
114             if query:
115                 context["db"]["statement"] = "\n\n".join(query)
116 
117         # ES5: `host` is URL, no `port` attribute
118         # ES6, ES7: `host` URL, `hostname` is host, `port` is port
119         # ES8: `host` is hostname, no `hostname` attribute, `port` is `port`
120         if not hasattr(instance, "port"):
121             # ES5, parse hostname and port from URL stored in `host`
122             parsed_url = urlparse(instance.host)
123             host = parsed_url.hostname
124             port = parsed_url.port
125         elif not hasattr(instance, "hostname"):
126             # ES8 (and up, one can hope)
127             host = instance.host
128             port = instance.port
129         else:
130             # ES6, ES7
131             host = instance.hostname
132             port = instance.port
133 
134         context["destination"] = {"address": host, "port": port}
135 
136 
137 class ElasticsearchTransportInstrumentation(AbstractInstrumentedModule):
138     name = "elasticsearch_connection"
139 
140     def get_instrument_list(self):
141         try:
142             import elastic_transport  # noqa: F401
143 
144             return [
145                 ("elastic_transport", "Transport.perform_request"),
146             ]
147         except ImportError:
148             return [
149                 ("elasticsearch.transport", "Transport.perform_request"),
150             ]
151 
152     def call(self, module, method, wrapped, instance, args, kwargs):
153         with elasticapm.capture_span(
154             self._get_signature(args, kwargs),
155             span_type="db",
156             span_subtype="elasticsearch",
157             span_action="query",
158             extra={},
159             skip_frames=2,
160             leaf=True,
161         ) as span:
162             result_data = wrapped(*args, **kwargs)
163 
164             hits = self._get_hits(result_data)
165             if hits:
166                 span.context["db"]["rows_affected"] = hits
167 
168             return result_data
169 
170     def _get_signature(self, args, kwargs):
171         args_len = len(args)
172         http_method = args[0] if args_len else kwargs.get("method")
173         http_path = args[1] if args_len > 1 else kwargs.get("url")
174         http_path = http_path.split("?", 1)[0]  # we don't want to capture a potential query string in the span name
175 
176         return "ES %s %s" % (http_method, http_path)
177 
178     def _get_hits(self, result) -> Optional[int]:
179         if getattr(result, "body", None) and "hits" in result.body:  # ES >= 8
180             return result.body["hits"].get("total", {}).get("value")
181         elif isinstance(result, dict) and "hits" in result and "total" in result["hits"]:
182             return (
183                 result["hits"]["total"]["value"]
184                 if isinstance(result["hits"]["total"], dict)
185                 else result["hits"]["total"]
186             )
187
```
--- 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
@@ -163,7 +163,7 @@
 
             hits = self._get_hits(result_data)
             if hits:
-                span.context["db"]["rows_affected"] = hits
+                span.update_context("db", {"rows_affected": hits})
 
             return result_data

{"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@@ -163,7 +163,7 @@\n \n             hits = self._get_hits(result_data)\n             if hits:\n-                span.context[\"db\"][\"rows_affected\"] = hits\n+                span.update_context(\"db\", {\"rows_affected\": hits})\n \n             return result_data\n", "issue": "KeyError: 'db' for DroppedSpan when running scan query on Elasticsearch\n**Describe the bug**: ...\r\n\r\nElastic APM fails with a `KeyError: db key not found`.\r\n\r\nThe application where this happens is a Django project that stores/reads data from Elasticsearch. I have APM enable (APM server and Elasticsearch cluster are both running on Elastic Cloud). The library fails with the aforementioned error (shown in the snippet and screenshots below) while running a scan query on Elasticsearch. It looks like it's dropping some spans, which ends up in this case:\r\n\r\n```\r\n            hits = self._get_hits(result_data)\r\n            if hits:\r\n                span.context[\"db\"][\"rows_affected\"] = hits\r\n \r\n            return result_data\r\n```\r\n\r\nhere's a screenshot of what I see on the APM Error page:\r\n\r\n![image](https://user-images.githubusercontent.com/13816/200343191-fcff09cb-17f0-4f95-90db-f49fd1ab632d.png)\r\n\r\nFew variables from the context:\r\n\r\n![image](https://user-images.githubusercontent.com/13816/200343396-b95e0f90-a62e-4a4d-ad5d-87554b2a1d53.png)\r\n\r\n\r\n**To Reproduce**\r\n\r\nUnfortunately, I don't have a reproducible snippet. \r\n\r\n**Environment (please complete the following information)**\r\n- OS: Linux (containerized)\r\n- Python version: 3.9.15\r\n- Framework and version [e.g. Django 2.1]:\r\n- APM Server version: 7.17.4\r\n- Agent version: \r\n\r\n```\r\n$ pip freeze | grep elastic\r\ndjango-elasticsearch-dsl==7.2.2\r\nelastic-apm==6.13.1\r\nelasticsearch==7.17.4\r\nelasticsearch-dsl==7.4.0\r\n```\r\n\r\n\r\n**Additional context**\r\n\r\nAdd any other context about the problem here.\r\n\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            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\"].get(\"total\", {}).get(\"value\")\n        elif isinstance(result, dict) and \"hits\" in result and \"total\" in result[\"hits\"]:\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.update_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\"].get(\"total\", {}).get(\"value\")\n        elif isinstance(result, dict) and \"hits\" in result and \"total\" in result[\"hits\"]:\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"}]}