feat(aws): enhance service representation for serverless

ddtrace/_trace/utils_botocore/span_tags.py

@@ -12,12 +12,55 @@
 from ddtrace.ext import aws
 from ddtrace.ext import http
 from ddtrace.internal.constants import COMPONENT
+from ddtrace.internal.serverless import in_aws_lambda
 from ddtrace.internal.utils.formats import deep_getattr
 
 
 _PAYLOAD_TAGGER = AWSPayloadTagging()
 
 
+# Helper to build AWS hostname from service, region and parameters
+def _derive_peer_hostname(service: str, region: str, params: Optional[Dict[str, Any]] = None) -> Optional[str]:
+    """Return hostname for given AWS service according to Datadog peer hostname rules.
+
+    Only returns hostnames for specific AWS services:
+        - eventbridge/events -> events.<region>.amazonaws.com
+        - sqs               -> sqs.<region>.amazonaws.com
+        - sns               -> sns.<region>.amazonaws.com
+        - kinesis           -> kinesis.<region>.amazonaws.com
+        - dynamodb          -> dynamodb.<region>.amazonaws.com
+        - s3                -> <bucket>.s3.<region>.amazonaws.com (if Bucket param present)
+                              s3.<region>.amazonaws.com          (otherwise)
+
+    Other services return ``None``.
+    """
+
+    if not region:
+        return None
+
+    aws_service = service.lower()
+
+    # Only set peer.service for specific services
+    if aws_service in {"eventbridge", "events"}:
+        return f"events.{region}.amazonaws.com"
+    if aws_service == "sqs":
+        return f"sqs.{region}.amazonaws.com"
+    if aws_service == "sns":
+        return f"sns.{region}.amazonaws.com"
+    if aws_service == "kinesis":
+        return f"kinesis.{region}.amazonaws.com"
+    if aws_service in {"dynamodb", "dynamodbdocument"}:
+        return f"dynamodb.{region}.amazonaws.com"
+    if aws_service == "s3":
+        bucket = params.get("Bucket") if params else None
+        if bucket:
+            return f"{bucket}.s3.{region}.amazonaws.com"
+        return f"s3.{region}.amazonaws.com"
+
+    # Return None for all other services
+    return None
+
+
 def set_botocore_patched_api_call_span_tags(span: Span, instance, args, params, endpoint_name, operation):
     span.set_tag_str(COMPONENT, config.botocore.integration_name)
     # set span.kind to the type of request being performed
@@ -51,6 +94,13 @@ def set_botocore_patched_api_call_span_tags(span: Span, instance, args, params,
         span.set_tag_str("aws.region", region_name)
         span.set_tag_str("region", region_name)
 
+        # Derive peer hostname only in serverless environments to avoid
+        # unnecessary tag noise in traditional hosts/containers.
+        if in_aws_lambda():
+            hostname = _derive_peer_hostname(endpoint_name, region_name, params)
+            if hostname:
+                span.set_tag_str("peer.service", hostname)
+
 
 def set_botocore_response_metadata_tags(
     span: Span, result: Dict[str, Any], is_error_code_fn: Optional[Callable] = None

ddtrace/contrib/internal/aiobotocore/patch.py

@@ -5,6 +5,7 @@
 import wrapt
 
 from ddtrace import config
+from ddtrace._trace.utils_botocore.span_tags import _derive_peer_hostname
 from ddtrace.constants import _SPAN_MEASURED_KEY
 from ddtrace.constants import SPAN_KIND
 from ddtrace.contrib.internal.trace_utils import ext_service
@@ -16,6 +17,7 @@
 from ddtrace.internal.constants import COMPONENT
 from ddtrace.internal.schema import schematize_cloud_api_operation
 from ddtrace.internal.schema import schematize_service_name
+from ddtrace.internal.serverless import in_aws_lambda
 from ddtrace.internal.utils import ArgumentError
 from ddtrace.internal.utils import get_argument_value
 from ddtrace.internal.utils.formats import asbool
@@ -145,6 +147,12 @@ async def _wrapped_api_call(original_func, instance, args, kwargs):
 
         region_name = deep_getattr(instance, "meta.region_name")
 
+        if in_aws_lambda():
+            # Derive the peer hostname now that we have both service and region.
+            hostname = _derive_peer_hostname(endpoint_name, region_name, params)
+            if hostname:
+                span.set_tag_str("peer.service", hostname)
+
         meta = {
             "aws.agent": "aiobotocore",
             "aws.operation": operation,

ddtrace/contrib/internal/boto/patch.py

@@ -7,6 +7,7 @@
 import wrapt
 
 from ddtrace import config
+from ddtrace._trace.utils_botocore.span_tags import _derive_peer_hostname
 from ddtrace.constants import _SPAN_MEASURED_KEY
 from ddtrace.constants import SPAN_KIND
 from ddtrace.ext import SpanKind
@@ -16,6 +17,7 @@
 from ddtrace.internal.constants import COMPONENT
 from ddtrace.internal.schema import schematize_cloud_api_operation
 from ddtrace.internal.schema import schematize_service_name
+from ddtrace.internal.serverless import in_aws_lambda
 from ddtrace.internal.utils import get_argument_value
 from ddtrace.internal.utils.formats import asbool
 from ddtrace.internal.utils.wrappers import unwrap
@@ -122,6 +124,12 @@ def patched_query_request(original_func, instance, args, kwargs):
             meta[aws.REGION] = region_name
             meta[aws.AWSREGION] = region_name
 
+            if in_aws_lambda():
+                # Derive the peer hostname now that we have both service and region.
+                hostname = _derive_peer_hostname(endpoint_name, region_name, params)
+                if hostname:
+                    meta["peer.service"] = hostname
+
         span.set_tags(meta)
 
         # Original func returns a boto.connection.HTTPResponse object
@@ -183,6 +191,12 @@ def patched_auth_request(original_func, instance, args, kwargs):
             meta[aws.REGION] = region_name
             meta[aws.AWSREGION] = region_name
 
+            if in_aws_lambda():
+                # Derive the peer hostname
+                hostname = _derive_peer_hostname(endpoint_name, region_name, None)
+                if hostname:
+                    meta["peer.service"] = hostname
+
         span.set_tags(meta)
 
         # Original func returns a boto.connection.HTTPResponse object

ddtrace/internal/schema/processor.py

@@ -1,17 +1,33 @@
 from ddtrace._trace.processor import TraceProcessor
 from ddtrace.constants import _BASE_SERVICE_KEY
+from ddtrace.internal.serverless import in_aws_lambda
+from ddtrace.internal.serverless import in_azure_function
+from ddtrace.internal.serverless import in_gcp_function
 from ddtrace.settings._config import config
 
 from . import schematize_service_name
 
 
 class BaseServiceProcessor(TraceProcessor):
     def __init__(self):
+        # Determine the global (root) service for this process according to the
+        # active schema.  In serverless environments the inferred base service
+        # often resolves to the string ``"runtime"`` which is not useful to
+        # users and pollutes span metadata.  Detect that situation once and, if
+        # applicable, disable tagging entirely.
+
         self._global_service = schematize_service_name((config.service or "").lower())
 
+        # Skip tagging when running in a serverless runtime *and* the inferred
+        # service name is the generic "runtime" placeholder.
+        self._skip_tagging = self._global_service == "runtime" and (
+            in_aws_lambda() or in_gcp_function() or in_azure_function()
+        )
+
     def process_trace(self, trace):
-        if not trace:
-            return
+        if not trace or self._skip_tagging:
+            # Nothing to do (either no spans, or tagging disabled for this env)
+            return trace
 
         traces_to_process = filter(
             lambda x: x.service and x.service.lower() != self._global_service,

releasenotes/notes/fix-serverless-service-representation-b1f81fe0553a7a45.yaml

@@ -0,0 +1,65 @@
+---
+#instructions:
+#    The style guide below provides explanations, instructions, and templates to write your own release note.
+#    Once finished, all irrelevant sections (including this instruction section) should be removed,
+#    and the release note should be committed with the rest of the changes.
+#
+#    The main goal of a release note is to provide a brief overview of a change and provide actionable steps to the user.
+#    The release note should clearly communicate what the change is, why the change was made, and how a user can migrate their code.
+#
+#    The release note should also clearly distinguish between announcements and user instructions. Use:
+#    * Past tense for previous/existing behavior (ex: ``resulted, caused, failed``)
+#    * Third person present tense for the change itself (ex: ``adds, fixes, upgrades``)
+#    * Active present infinitive for user instructions (ex: ``set, use, add``)
+#
+#    Release notes should:
+#    * Use plain language
+#    * Be concise
+#    * Include actionable steps with the necessary code changes
+#    * Include relevant links (bug issues, upstream issues or release notes, documentation pages)
+#    * Use full sentences with sentence-casing and punctuation.
+#    * Before using Datadog specific acronyms/terminology, a release note must first introduce them with a definition.
+#
+#    Release notes should not:
+#    * Be vague. Example: ``fixes an issue in tracing``.
+#    * Use overly technical language
+#    * Use dynamic links (``stable/latest/1.x`` URLs). Instead, use static links (specific version, commit hash) whenever possible so that they don't break in the future.
+prelude: >
+    Usually in tandem with a new feature or major change, meant to provide context or background for a major change. 
+    No specific format other than a required scope is provided and the author is requested to use their best judgment.
+    Format: <scope>: <add_prelude_and_context_here>.
+features:
+  - |
+    For new features such as a new integration or component. Use present tense with the following format:
+    Format: <scope>: This introduces <new_feature_or_component>.
+issues:
+  - |
+    For known issues. Use present tense with the following format:
+    Format: <scope>: There is a known <symptom_of_issue> issue with <affected_code>.
+    <provide_actionable_workaround_here>.
+upgrade:
+  - |
+    For enhanced functionality or if package dependencies are upgraded. If applicable, include instructions
+    for how a user can migrate their code.
+    Use present tense with the following formats, respectively for enhancements or removals:
+    Format: <scope>: This upgrades <present_tense_explanation>. With this upgrade, you can <actionable_step_for_user>.
+  - |
+    Format: <scope>: <affected_code> is now removed. As an alternative to <affected_code>, you can use <alternative> instead.
+deprecations:
+  - |
+    Warning of a component or member of the public API being removed in the future.
+    Use present tense for when deprecation actually happens and future tense for when removal is planned to happen. 
+    Include deprecation/removal timeline, as well as workarounds and alternatives in the following format:
+    Format: <scope>: <affected_code> is deprecated and will be removed in <version_to_be_removed>.
+            As an alternative to <affected_code>, you can use <alternative> instead.
+fixes:
+  - |
+    For reporting bug fixes.
+    Use past tense for the problem and present tense for the fix and solution in the following format:
+    Format: <scope>: This fix resolves an issue where <ABC_bug> caused <XYZ_situation>.
+other:
+  - |
+    For any change which does not fall into any of the above categories. Since changes falling into this category are 
+    likely rare and not very similar to each other, no specific format other than a required scope is provided. 
+    The author is requested to use their best judgment to ensure a quality release note.
+    Format: <scope>: <add_release_note_here>.

tests/contrib/aiobotocore/test.py

@@ -553,3 +553,102 @@ async def test_response_context_manager(tracer):
         assert span.name == "s3.command"
         assert span.get_tag("component") == "aiobotocore"
         assert span.get_tag("span.kind") == "client"
+
+
+# Peer service tests
+@pytest.mark.asyncio
+async def test_sqs_client_peer_service_in_lambda(tracer, monkeypatch):
+    """Test that peer.service tag is set for SQS when running in AWS Lambda"""
+    monkeypatch.setenv("AWS_LAMBDA_FUNCTION_NAME", "my-func")
+
+    async with aiobotocore_client("sqs", tracer) as sqs:
+        await sqs.list_queues()
+
+    traces = tracer.pop_traces()
+    assert len(traces) == 1
+    assert len(traces[0]) == 1
+    span = traces[0][0]
+    # Should have peer.service set to sqs hostname
+    assert span.get_tag("peer.service") == "sqs.us-west-2.amazonaws.com"
+
+
+@pytest.mark.asyncio
+async def test_s3_client_peer_service_in_lambda(tracer, monkeypatch):
+    """Test that peer.service tag is set for S3 when running in AWS Lambda"""
+    monkeypatch.setenv("AWS_LAMBDA_FUNCTION_NAME", "my-func")
+    bucket_name = f"{time.time()}bucket".replace(".", "")
+
+    async with aiobotocore_client("s3", tracer) as s3:
+        # Test with bucket parameter
+        await s3.create_bucket(Bucket=bucket_name, CreateBucketConfiguration={"LocationConstraint": "us-west-2"})
+
+    traces = tracer.pop_traces()
+    assert len(traces) == 1
+    assert len(traces[0]) == 1
+    span = traces[0][0]
+    # Should have peer.service set to bucket-specific hostname
+    assert span.get_tag("peer.service") == f"{bucket_name}.s3.us-west-2.amazonaws.com"
+
+
+@pytest.mark.asyncio
+async def test_dynamodb_client_peer_service_in_lambda(tracer, monkeypatch):
+    """Test that peer.service tag is set for DynamoDB when running in AWS Lambda"""
+    monkeypatch.setenv("AWS_LAMBDA_FUNCTION_NAME", "my-func")
+
+    async with aiobotocore_client("dynamodb", tracer) as dynamodb:
+        await dynamodb.list_tables()
+
+    traces = tracer.pop_traces()
+    assert len(traces) == 1
+    assert len(traces[0]) == 1
+    span = traces[0][0]
+    # Should have peer.service set to dynamodb hostname
+    assert span.get_tag("peer.service") == "dynamodb.us-west-2.amazonaws.com"
+
+
+@pytest.mark.asyncio
+async def test_kinesis_client_peer_service_in_lambda(tracer, monkeypatch):
+    """Test that peer.service tag is set for Kinesis when running in AWS Lambda"""
+    monkeypatch.setenv("AWS_LAMBDA_FUNCTION_NAME", "my-func")
+
+    async with aiobotocore_client("kinesis", tracer) as kinesis:
+        await kinesis.list_streams()
+
+    traces = tracer.pop_traces()
+    assert len(traces) == 1
+    assert len(traces[0]) == 1
+    span = traces[0][0]
+    # Should have peer.service set to kinesis hostname
+    assert span.get_tag("peer.service") == "kinesis.us-west-2.amazonaws.com"
+
+
+@pytest.mark.asyncio
+async def test_sns_client_peer_service_in_lambda(tracer, monkeypatch):
+    """Test that peer.service tag is set for SNS when running in AWS Lambda"""
+    monkeypatch.setenv("AWS_LAMBDA_FUNCTION_NAME", "my-func")
+
+    async with aiobotocore_client("sns", tracer) as sns:
+        await sns.list_topics()
+
+    traces = tracer.pop_traces()
+    assert len(traces) == 1
+    assert len(traces[0]) == 1
+    span = traces[0][0]
+    # Should have peer.service set to sns hostname
+    assert span.get_tag("peer.service") == "sns.us-west-2.amazonaws.com"
+
+
+@pytest.mark.asyncio
+async def test_eventbridge_client_peer_service_in_lambda(tracer, monkeypatch):
+    """Test that peer.service tag is set for EventBridge when running in AWS Lambda"""
+    monkeypatch.setenv("AWS_LAMBDA_FUNCTION_NAME", "my-func")
+
+    async with aiobotocore_client("events", tracer) as events:
+        await events.list_rules()
+
+    traces = tracer.pop_traces()
+    assert len(traces) == 1
+    assert len(traces[0]) == 1
+    span = traces[0][0]
+    # Should have peer.service set to events hostname
+    assert span.get_tag("peer.service") == "events.us-west-2.amazonaws.com"