We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Update instrumentation of SQS for batch send, delete, and batch delete operations
The SQS API also includes batch send, delete, and batch delete messages. These actions were not described in the original SQS spec so the instrumentation needs to be updated as per the description in this PR https://github.com/elastic/apm/pull/419/files

--- 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/botocore.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 import urllib.parse
32 from collections import namedtuple
33 
34 from elasticapm.conf import constants
35 from elasticapm.instrumentation.packages.base import AbstractInstrumentedModule
36 from elasticapm.traces import capture_span, execution_context
37 from elasticapm.utils.logging import get_logger
38 
39 logger = get_logger("elasticapm.instrument")
40 
41 SQS_MAX_ATTRIBUTES = 10
42 
43 
44 HandlerInfo = namedtuple("HandlerInfo", ("signature", "span_type", "span_subtype", "span_action", "context"))
45 
46 # Used for boto3 < 1.7
47 endpoint_to_service_id = {"SNS": "SNS", "S3": "S3", "DYNAMODB": "DynamoDB", "SQS": "SQS"}
48 
49 
50 class BotocoreInstrumentation(AbstractInstrumentedModule):
51     name = "botocore"
52 
53     instrument_list = [("botocore.client", "BaseClient._make_api_call")]
54 
55     capture_span_ctx = capture_span
56 
57     def _call(self, service, instance, args, kwargs):
58         """
59         This is split out from `call()` so that it can be re-used by the
60         aiobotocore instrumentation without duplicating all of this code.
61         """
62         if "operation_name" in kwargs:
63             operation_name = kwargs["operation_name"]
64         else:
65             operation_name = args[0]
66 
67         parsed_url = urllib.parse.urlparse(instance.meta.endpoint_url)
68         context = {
69             "destination": {
70                 "address": parsed_url.hostname,
71                 "port": parsed_url.port,
72                 "cloud": {"region": instance.meta.region_name},
73             }
74         }
75 
76         handler_info = None
77         handler = handlers.get(service, False)
78         if handler:
79             handler_info = handler(operation_name, service, instance, args, kwargs, context)
80         if not handler_info:
81             handler_info = handle_default(operation_name, service, instance, args, kwargs, context)
82 
83         return self.capture_span_ctx(
84             handler_info.signature,
85             span_type=handler_info.span_type,
86             leaf=True,
87             span_subtype=handler_info.span_subtype,
88             span_action=handler_info.span_action,
89             extra=handler_info.context,
90         )
91 
92     def _get_service(self, instance):
93         service_model = instance.meta.service_model
94         if hasattr(service_model, "service_id"):  # added in boto3 1.7
95             service = service_model.service_id
96         else:
97             service = service_model.service_name.upper()
98             service = endpoint_to_service_id.get(service, service)
99         return service
100 
101     def call(self, module, method, wrapped, instance, args, kwargs):
102         service = self._get_service(instance)
103 
104         ctx = self._call(service, instance, args, kwargs)
105         with ctx as span:
106             if service in span_modifiers:
107                 span_modifiers[service](span, args, kwargs)
108             return wrapped(*args, **kwargs)
109 
110 
111 def handle_s3(operation_name, service, instance, args, kwargs, context):
112     span_type = "storage"
113     span_subtype = "s3"
114     span_action = operation_name
115     if len(args) > 1 and "Bucket" in args[1]:
116         bucket = args[1]["Bucket"]
117     else:
118         # TODO handle Access Points
119         bucket = ""
120     signature = f"S3 {operation_name} {bucket}"
121 
122     context["destination"]["service"] = {"name": span_subtype, "resource": bucket, "type": span_type}
123 
124     return HandlerInfo(signature, span_type, span_subtype, span_action, context)
125 
126 
127 def handle_dynamodb(operation_name, service, instance, args, kwargs, context):
128     span_type = "db"
129     span_subtype = "dynamodb"
130     span_action = "query"
131     if len(args) > 1 and "TableName" in args[1]:
132         table = args[1]["TableName"]
133     else:
134         table = ""
135     signature = f"DynamoDB {operation_name} {table}".rstrip()
136 
137     context["db"] = {"type": "dynamodb", "instance": instance.meta.region_name}
138     if operation_name == "Query" and len(args) > 1 and "KeyConditionExpression" in args[1]:
139         context["db"]["statement"] = args[1]["KeyConditionExpression"]
140 
141     context["destination"]["service"] = {"name": span_subtype, "resource": table, "type": span_type}
142     return HandlerInfo(signature, span_type, span_subtype, span_action, context)
143 
144 
145 def handle_sns(operation_name, service, instance, args, kwargs, context):
146     if operation_name != "Publish":
147         # only "publish" is handled specifically, other endpoints get the default treatment
148         return False
149     span_type = "messaging"
150     span_subtype = "sns"
151     span_action = "send"
152     topic_name = ""
153     if len(args) > 1:
154         if "Name" in args[1]:
155             topic_name = args[1]["Name"]
156         if "TopicArn" in args[1]:
157             topic_name = args[1]["TopicArn"].rsplit(":", maxsplit=1)[-1]
158     signature = f"SNS {operation_name} {topic_name}".rstrip()
159     context["destination"]["service"] = {
160         "name": span_subtype,
161         "resource": f"{span_subtype}/{topic_name}" if topic_name else span_subtype,
162         "type": span_type,
163     }
164     return HandlerInfo(signature, span_type, span_subtype, span_action, context)
165 
166 
167 def handle_sqs(operation_name, service, instance, args, kwargs, context):
168     if operation_name not in ("SendMessage", "SendMessageBatch", "ReceiveMessage"):
169         # only "publish" is handled specifically, other endpoints get the default treatment
170         return False
171     span_type = "messaging"
172     span_subtype = "sqs"
173     span_action = "send" if operation_name in ("SendMessage", "SendMessageBatch") else "receive"
174     topic_name = ""
175     batch = "_BATCH" if operation_name == "SendMessageBatch" else ""
176     signature_type = "RECEIVE from" if span_action == "receive" else f"SEND{batch} to"
177 
178     if len(args) > 1:
179         topic_name = args[1]["QueueUrl"].rsplit("/", maxsplit=1)[-1]
180     signature = f"SQS {signature_type} {topic_name}".rstrip() if topic_name else f"SQS {signature_type}"
181     context["destination"]["service"] = {
182         "name": span_subtype,
183         "resource": f"{span_subtype}/{topic_name}" if topic_name else span_subtype,
184         "type": span_type,
185     }
186     return HandlerInfo(signature, span_type, span_subtype, span_action, context)
187 
188 
189 def modify_span_sqs(span, args, kwargs):
190     if span.id:
191         trace_parent = span.transaction.trace_parent.copy_from(span_id=span.id)
192     else:
193         # this is a dropped span, use transaction id instead
194         transaction = execution_context.get_transaction()
195         trace_parent = transaction.trace_parent.copy_from(span_id=transaction.id)
196     attributes = {constants.TRACEPARENT_HEADER_NAME: {"DataType": "String", "StringValue": trace_parent.to_string()}}
197     if trace_parent.tracestate:
198         attributes[constants.TRACESTATE_HEADER_NAME] = {"DataType": "String", "StringValue": trace_parent.tracestate}
199     if len(args) > 1:
200         attributes_count = len(attributes)
201         if "MessageAttributes" in args[1]:
202             messages = [args[1]]
203         elif "Entries" in args[1]:
204             messages = args[1]["Entries"]
205         else:
206             messages = []
207         for message in messages:
208             if len(message["MessageAttributes"]) + attributes_count <= SQS_MAX_ATTRIBUTES:
209                 message["MessageAttributes"].update(attributes)
210             else:
211                 logger.info("Not adding disttracing headers to message due to attribute limit reached")
212 
213 
214 def handle_default(operation_name, service, instance, args, kwargs, destination):
215     span_type = "aws"
216     span_subtype = service.lower()
217     span_action = operation_name
218 
219     destination["service"] = {"name": span_subtype, "resource": span_subtype, "type": span_type}
220 
221     signature = f"{service}:{operation_name}"
222     return HandlerInfo(signature, span_type, span_subtype, span_action, destination)
223 
224 
225 handlers = {
226     "S3": handle_s3,
227     "DynamoDB": handle_dynamodb,
228     "SNS": handle_sns,
229     "SQS": handle_sqs,
230     "default": handle_default,
231 }
232 
233 span_modifiers = {
234     "SQS": modify_span_sqs,
235 }
236
```
--- 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/botocore.py b/elasticapm/instrumentation/packages/botocore.py
--- a/elasticapm/instrumentation/packages/botocore.py
+++ b/elasticapm/instrumentation/packages/botocore.py
@@ -164,26 +164,33 @@
     return HandlerInfo(signature, span_type, span_subtype, span_action, context)
 
 
+SQS_OPERATIONS = {
+    "SendMessage": {"span_action": "send", "signature": "SEND to"},
+    "SendMessageBatch": {"span_action": "send_batch", "signature": "SEND_BATCH to"},
+    "ReceiveMessage": {"span_action": "receive", "signature": "RECEIVE from"},
+    "DeleteMessage": {"span_action": "delete", "signature": "DELETE from"},
+    "DeleteMessageBatch": {"span_action": "delete_batch", "signature": "DELETE_BATCH from"},
+}
+
+
 def handle_sqs(operation_name, service, instance, args, kwargs, context):
-    if operation_name not in ("SendMessage", "SendMessageBatch", "ReceiveMessage"):
+    op = SQS_OPERATIONS.get(operation_name, None)
+    if not op:
         # only "publish" is handled specifically, other endpoints get the default treatment
         return False
     span_type = "messaging"
     span_subtype = "sqs"
-    span_action = "send" if operation_name in ("SendMessage", "SendMessageBatch") else "receive"
     topic_name = ""
-    batch = "_BATCH" if operation_name == "SendMessageBatch" else ""
-    signature_type = "RECEIVE from" if span_action == "receive" else f"SEND{batch} to"
 
     if len(args) > 1:
         topic_name = args[1]["QueueUrl"].rsplit("/", maxsplit=1)[-1]
-    signature = f"SQS {signature_type} {topic_name}".rstrip() if topic_name else f"SQS {signature_type}"
+    signature = f"SQS {op['signature']} {topic_name}".rstrip() if topic_name else f"SQS {op['signature']}"
     context["destination"]["service"] = {
         "name": span_subtype,
         "resource": f"{span_subtype}/{topic_name}" if topic_name else span_subtype,
         "type": span_type,
     }
-    return HandlerInfo(signature, span_type, span_subtype, span_action, context)
+    return HandlerInfo(signature, span_type, span_subtype, op["span_action"], context)
 
 
 def modify_span_sqs(span, args, kwargs):
@@ -200,7 +207,9 @@
         attributes_count = len(attributes)
         if "MessageAttributes" in args[1]:
             messages = [args[1]]
-        elif "Entries" in args[1]:
+        # both send_batch and delete_batch use the same "Entries" list. We only want to add the
+        # traceparent to send_batch. We use the existence of ReceiptHandle to differentiate between the two
+        elif "Entries" in args[1] and args[1]["Entries"] and "ReceiptHandle" not in args[1]["Entries"][0]:
             messages = args[1]["Entries"]
         else:
             messages = []

{"golden_diff": "diff --git a/elasticapm/instrumentation/packages/botocore.py b/elasticapm/instrumentation/packages/botocore.py\n--- a/elasticapm/instrumentation/packages/botocore.py\n+++ b/elasticapm/instrumentation/packages/botocore.py\n@@ -164,26 +164,33 @@\n     return HandlerInfo(signature, span_type, span_subtype, span_action, context)\n \n \n+SQS_OPERATIONS = {\n+    \"SendMessage\": {\"span_action\": \"send\", \"signature\": \"SEND to\"},\n+    \"SendMessageBatch\": {\"span_action\": \"send_batch\", \"signature\": \"SEND_BATCH to\"},\n+    \"ReceiveMessage\": {\"span_action\": \"receive\", \"signature\": \"RECEIVE from\"},\n+    \"DeleteMessage\": {\"span_action\": \"delete\", \"signature\": \"DELETE from\"},\n+    \"DeleteMessageBatch\": {\"span_action\": \"delete_batch\", \"signature\": \"DELETE_BATCH from\"},\n+}\n+\n+\n def handle_sqs(operation_name, service, instance, args, kwargs, context):\n-    if operation_name not in (\"SendMessage\", \"SendMessageBatch\", \"ReceiveMessage\"):\n+    op = SQS_OPERATIONS.get(operation_name, None)\n+    if not op:\n         # only \"publish\" is handled specifically, other endpoints get the default treatment\n         return False\n     span_type = \"messaging\"\n     span_subtype = \"sqs\"\n-    span_action = \"send\" if operation_name in (\"SendMessage\", \"SendMessageBatch\") else \"receive\"\n     topic_name = \"\"\n-    batch = \"_BATCH\" if operation_name == \"SendMessageBatch\" else \"\"\n-    signature_type = \"RECEIVE from\" if span_action == \"receive\" else f\"SEND{batch} to\"\n \n     if len(args) > 1:\n         topic_name = args[1][\"QueueUrl\"].rsplit(\"/\", maxsplit=1)[-1]\n-    signature = f\"SQS {signature_type} {topic_name}\".rstrip() if topic_name else f\"SQS {signature_type}\"\n+    signature = f\"SQS {op['signature']} {topic_name}\".rstrip() if topic_name else f\"SQS {op['signature']}\"\n     context[\"destination\"][\"service\"] = {\n         \"name\": span_subtype,\n         \"resource\": f\"{span_subtype}/{topic_name}\" if topic_name else span_subtype,\n         \"type\": span_type,\n     }\n-    return HandlerInfo(signature, span_type, span_subtype, span_action, context)\n+    return HandlerInfo(signature, span_type, span_subtype, op[\"span_action\"], context)\n \n \n def modify_span_sqs(span, args, kwargs):\n@@ -200,7 +207,9 @@\n         attributes_count = len(attributes)\n         if \"MessageAttributes\" in args[1]:\n             messages = [args[1]]\n-        elif \"Entries\" in args[1]:\n+        # both send_batch and delete_batch use the same \"Entries\" list. We only want to add the\n+        # traceparent to send_batch. We use the existence of ReceiptHandle to differentiate between the two\n+        elif \"Entries\" in args[1] and args[1][\"Entries\"] and \"ReceiptHandle\" not in args[1][\"Entries\"][0]:\n             messages = args[1][\"Entries\"]\n         else:\n             messages = []\n", "issue": "Update instrumentation of SQS for batch send, delete, and batch delete operations\nThe SQS API also includes batch send, delete, and batch delete messages. These actions were not described in the original SQS spec so the instrumentation needs to be updated as per the description in this PR https://github.com/elastic/apm/pull/419/files\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\nimport urllib.parse\nfrom collections import namedtuple\n\nfrom elasticapm.conf import constants\nfrom elasticapm.instrumentation.packages.base import AbstractInstrumentedModule\nfrom elasticapm.traces import capture_span, execution_context\nfrom elasticapm.utils.logging import get_logger\n\nlogger = get_logger(\"elasticapm.instrument\")\n\nSQS_MAX_ATTRIBUTES = 10\n\n\nHandlerInfo = namedtuple(\"HandlerInfo\", (\"signature\", \"span_type\", \"span_subtype\", \"span_action\", \"context\"))\n\n# Used for boto3 < 1.7\nendpoint_to_service_id = {\"SNS\": \"SNS\", \"S3\": \"S3\", \"DYNAMODB\": \"DynamoDB\", \"SQS\": \"SQS\"}\n\n\nclass BotocoreInstrumentation(AbstractInstrumentedModule):\n    name = \"botocore\"\n\n    instrument_list = [(\"botocore.client\", \"BaseClient._make_api_call\")]\n\n    capture_span_ctx = capture_span\n\n    def _call(self, service, instance, args, kwargs):\n        \"\"\"\n        This is split out from `call()` so that it can be re-used by the\n        aiobotocore instrumentation without duplicating all of this code.\n        \"\"\"\n        if \"operation_name\" in kwargs:\n            operation_name = kwargs[\"operation_name\"]\n        else:\n            operation_name = args[0]\n\n        parsed_url = urllib.parse.urlparse(instance.meta.endpoint_url)\n        context = {\n            \"destination\": {\n                \"address\": parsed_url.hostname,\n                \"port\": parsed_url.port,\n                \"cloud\": {\"region\": instance.meta.region_name},\n            }\n        }\n\n        handler_info = None\n        handler = handlers.get(service, False)\n        if handler:\n            handler_info = handler(operation_name, service, instance, args, kwargs, context)\n        if not handler_info:\n            handler_info = handle_default(operation_name, service, instance, args, kwargs, context)\n\n        return self.capture_span_ctx(\n            handler_info.signature,\n            span_type=handler_info.span_type,\n            leaf=True,\n            span_subtype=handler_info.span_subtype,\n            span_action=handler_info.span_action,\n            extra=handler_info.context,\n        )\n\n    def _get_service(self, instance):\n        service_model = instance.meta.service_model\n        if hasattr(service_model, \"service_id\"):  # added in boto3 1.7\n            service = service_model.service_id\n        else:\n            service = service_model.service_name.upper()\n            service = endpoint_to_service_id.get(service, service)\n        return service\n\n    def call(self, module, method, wrapped, instance, args, kwargs):\n        service = self._get_service(instance)\n\n        ctx = self._call(service, instance, args, kwargs)\n        with ctx as span:\n            if service in span_modifiers:\n                span_modifiers[service](span, args, kwargs)\n            return wrapped(*args, **kwargs)\n\n\ndef handle_s3(operation_name, service, instance, args, kwargs, context):\n    span_type = \"storage\"\n    span_subtype = \"s3\"\n    span_action = operation_name\n    if len(args) > 1 and \"Bucket\" in args[1]:\n        bucket = args[1][\"Bucket\"]\n    else:\n        # TODO handle Access Points\n        bucket = \"\"\n    signature = f\"S3 {operation_name} {bucket}\"\n\n    context[\"destination\"][\"service\"] = {\"name\": span_subtype, \"resource\": bucket, \"type\": span_type}\n\n    return HandlerInfo(signature, span_type, span_subtype, span_action, context)\n\n\ndef handle_dynamodb(operation_name, service, instance, args, kwargs, context):\n    span_type = \"db\"\n    span_subtype = \"dynamodb\"\n    span_action = \"query\"\n    if len(args) > 1 and \"TableName\" in args[1]:\n        table = args[1][\"TableName\"]\n    else:\n        table = \"\"\n    signature = f\"DynamoDB {operation_name} {table}\".rstrip()\n\n    context[\"db\"] = {\"type\": \"dynamodb\", \"instance\": instance.meta.region_name}\n    if operation_name == \"Query\" and len(args) > 1 and \"KeyConditionExpression\" in args[1]:\n        context[\"db\"][\"statement\"] = args[1][\"KeyConditionExpression\"]\n\n    context[\"destination\"][\"service\"] = {\"name\": span_subtype, \"resource\": table, \"type\": span_type}\n    return HandlerInfo(signature, span_type, span_subtype, span_action, context)\n\n\ndef handle_sns(operation_name, service, instance, args, kwargs, context):\n    if operation_name != \"Publish\":\n        # only \"publish\" is handled specifically, other endpoints get the default treatment\n        return False\n    span_type = \"messaging\"\n    span_subtype = \"sns\"\n    span_action = \"send\"\n    topic_name = \"\"\n    if len(args) > 1:\n        if \"Name\" in args[1]:\n            topic_name = args[1][\"Name\"]\n        if \"TopicArn\" in args[1]:\n            topic_name = args[1][\"TopicArn\"].rsplit(\":\", maxsplit=1)[-1]\n    signature = f\"SNS {operation_name} {topic_name}\".rstrip()\n    context[\"destination\"][\"service\"] = {\n        \"name\": span_subtype,\n        \"resource\": f\"{span_subtype}/{topic_name}\" if topic_name else span_subtype,\n        \"type\": span_type,\n    }\n    return HandlerInfo(signature, span_type, span_subtype, span_action, context)\n\n\ndef handle_sqs(operation_name, service, instance, args, kwargs, context):\n    if operation_name not in (\"SendMessage\", \"SendMessageBatch\", \"ReceiveMessage\"):\n        # only \"publish\" is handled specifically, other endpoints get the default treatment\n        return False\n    span_type = \"messaging\"\n    span_subtype = \"sqs\"\n    span_action = \"send\" if operation_name in (\"SendMessage\", \"SendMessageBatch\") else \"receive\"\n    topic_name = \"\"\n    batch = \"_BATCH\" if operation_name == \"SendMessageBatch\" else \"\"\n    signature_type = \"RECEIVE from\" if span_action == \"receive\" else f\"SEND{batch} to\"\n\n    if len(args) > 1:\n        topic_name = args[1][\"QueueUrl\"].rsplit(\"/\", maxsplit=1)[-1]\n    signature = f\"SQS {signature_type} {topic_name}\".rstrip() if topic_name else f\"SQS {signature_type}\"\n    context[\"destination\"][\"service\"] = {\n        \"name\": span_subtype,\n        \"resource\": f\"{span_subtype}/{topic_name}\" if topic_name else span_subtype,\n        \"type\": span_type,\n    }\n    return HandlerInfo(signature, span_type, span_subtype, span_action, context)\n\n\ndef modify_span_sqs(span, args, kwargs):\n    if span.id:\n        trace_parent = span.transaction.trace_parent.copy_from(span_id=span.id)\n    else:\n        # this is a dropped span, use transaction id instead\n        transaction = execution_context.get_transaction()\n        trace_parent = transaction.trace_parent.copy_from(span_id=transaction.id)\n    attributes = {constants.TRACEPARENT_HEADER_NAME: {\"DataType\": \"String\", \"StringValue\": trace_parent.to_string()}}\n    if trace_parent.tracestate:\n        attributes[constants.TRACESTATE_HEADER_NAME] = {\"DataType\": \"String\", \"StringValue\": trace_parent.tracestate}\n    if len(args) > 1:\n        attributes_count = len(attributes)\n        if \"MessageAttributes\" in args[1]:\n            messages = [args[1]]\n        elif \"Entries\" in args[1]:\n            messages = args[1][\"Entries\"]\n        else:\n            messages = []\n        for message in messages:\n            if len(message[\"MessageAttributes\"]) + attributes_count <= SQS_MAX_ATTRIBUTES:\n                message[\"MessageAttributes\"].update(attributes)\n            else:\n                logger.info(\"Not adding disttracing headers to message due to attribute limit reached\")\n\n\ndef handle_default(operation_name, service, instance, args, kwargs, destination):\n    span_type = \"aws\"\n    span_subtype = service.lower()\n    span_action = operation_name\n\n    destination[\"service\"] = {\"name\": span_subtype, \"resource\": span_subtype, \"type\": span_type}\n\n    signature = f\"{service}:{operation_name}\"\n    return HandlerInfo(signature, span_type, span_subtype, span_action, destination)\n\n\nhandlers = {\n    \"S3\": handle_s3,\n    \"DynamoDB\": handle_dynamodb,\n    \"SNS\": handle_sns,\n    \"SQS\": handle_sqs,\n    \"default\": handle_default,\n}\n\nspan_modifiers = {\n    \"SQS\": modify_span_sqs,\n}\n", "path": "elasticapm/instrumentation/packages/botocore.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\nimport urllib.parse\nfrom collections import namedtuple\n\nfrom elasticapm.conf import constants\nfrom elasticapm.instrumentation.packages.base import AbstractInstrumentedModule\nfrom elasticapm.traces import capture_span, execution_context\nfrom elasticapm.utils.logging import get_logger\n\nlogger = get_logger(\"elasticapm.instrument\")\n\nSQS_MAX_ATTRIBUTES = 10\n\n\nHandlerInfo = namedtuple(\"HandlerInfo\", (\"signature\", \"span_type\", \"span_subtype\", \"span_action\", \"context\"))\n\n# Used for boto3 < 1.7\nendpoint_to_service_id = {\"SNS\": \"SNS\", \"S3\": \"S3\", \"DYNAMODB\": \"DynamoDB\", \"SQS\": \"SQS\"}\n\n\nclass BotocoreInstrumentation(AbstractInstrumentedModule):\n    name = \"botocore\"\n\n    instrument_list = [(\"botocore.client\", \"BaseClient._make_api_call\")]\n\n    capture_span_ctx = capture_span\n\n    def _call(self, service, instance, args, kwargs):\n        \"\"\"\n        This is split out from `call()` so that it can be re-used by the\n        aiobotocore instrumentation without duplicating all of this code.\n        \"\"\"\n        if \"operation_name\" in kwargs:\n            operation_name = kwargs[\"operation_name\"]\n        else:\n            operation_name = args[0]\n\n        parsed_url = urllib.parse.urlparse(instance.meta.endpoint_url)\n        context = {\n            \"destination\": {\n                \"address\": parsed_url.hostname,\n                \"port\": parsed_url.port,\n                \"cloud\": {\"region\": instance.meta.region_name},\n            }\n        }\n\n        handler_info = None\n        handler = handlers.get(service, False)\n        if handler:\n            handler_info = handler(operation_name, service, instance, args, kwargs, context)\n        if not handler_info:\n            handler_info = handle_default(operation_name, service, instance, args, kwargs, context)\n\n        return self.capture_span_ctx(\n            handler_info.signature,\n            span_type=handler_info.span_type,\n            leaf=True,\n            span_subtype=handler_info.span_subtype,\n            span_action=handler_info.span_action,\n            extra=handler_info.context,\n        )\n\n    def _get_service(self, instance):\n        service_model = instance.meta.service_model\n        if hasattr(service_model, \"service_id\"):  # added in boto3 1.7\n            service = service_model.service_id\n        else:\n            service = service_model.service_name.upper()\n            service = endpoint_to_service_id.get(service, service)\n        return service\n\n    def call(self, module, method, wrapped, instance, args, kwargs):\n        service = self._get_service(instance)\n\n        ctx = self._call(service, instance, args, kwargs)\n        with ctx as span:\n            if service in span_modifiers:\n                span_modifiers[service](span, args, kwargs)\n            return wrapped(*args, **kwargs)\n\n\ndef handle_s3(operation_name, service, instance, args, kwargs, context):\n    span_type = \"storage\"\n    span_subtype = \"s3\"\n    span_action = operation_name\n    if len(args) > 1 and \"Bucket\" in args[1]:\n        bucket = args[1][\"Bucket\"]\n    else:\n        # TODO handle Access Points\n        bucket = \"\"\n    signature = f\"S3 {operation_name} {bucket}\"\n\n    context[\"destination\"][\"service\"] = {\"name\": span_subtype, \"resource\": bucket, \"type\": span_type}\n\n    return HandlerInfo(signature, span_type, span_subtype, span_action, context)\n\n\ndef handle_dynamodb(operation_name, service, instance, args, kwargs, context):\n    span_type = \"db\"\n    span_subtype = \"dynamodb\"\n    span_action = \"query\"\n    if len(args) > 1 and \"TableName\" in args[1]:\n        table = args[1][\"TableName\"]\n    else:\n        table = \"\"\n    signature = f\"DynamoDB {operation_name} {table}\".rstrip()\n\n    context[\"db\"] = {\"type\": \"dynamodb\", \"instance\": instance.meta.region_name}\n    if operation_name == \"Query\" and len(args) > 1 and \"KeyConditionExpression\" in args[1]:\n        context[\"db\"][\"statement\"] = args[1][\"KeyConditionExpression\"]\n\n    context[\"destination\"][\"service\"] = {\"name\": span_subtype, \"resource\": table, \"type\": span_type}\n    return HandlerInfo(signature, span_type, span_subtype, span_action, context)\n\n\ndef handle_sns(operation_name, service, instance, args, kwargs, context):\n    if operation_name != \"Publish\":\n        # only \"publish\" is handled specifically, other endpoints get the default treatment\n        return False\n    span_type = \"messaging\"\n    span_subtype = \"sns\"\n    span_action = \"send\"\n    topic_name = \"\"\n    if len(args) > 1:\n        if \"Name\" in args[1]:\n            topic_name = args[1][\"Name\"]\n        if \"TopicArn\" in args[1]:\n            topic_name = args[1][\"TopicArn\"].rsplit(\":\", maxsplit=1)[-1]\n    signature = f\"SNS {operation_name} {topic_name}\".rstrip()\n    context[\"destination\"][\"service\"] = {\n        \"name\": span_subtype,\n        \"resource\": f\"{span_subtype}/{topic_name}\" if topic_name else span_subtype,\n        \"type\": span_type,\n    }\n    return HandlerInfo(signature, span_type, span_subtype, span_action, context)\n\n\nSQS_OPERATIONS = {\n    \"SendMessage\": {\"span_action\": \"send\", \"signature\": \"SEND to\"},\n    \"SendMessageBatch\": {\"span_action\": \"send_batch\", \"signature\": \"SEND_BATCH to\"},\n    \"ReceiveMessage\": {\"span_action\": \"receive\", \"signature\": \"RECEIVE from\"},\n    \"DeleteMessage\": {\"span_action\": \"delete\", \"signature\": \"DELETE from\"},\n    \"DeleteMessageBatch\": {\"span_action\": \"delete_batch\", \"signature\": \"DELETE_BATCH from\"},\n}\n\n\ndef handle_sqs(operation_name, service, instance, args, kwargs, context):\n    op = SQS_OPERATIONS.get(operation_name, None)\n    if not op:\n        # only \"publish\" is handled specifically, other endpoints get the default treatment\n        return False\n    span_type = \"messaging\"\n    span_subtype = \"sqs\"\n    topic_name = \"\"\n\n    if len(args) > 1:\n        topic_name = args[1][\"QueueUrl\"].rsplit(\"/\", maxsplit=1)[-1]\n    signature = f\"SQS {op['signature']} {topic_name}\".rstrip() if topic_name else f\"SQS {op['signature']}\"\n    context[\"destination\"][\"service\"] = {\n        \"name\": span_subtype,\n        \"resource\": f\"{span_subtype}/{topic_name}\" if topic_name else span_subtype,\n        \"type\": span_type,\n    }\n    return HandlerInfo(signature, span_type, span_subtype, op[\"span_action\"], context)\n\n\ndef modify_span_sqs(span, args, kwargs):\n    if span.id:\n        trace_parent = span.transaction.trace_parent.copy_from(span_id=span.id)\n    else:\n        # this is a dropped span, use transaction id instead\n        transaction = execution_context.get_transaction()\n        trace_parent = transaction.trace_parent.copy_from(span_id=transaction.id)\n    attributes = {constants.TRACEPARENT_HEADER_NAME: {\"DataType\": \"String\", \"StringValue\": trace_parent.to_string()}}\n    if trace_parent.tracestate:\n        attributes[constants.TRACESTATE_HEADER_NAME] = {\"DataType\": \"String\", \"StringValue\": trace_parent.tracestate}\n    if len(args) > 1:\n        attributes_count = len(attributes)\n        if \"MessageAttributes\" in args[1]:\n            messages = [args[1]]\n        # both send_batch and delete_batch use the same \"Entries\" list. We only want to add the\n        # traceparent to send_batch. We use the existence of ReceiptHandle to differentiate between the two\n        elif \"Entries\" in args[1] and args[1][\"Entries\"] and \"ReceiptHandle\" not in args[1][\"Entries\"][0]:\n            messages = args[1][\"Entries\"]\n        else:\n            messages = []\n        for message in messages:\n            if len(message[\"MessageAttributes\"]) + attributes_count <= SQS_MAX_ATTRIBUTES:\n                message[\"MessageAttributes\"].update(attributes)\n            else:\n                logger.info(\"Not adding disttracing headers to message due to attribute limit reached\")\n\n\ndef handle_default(operation_name, service, instance, args, kwargs, destination):\n    span_type = \"aws\"\n    span_subtype = service.lower()\n    span_action = operation_name\n\n    destination[\"service\"] = {\"name\": span_subtype, \"resource\": span_subtype, \"type\": span_type}\n\n    signature = f\"{service}:{operation_name}\"\n    return HandlerInfo(signature, span_type, span_subtype, span_action, destination)\n\n\nhandlers = {\n    \"S3\": handle_s3,\n    \"DynamoDB\": handle_dynamodb,\n    \"SNS\": handle_sns,\n    \"SQS\": handle_sqs,\n    \"default\": handle_default,\n}\n\nspan_modifiers = {\n    \"SQS\": modify_span_sqs,\n}\n", "path": "elasticapm/instrumentation/packages/botocore.py"}]}