feat: mark manually created spans with service source [APMLP-1016]

.github/workflows/system-tests.yml

@@ -95,7 +95,7 @@ jobs:
           persist-credentials: false
           repository: 'DataDog/system-tests'
           # Automatically managed, use scripts/update-system-tests-version to update
-          ref: 'baa899e768859fce1be7e98cc2518d810745d65a'
+          ref: '5ede8c7064df9245bf0323d7e027e92fc8740a9e'
 
       - name: Download wheels to binaries directory
         uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
@@ -148,7 +148,7 @@ jobs:
           persist-credentials: false
           repository: 'DataDog/system-tests'
           # Automatically managed, use scripts/update-system-tests-version to update
-          ref: 'baa899e768859fce1be7e98cc2518d810745d65a'
+          ref: '5ede8c7064df9245bf0323d7e027e92fc8740a9e'
 
       - name: Build runner
         uses: ./.github/actions/install_runner
@@ -369,7 +369,7 @@ jobs:
           persist-credentials: false
           repository: 'DataDog/system-tests'
           # Automatically managed, use scripts/update-system-tests-version to update
-          ref: 'baa899e768859fce1be7e98cc2518d810745d65a'
+          ref: '5ede8c7064df9245bf0323d7e027e92fc8740a9e'
       - name: Download wheels to binaries directory
         uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
         with:
@@ -513,4 +513,4 @@ jobs:
           needs.integration-frameworks-system-tests.result == 'cancelled'||
           needs.tracer-release.result == 'failure' ||
           needs.tracer-release.result == 'cancelled'
-        run: exit 1
+        run: exit 1

.gitlab-ci.yml

@@ -15,7 +15,7 @@ variables:
   DD_VPA_TEMPLATE: "vpa-template-cpu-p70-10percent-2x-oom-min-cap"
   # CI_DEBUG_SERVICES: "true"
   # Automatically managed, use scripts/update-system-tests-version to update
-  SYSTEM_TESTS_REF: "baa899e768859fce1be7e98cc2518d810745d65a"
+  SYSTEM_TESTS_REF: "5ede8c7064df9245bf0323d7e027e92fc8740a9e"
 
   # Profiling native build image (built from dd/images/dd-trace-py/profiling_native)
   PROFILING_NATIVE_IMAGE: "registry.ddbuild.io/dd-trace-py:v103334885-be1888c-profiling_native"

.gitlab/benchmarks/bp-runner.microbenchmarks.fail-on-breach.template.yml

@@ -976,7 +976,7 @@ experiments:
               - max_rss_usage < 53.00 MB
           - name: span-start-finish
             thresholds:
-              - execution_time < 58.00 ms
+              - execution_time < 58.50 ms
               - max_rss_usage < 38.00 MB
           - name: span-start-finish-telemetry
             thresholds:
@@ -1064,11 +1064,11 @@ experiments:
               - max_rss_usage < 39.25 MB
           - name: tracer-medium
             thresholds:
-              - execution_time < 3.30 ms
+              - execution_time < 3.50 ms
               - max_rss_usage < 38.75 MB
           - name: tracer-small
             thresholds:
-              - execution_time < 0.38 ms
+              - execution_time < 0.39 ms
               - max_rss_usage < 38.75 MB
 
           # codeprovenancefork

.gitlab/services.yml

@@ -25,7 +25,7 @@
       DD_POOL_TRACE_CHECK_FAILURES: true
       DD_DISABLE_ERROR_RESPONSES: true
       ENABLED_CHECKS: trace_content_length,trace_stall,meta_tracer_version_header,trace_count_header,trace_peer_service,trace_dd_service
-      SNAPSHOT_IGNORED_ATTRS: span_id,trace_id,parent_id,duration,start,metrics.system.pid,metrics.system.process_id,metrics.process_id,meta.runtime-id,meta._dd.p.tid,meta.pathway.hash,metrics._dd.tracer_kr,meta._dd.parent_id,meta.kafka.cluster_id,meta._dd.git.repository_url
+      SNAPSHOT_IGNORED_ATTRS: span_id,trace_id,parent_id,duration,start,metrics.system.pid,metrics.system.process_id,metrics.process_id,meta.runtime-id,meta._dd.p.tid,meta.pathway.hash,metrics._dd.tracer_kr,meta._dd.parent_id,meta.kafka.cluster_id,meta._dd.git.repository_url,meta._dd.svc_src
       VCR_PROVIDER_MAP: azure_openai=https://llmobs-test-resource.openai.azure.com/
   mongo:
     name: registry.ddbuild.io/images/mirror/mongo:6.0.5

ddtrace/_trace/subscribers/_base.py

@@ -10,7 +10,7 @@
 from ddtrace.constants import _SPAN_MEASURED_KEY
 from ddtrace.constants import SPAN_KIND
 from ddtrace.contrib import trace_utils
-from ddtrace.contrib.internal.trace_utils import maybe_set_service_source_tag
+from ddtrace.contrib.internal.trace_utils import set_service_and_source
 from ddtrace.internal import core
 from ddtrace.internal.constants import COMPONENT
 from ddtrace.internal.core.subscriber import ContextSubscriber
@@ -36,7 +36,7 @@ def _finish_span(
         return
 
     integration_config = ctx.get_item("integration_config")
-    maybe_set_service_source_tag(span, integration_config or dict())
+    set_service_and_source(span, ctx.get_item("service"), integration_config or dict())
 
     exc_type, exc_value, exc_traceback = exc_info
     if exc_type and exc_value and exc_traceback:
@@ -96,7 +96,7 @@ def _start_span(ctx: core.ExecutionContext[TracingEventType]) -> Span:
     if event.measured:
         span._set_attribute(_SPAN_MEASURED_KEY, 1)
 
-    maybe_set_service_source_tag(span, integration_config or dict())
+    set_service_and_source(span, ctx.get_item("service"), integration_config or dict())
     ctx.span = span
 
     if config._inferred_proxy_services_enabled:

ddtrace/_trace/trace_handlers.py

@@ -36,7 +36,7 @@
 from ddtrace.contrib.internal.mlflow.constants import MLflowLogType
 from ddtrace.contrib.internal.trace_utils import _copy_trace_level_tags
 from ddtrace.contrib.internal.trace_utils import _set_url_tag
-from ddtrace.contrib.internal.trace_utils import maybe_set_service_source_tag
+from ddtrace.contrib.internal.trace_utils import set_service_and_source
 from ddtrace.ext import SpanKind
 from ddtrace.ext import SpanLinkKind
 from ddtrace.ext import SpanTypes
@@ -175,7 +175,7 @@ def _start_span(ctx: core.ExecutionContext, call_trace: bool = True, **kwargs) -
     if ctx.get_item("measured"):
         span._set_attribute(_SPAN_MEASURED_KEY, 1)
 
-    maybe_set_service_source_tag(span, integration_config or dict())
+    set_service_and_source(span, ctx.get_item("service"), integration_config or dict())
     ctx.span = span
 
     if config._inferred_proxy_services_enabled:
@@ -198,7 +198,7 @@ def _finish_span(
         return
 
     integration_config = ctx.get_item("integration_config")
-    maybe_set_service_source_tag(span, integration_config or dict())
+    set_service_and_source(span, ctx.get_item("service"), integration_config or dict())
 
     exc_type, exc_value, exc_traceback = exc_info
     if exc_type and exc_value and exc_traceback:

ddtrace/_trace/tracer.py

@@ -32,6 +32,7 @@
 from ddtrace.internal import debug
 from ddtrace.internal import forksafe
 from ddtrace.internal import hostname
+from ddtrace.internal.constants import _SERVICE_SOURCE
 from ddtrace.internal.constants import LOG_ATTR_ENV
 from ddtrace.internal.constants import LOG_ATTR_SERVICE
 from ddtrace.internal.constants import LOG_ATTR_SPAN_ID
@@ -470,11 +471,15 @@ def start_span(
         # 2. Parent's service name (if defined)
         # 3. Globally configured service name
         #     a. `config.service`/`DD_SERVICE`/`DD_TAGS`
+        service_source: str = ""
         if service is None:
             if parent:
                 service = parent.service
+                service_source = parent.get_tag(_SERVICE_SOURCE) or ""
             else:
-                service = config.service
+                service = service_source = config.service
+        else:
+            service_source = "m"
 
         # Update the service name based on any mapping
         if service is not None:
@@ -530,6 +535,9 @@ def start_span(
         if self._tags:
             span.set_tags(self._tags)
 
+        if service and service_source:
+            span._set_attribute(_SERVICE_SOURCE, service_source)
+
         if config.env:
             span._set_attribute(ENV_KEY, config.env)
 

ddtrace/contrib/internal/aiobotocore/patch.py

@@ -7,6 +7,7 @@
 from ddtrace.constants import _SPAN_MEASURED_KEY
 from ddtrace.constants import SPAN_KIND
 from ddtrace.contrib.internal.trace_utils import ext_service
+from ddtrace.contrib.internal.trace_utils import set_service_and_source
 from ddtrace.contrib.internal.trace_utils import unwrap
 from ddtrace.ext import SpanKind
 from ddtrace.ext import SpanTypes
@@ -120,9 +121,13 @@ async def _wrapped_api_call(original_func, instance, args, kwargs):
         schematize_cloud_api_operation(
             "{}.command".format(endpoint_name), cloud_provider="aws", cloud_service=endpoint_name
         ),
-        service=ext_service(pin, config.aiobotocore, default=schematize_service_name(fallback_service)),
         span_type=SpanTypes.HTTP,
     ) as span:
+        set_service_and_source(
+            span,
+            ext_service(pin, config.aiobotocore, default=schematize_service_name(fallback_service)),
+            config.aiobotocore,
+        )
         span._set_attribute(COMPONENT, config.aiobotocore.integration_name)
 
         # set span.kind tag equal to type of request

ddtrace/contrib/internal/aiohttp/patch.py

@@ -7,8 +7,8 @@
 from ddtrace.constants import SPAN_KIND
 from ddtrace.contrib.internal.trace_utils import ext_service
 from ddtrace.contrib.internal.trace_utils import extract_netloc_and_query_info_from_url
-from ddtrace.contrib.internal.trace_utils import maybe_set_service_source_tag
 from ddtrace.contrib.internal.trace_utils import set_http_meta
+from ddtrace.contrib.internal.trace_utils import set_service_and_source
 from ddtrace.contrib.internal.trace_utils import unwrap
 from ddtrace.contrib.internal.trace_utils import wrap
 from ddtrace.ext import SpanKind
@@ -84,16 +84,13 @@ async def _traced_clientsession_request(func, instance, args, kwargs):
     url: URL = base_url.join(raw_url) if base_url is not None else raw_url
     params = kwargs.get("params")
     headers = kwargs.get("headers") or {}
+    service: str = url.host if config.aiohttp_client.split_by_domain else ext_service(None, config.aiohttp_client)
 
     with tracer.trace(
         schematize_url_operation("aiohttp.request", protocol="http", direction=SpanDirection.OUTBOUND),
         span_type=SpanTypes.HTTP,
-        service=ext_service(None, config.aiohttp_client),
     ) as span:
-        if config.aiohttp_client.split_by_domain:
-            span.service = url.host
-
-        maybe_set_service_source_tag(span, config.aiohttp)
+        set_service_and_source(span, service, config.aiohttp)
 
         if config.aiohttp.distributed_tracing:
             HTTPPropagator.inject(span.context, headers)

ddtrace/contrib/internal/aiomysql/patch.py

@@ -8,6 +8,7 @@
 from ddtrace.contrib import dbapi
 from ddtrace.contrib import trace_utils
 from ddtrace.contrib.internal.trace_utils import _convert_to_string
+from ddtrace.contrib.internal.trace_utils import set_service_and_source
 from ddtrace.ext import SpanKind
 from ddtrace.ext import SpanTypes
 from ddtrace.ext import db
@@ -76,10 +77,10 @@ async def _trace_method(self, method, resource, extra_tags, *args, **kwargs):
 
         with tracer.trace(
             self._self_datadog_name,
-            service=trace_utils.ext_service(pin, config.aiomysql),
             resource=resource,
             span_type=SpanTypes.SQL,
         ) as s:
+            set_service_and_source(s, trace_utils.ext_service(pin, config.aiomysql), config.aiomysql)
             s._set_attribute(COMPONENT, config.aiomysql.integration_name)
 
             # set span.kind to the type of request being performed

ddtrace/contrib/internal/aiopg/connection.py

@@ -8,6 +8,7 @@
 from ddtrace.constants import SPAN_KIND
 from ddtrace.contrib import dbapi
 from ddtrace.contrib import trace_utils
+from ddtrace.contrib.internal.trace_utils import set_service_and_source
 from ddtrace.ext import SpanKind
 from ddtrace.ext import SpanTypes
 from ddtrace.ext import db
@@ -37,10 +38,10 @@ async def _trace_method(self, method, resource, extra_tags, *args, **kwargs):
 
         with tracer.trace(
             self._datadog_name,
-            service=trace_utils.ext_service(pin, config.aiopg),
             resource=resource,
             span_type=SpanTypes.SQL,
         ) as s:
+            set_service_and_source(s, trace_utils.ext_service(pin, config.aiopg), config.aiopg)
             s._set_attribute(COMPONENT, config.aiopg.integration_name)
             s._set_attribute(db.SYSTEM, "postgresql")
 

ddtrace/contrib/internal/algoliasearch/patch.py

@@ -5,6 +5,7 @@
 from ddtrace.constants import _SPAN_MEASURED_KEY
 from ddtrace.constants import SPAN_KIND
 from ddtrace.contrib import trace_utils
+from ddtrace.contrib.internal.trace_utils import set_service_and_source
 from ddtrace.ext import SpanKind
 from ddtrace.ext import SpanTypes
 from ddtrace.internal.constants import COMPONENT
@@ -124,9 +125,9 @@ def _patched_search(func, instance, wrapt_args, wrapt_kwargs):
 
     with tracer.trace(
         schematize_cloud_api_operation("algoliasearch.search", cloud_provider="algoliasearch", cloud_service="search"),
-        service=trace_utils.ext_service(pin, config.algoliasearch),
         span_type=SpanTypes.HTTP,
     ) as span:
+        set_service_and_source(span, trace_utils.ext_service(pin, config.algoliasearch), config.algoliasearch)
         span._set_attribute(COMPONENT, config.algoliasearch.integration_name)
 
         # set span.kind to the type of request being performed

ddtrace/contrib/internal/asyncpg/patch.py

@@ -10,7 +10,7 @@
 from ddtrace.constants import _SPAN_MEASURED_KEY
 from ddtrace.constants import SPAN_KIND
 from ddtrace.contrib.internal.trace_utils import ext_service
-from ddtrace.contrib.internal.trace_utils import maybe_set_service_source_tag
+from ddtrace.contrib.internal.trace_utils import set_service_and_source
 from ddtrace.contrib.internal.trace_utils import unwrap
 from ddtrace.contrib.internal.trace_utils import wrap
 from ddtrace.contrib.internal.trace_utils_async import with_traced_module
@@ -95,8 +95,8 @@ async def _traced_connect(asyncpg, pin, func, instance, args, kwargs):
     # When using a pool, there's a connection_class args
     is_pool_context = "connection_class" in kwargs
 
-    with tracer.trace("postgres.connect", span_type=SpanTypes.SQL, service=ext_service(pin, config.asyncpg)) as span:
-        maybe_set_service_source_tag(span, config.asyncpg)
+    with tracer.trace("postgres.connect", span_type=SpanTypes.SQL) as span:
+        set_service_and_source(span, ext_service(pin, config.asyncpg), config.asyncpg)
         span._set_attribute(COMPONENT, config.asyncpg.integration_name)
         span._set_attribute(db.SYSTEM, DBMS_NAME)
 
@@ -126,9 +126,9 @@ async def _traced_query(pin, method, query, args, kwargs):
     with tracer.trace(
         schematize_database_operation("postgres.query", database_provider="postgresql"),
         resource=query,
-        service=ext_service(pin, config.asyncpg),
         span_type=SpanTypes.SQL,
     ) as span:
+        set_service_and_source(span, ext_service(pin, config.asyncpg), config.asyncpg)
         span._set_attribute(COMPONENT, config.asyncpg.integration_name)
         span._set_attribute(db.SYSTEM, DBMS_NAME)
 

ddtrace/contrib/internal/boto/patch.py

@@ -9,6 +9,7 @@
 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 set_service_and_source
 from ddtrace.ext import SpanKind
 from ddtrace.ext import SpanTypes
 from ddtrace.ext import aws
@@ -89,9 +90,9 @@ def patched_query_request(original_func, instance, args, kwargs):
         schematize_cloud_api_operation(
             "{}.command".format(endpoint_name), cloud_provider="aws", cloud_service=endpoint_name
         ),
-        service=schematize_service_name("{}.{}".format(pin.service, endpoint_name)),
         span_type=SpanTypes.HTTP,
     ) as span:
+        set_service_and_source(span, schematize_service_name("{}.{}".format(pin.service, endpoint_name)), config.boto)
         span._set_attribute(COMPONENT, config.boto.integration_name)
 
         # set span.kind to the type of request being performed
@@ -169,9 +170,9 @@ def patched_auth_request(original_func, instance, args, kwargs):
         schematize_cloud_api_operation(
             "{}.command".format(endpoint_name), cloud_provider="aws", cloud_service=endpoint_name
         ),
-        service=schematize_service_name("{}.{}".format(pin.service, endpoint_name)),
         span_type=SpanTypes.HTTP,
     ) as span:
+        set_service_and_source(span, schematize_service_name("{}.{}".format(pin.service, endpoint_name)), config.boto)
         span._set_attribute(_SPAN_MEASURED_KEY, 1)
         if args:
             http_method = get_argument_value(args, kwargs, 0, "method")

ddtrace/contrib/internal/celery/app.py

@@ -15,6 +15,7 @@
 from ddtrace.contrib.internal.celery.signals import trace_postrun
 from ddtrace.contrib.internal.celery.signals import trace_prerun
 from ddtrace.contrib.internal.celery.signals import trace_retry
+from ddtrace.contrib.internal.trace_utils import set_service_and_source
 from ddtrace.ext import SpanKind
 from ddtrace.ext import SpanTypes
 from ddtrace.internal import core
@@ -103,8 +104,8 @@ def _traced_beat_inner(func, instance, args, kwargs):
         with tracer.trace(
             "celery.beat.{}".format(fn_name),
             span_type=SpanTypes.WORKER,
-            service=trace_utils.ext_service(pin, integration_config),
         ) as span:
+            set_service_and_source(span, trace_utils.ext_service(pin, integration_config), integration_config)
             if resource_fn:
                 span.resource = resource_fn(args)
             span._set_attribute(SPAN_KIND, SpanKind.PRODUCER)

ddtrace/contrib/internal/celery/signals.py

@@ -17,6 +17,7 @@
 from ddtrace.contrib.internal.celery.utils import retrieve_span_context
 from ddtrace.contrib.internal.celery.utils import retrieve_task_id
 from ddtrace.contrib.internal.celery.utils import set_tags_from_context
+from ddtrace.contrib.internal.trace_utils import set_service_and_source
 from ddtrace.ext import SpanKind
 from ddtrace.ext import SpanTypes
 from ddtrace.ext import net
@@ -53,10 +54,8 @@ def trace_prerun(*args, **kwargs):
 
     # propagate the `Span` in the current task Context
     service = config.celery["worker_service_name"]
-    span = tracer.trace(c.WORKER_ROOT_SPAN, service=service, resource=task.name, span_type=SpanTypes.WORKER)
-    # Use inline check for worker default service
-    if span.service == config.celery["_default_service_worker"]:
-        span.set_tag("_dd.svc_src", "celery")
+    span = tracer.trace(c.WORKER_ROOT_SPAN, resource=task.name, span_type=SpanTypes.WORKER)
+    set_service_and_source(span, service, config.celery, default_service_key="_default_service_worker")
 
     # set span.kind to the type of request being performed
     span._set_attribute(SPAN_KIND, SpanKind.CONSUMER)
@@ -128,10 +127,8 @@ def trace_before_publish(*args, **kwargs):
     # apply some tags here because most of the data is not available
     # in the task_after_publish signal
     service = config.celery["producer_service_name"]
-    span = tracer.trace(c.PRODUCER_ROOT_SPAN, service=service, resource=task_name)
-    # Use inline check for producer default service
-    if span.service == config.celery["_default_service_producer"]:
-        span.set_tag("_dd.svc_src", "celery")
+    span = tracer.trace(c.PRODUCER_ROOT_SPAN, resource=task_name)
+    set_service_and_source(span, service, config.celery, default_service_key="_default_service_producer")
 
     # Store an item called "task span" in case after_task_publish doesn't get called
     core.set_item("task_span", span)