We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[META 739] Add "s3/"-prefix to S3 destination.service.resource
See meta issue for the description and details:
- Meta issue: https://github.com/elastic/apm/issues/739

[META 739] Add "s3/"-prefix to S3 destination.service.resource
See meta issue for the description and details:
- Meta issue: https://github.com/elastic/apm/issues/739


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

{"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@@ -123,7 +123,7 @@\n         bucket = \"\"\n     signature = f\"S3 {operation_name} {bucket}\"\n \n-    context[\"destination\"][\"service\"] = {\"name\": span_subtype, \"resource\": bucket, \"type\": span_type}\n+    context[\"destination\"][\"service\"] = {\"name\": span_subtype, \"resource\": f\"s3/{bucket}\", \"type\": span_type}\n \n     return HandlerInfo(signature, span_type, span_subtype, span_action, context)\n", "issue": "[META 739] Add \"s3/\"-prefix to S3 destination.service.resource\nSee meta issue for the description and details:\r\n- Meta issue: https://github.com/elastic/apm/issues/739\r\n\n[META 739] Add \"s3/\"-prefix to S3 destination.service.resource\nSee meta issue for the description and details:\r\n- Meta issue: https://github.com/elastic/apm/issues/739\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\nimport urllib.parse\nfrom collections import namedtuple\n\nfrom elasticapm.conf import constants\nfrom elasticapm.instrumentation.packages.base import AbstractInstrumentedModule\nfrom elasticapm.traces import SpanType, capture_span, execution_context\nfrom elasticapm.utils.disttracing import TraceParent\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        operation_name = kwargs.get(\"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 pre_span_modifiers:\n                pre_span_modifiers[service](span, args, kwargs)\n            result = wrapped(*args, **kwargs)\n            if service in post_span_modifiers:\n                post_span_modifiers[service](span, args, kwargs, result)\n            request_id = result.get(\"ResponseMetadata\", {}).get(\"RequestId\")\n            if request_id:\n                span.update_context(\"http\", {\"request\": {\"id\": request_id}})\n            return result\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_pre(span, args, kwargs):\n    operation_name = kwargs.get(\"operation_name\", args[0])\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        if operation_name in (\"SendMessage\", \"SendMessageBatch\"):\n            attributes_count = len(attributes)\n            if operation_name == \"SendMessage\":\n                messages = [args[1]]\n            else:\n                messages = args[1][\"Entries\"]\n            for message in messages:\n                message[\"MessageAttributes\"] = message.get(\"MessageAttributes\") or {}\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        elif operation_name == \"ReceiveMessage\":\n            message_attributes = args[1].setdefault(\"MessageAttributeNames\", [])\n            if \"All\" not in message_attributes:\n                message_attributes.extend([constants.TRACEPARENT_HEADER_NAME, constants.TRACESTATE_HEADER_NAME])\n\n\ndef modify_span_sqs_post(span: SpanType, args, kwargs, result):\n    operation_name = kwargs.get(\"operation_name\", args[0])\n    if operation_name == \"ReceiveMessage\" and \"Messages\" in result:\n        for message in result[\"Messages\"][:1000]:  # only up to 1000 span links are recorded\n            if \"MessageAttributes\" in message and constants.TRACEPARENT_HEADER_NAME in message[\"MessageAttributes\"]:\n                tp = TraceParent.from_string(\n                    message[\"MessageAttributes\"][constants.TRACEPARENT_HEADER_NAME][\"StringValue\"]\n                )\n                span.add_link(tp)\n\n\ndef handle_default(operation_name, service, instance, args, kwargs, context):\n    span_type = \"aws\"\n    span_subtype = service.lower()\n    span_action = operation_name\n\n    context[\"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, context)\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\npre_span_modifiers = {\n    \"SQS\": modify_span_sqs_pre,\n}\n\npost_span_modifiers = {\n    \"SQS\": modify_span_sqs_post,\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 SpanType, capture_span, execution_context\nfrom elasticapm.utils.disttracing import TraceParent\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        operation_name = kwargs.get(\"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 pre_span_modifiers:\n                pre_span_modifiers[service](span, args, kwargs)\n            result = wrapped(*args, **kwargs)\n            if service in post_span_modifiers:\n                post_span_modifiers[service](span, args, kwargs, result)\n            request_id = result.get(\"ResponseMetadata\", {}).get(\"RequestId\")\n            if request_id:\n                span.update_context(\"http\", {\"request\": {\"id\": request_id}})\n            return result\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\": f\"s3/{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_pre(span, args, kwargs):\n    operation_name = kwargs.get(\"operation_name\", args[0])\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        if operation_name in (\"SendMessage\", \"SendMessageBatch\"):\n            attributes_count = len(attributes)\n            if operation_name == \"SendMessage\":\n                messages = [args[1]]\n            else:\n                messages = args[1][\"Entries\"]\n            for message in messages:\n                message[\"MessageAttributes\"] = message.get(\"MessageAttributes\") or {}\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        elif operation_name == \"ReceiveMessage\":\n            message_attributes = args[1].setdefault(\"MessageAttributeNames\", [])\n            if \"All\" not in message_attributes:\n                message_attributes.extend([constants.TRACEPARENT_HEADER_NAME, constants.TRACESTATE_HEADER_NAME])\n\n\ndef modify_span_sqs_post(span: SpanType, args, kwargs, result):\n    operation_name = kwargs.get(\"operation_name\", args[0])\n    if operation_name == \"ReceiveMessage\" and \"Messages\" in result:\n        for message in result[\"Messages\"][:1000]:  # only up to 1000 span links are recorded\n            if \"MessageAttributes\" in message and constants.TRACEPARENT_HEADER_NAME in message[\"MessageAttributes\"]:\n                tp = TraceParent.from_string(\n                    message[\"MessageAttributes\"][constants.TRACEPARENT_HEADER_NAME][\"StringValue\"]\n                )\n                span.add_link(tp)\n\n\ndef handle_default(operation_name, service, instance, args, kwargs, context):\n    span_type = \"aws\"\n    span_subtype = service.lower()\n    span_action = operation_name\n\n    context[\"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, context)\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\npre_span_modifiers = {\n    \"SQS\": modify_span_sqs_pre,\n}\n\npost_span_modifiers = {\n    \"SQS\": modify_span_sqs_post,\n}\n", "path": "elasticapm/instrumentation/packages/botocore.py"}]}