feat(tracer): [SVLS-5262] Botocore Success Span Pointers (#10690)

Creating Span Pointers for successful botocore patched api calls. Still
a private feature.

## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

ddtrace/_trace/trace_handlers.py

@@ -9,10 +9,12 @@
 
 import wrapt
 
+from ddtrace._trace._span_pointer import _SpanPointerDescription
 from ddtrace._trace.span import Span
 from ddtrace._trace.utils import extract_DD_context_from_messages
 from ddtrace._trace.utils import set_botocore_patched_api_call_span_tags as set_patched_api_call_span_tags
 from ddtrace._trace.utils import set_botocore_response_metadata_tags
+from ddtrace._trace.utils_botocore.span_pointers import extract_span_pointers_from_successful_botocore_response
 from ddtrace.constants import _ANALYTICS_SAMPLE_RATE_KEY
 from ddtrace.constants import SPAN_KIND
 from ddtrace.constants import SPAN_MEASURED_KEY
@@ -620,7 +622,17 @@ def _on_botocore_patched_api_call_exception(ctx, response, exception_type, is_er
 
 
 def _on_botocore_patched_api_call_success(ctx, response):
-    set_botocore_response_metadata_tags(ctx.get_item(ctx.get_item("call_key")), response)
+    span = ctx.get_item(ctx.get_item("call_key"))
+
+    set_botocore_response_metadata_tags(span, response)
+
+    for span_pointer_description in extract_span_pointers_from_successful_botocore_response(
+        endpoint_name=ctx.get_item("endpoint_name"),
+        operation_name=ctx.get_item("operation"),
+        request_parameters=ctx.get_item("params"),
+        response=response,
+    ):
+        _set_span_pointer(span, span_pointer_description)
 
 
 def _on_botocore_trace_context_injection_prepared(
@@ -774,6 +786,15 @@ def _on_test_visibility_is_enabled() -> bool:
     return CIVisibility.enabled
 
 
+def _set_span_pointer(span: Span, span_pointer_description: _SpanPointerDescription) -> None:
+    span._add_span_pointer(
+        pointer_kind=span_pointer_description.pointer_kind,
+        pointer_direction=span_pointer_description.pointer_direction,
+        pointer_hash=span_pointer_description.pointer_hash,
+        extra_attributes=span_pointer_description.extra_attributes,
+    )
+
+
 def listen():
     core.on("wsgi.block.started", _wsgi_make_block_content, "status_headers_content")
     core.on("asgi.block.started", _asgi_make_block_content, "status_headers_content")

releasenotes/notes/span-pointers-aws-s3-putobject-32b66688c9cad5e6.yaml

@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    botocore: Adds span pointers to some successful AWS botocore spans. Currently only supports S3 PutObject.

tests/contrib/botocore/test.py

@@ -21,6 +21,8 @@
 from moto import mock_stepfunctions
 import pytest
 
+from ddtrace._trace._span_pointer import _SpanPointer
+from ddtrace._trace._span_pointer import _SpanPointerDirection
 from tests.utils import get_128_bit_trace_id_from_headers
 
 
@@ -284,6 +286,8 @@ def test_s3_client(self):
         assert span.service == "test-botocore-tracing.s3"
         assert span.resource == "s3.listbuckets"
 
+        assert not span._links, "no links, i.e. no span pointers"
+
         # testing for span error
         self.reset()
         try:
@@ -392,6 +396,17 @@ def test_s3_put(self):
         assert span.get_tag("aws.s3.bucket_name") == "mybucket"
         assert span.get_tag("bucketname") == "mybucket"
 
+        assert span._links == [
+            _SpanPointer(
+                pointer_kind="aws.s3.object",
+                pointer_direction=_SpanPointerDirection.DOWNSTREAM,
+                # We have more detailed tests for the hashing behavior
+                # elsewhere. Here we just want to make sure that the pointer is
+                # correctly attached to the span.
+                pointer_hash="def44fdefcd83bc907515567dc742be1",
+            ),
+        ]
+
     @mock_s3
     def test_s3_put_no_params(self):
         with self.override_config("botocore", dict(tag_no_params=True)):
@@ -403,6 +418,18 @@ def test_s3_put_no_params(self):
             assert span.get_tag("params.Body") is None
             assert span.get_tag("component") == "botocore"
 
+            # We still create the link since we're hashing the parameter data.
+            assert span._links == [
+                _SpanPointer(
+                    pointer_kind="aws.s3.object",
+                    pointer_direction=_SpanPointerDirection.DOWNSTREAM,
+                    # We have more detailed tests for the hashing behavior
+                    # elsewhere. Here we just want to make sure that the pointer is
+                    # correctly attached to the span.
+                    pointer_hash="def44fdefcd83bc907515567dc742be1",
+                ),
+            ]
+
     @mock_s3
     @TracerTestCase.run_in_subprocess(env_overrides=dict(DD_BOTOCORE_SERVICE="botocore"))
     def test_service_name_override(self):