chore(internal): add process_tags to APM payload

ddtrace/_trace/processor/__init__.py

@@ -14,11 +14,13 @@
 from ddtrace.constants import _APM_ENABLED_METRIC_KEY as MK_APM_ENABLED
 from ddtrace.constants import _SINGLE_SPAN_SAMPLING_MECHANISM
 from ddtrace.internal import gitmetadata
+from ddtrace.internal import process_tags
 from ddtrace.internal import telemetry
 from ddtrace.internal.constants import COMPONENT
 from ddtrace.internal.constants import HIGHER_ORDER_TRACE_ID_BITS
 from ddtrace.internal.constants import LAST_DD_PARENT_ID_KEY
 from ddtrace.internal.constants import MAX_UINT_64BITS
+from ddtrace.internal.constants import PROCESS_TAGS
 from ddtrace.internal.constants import SAMPLING_DECISION_TRACE_TAG_KEY
 from ddtrace.internal.constants import SamplingMechanism
 from ddtrace.internal.logger import get_logger
@@ -250,6 +252,8 @@ def process_trace(self, trace: List[Span]) -> Optional[List[Span]]:
             span._update_tags_from_context()
             self._set_git_metadata(span)
             span._set_tag_str("language", "python")
+            if p_tags := process_tags.process_tags:
+                span._set_tag_str(PROCESS_TAGS, p_tags)
             # for 128 bit trace ids
             if span.trace_id > MAX_UINT_64BITS:
                 trace_id_hob = _get_64_highest_order_bits_as_hex(span.trace_id)

ddtrace/internal/constants.py

@@ -71,6 +71,7 @@
 HTTP_REQUEST_PATH_PARAMETER = "http.request.path.parameter"
 REQUEST_PATH_PARAMS = "http.request.path_params"
 STATUS_403_TYPE_AUTO = {"status_code": 403, "type": "auto"}
+PROCESS_TAGS = "_dd.tags.process"
 
 CONTAINER_ID_HEADER_NAME = "Datadog-Container-Id"
 

ddtrace/internal/process_tags/__init__.py

@@ -0,0 +1,68 @@
+import os
+from pathlib import Path
+import re
+import sys
+from typing import Optional
+
+from ddtrace.internal.logger import get_logger
+from ddtrace.internal.settings._config import config
+
+
+log = get_logger(__name__)
+
+ENTRYPOINT_NAME_TAG = "entrypoint.name"
+ENTRYPOINT_WORKDIR_TAG = "entrypoint.workdir"
+ENTRYPOINT_TYPE_TAG = "entrypoint.type"
+ENTRYPOINT_TYPE_SCRIPT = "script"
+ENTRYPOINT_BASEDIR_TAG = "entrypoint.basedir"
+
+_CONSECUTIVE_UNDERSCORES_PATTERN = re.compile(r"_{2,}")
+_ALLOWED_CHARS = _ALLOWED_CHARS = frozenset("abcdefghijklmnopqrstuvwxyz0123456789/:._-")
+MAX_LENGTH = 200
+
+
+def normalize_tag_value(value: str) -> str:
+    # we copy the behavior of the agent which
+    # checks the size on the original value and not on
+    # an intermediary normalized step
+    if len(value) > MAX_LENGTH:
+        value = value[:MAX_LENGTH]
+
+    result = value.lower()
+
+    def is_allowed_char(char: str) -> str:
+        # ASCII alphanumeric and special chars: / : . _ -
+        if char in _ALLOWED_CHARS:
+            return char
+        # Unicode letters and digits
+        if char.isalpha() or char.isdigit():
+            return char
+        return "_"
+
+    result = "".join(is_allowed_char(char) for char in result)
+    result = _CONSECUTIVE_UNDERSCORES_PATTERN.sub("_", result)
+    return result.strip("_")
+
+
+def generate_process_tags() -> Optional[str]:
+    if not config._process_tags_enabled:
+        return None
+
+    try:
+        return ",".join(
+            f"{key}:{normalize_tag_value(value)}"
+            for key, value in sorted(
+                [
+                    (ENTRYPOINT_WORKDIR_TAG, os.path.basename(os.getcwd())),
+                    (ENTRYPOINT_BASEDIR_TAG, Path(sys.argv[0]).resolve().parent.name),
+                    (ENTRYPOINT_NAME_TAG, os.path.splitext(os.path.basename(sys.argv[0]))[0]),
+                    (ENTRYPOINT_TYPE_TAG, ENTRYPOINT_TYPE_SCRIPT),
+                ]
+            )
+        )
+    except Exception as e:
+        log.debug("failed to get process_tags: %s", e)
+        return None
+
+
+process_tags = generate_process_tags()

ddtrace/internal/settings/_config.py

@@ -665,6 +665,9 @@ def __init__(self):
         self._trace_resource_renaming_always_simplified_endpoint = _get_config(
             "DD_TRACE_RESOURCE_RENAMING_ALWAYS_SIMPLIFIED_ENDPOINT", default=False, modifier=asbool
         )
+        self._process_tags_enabled = _get_config(
+            "DD_EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED", default=False, modifier=asbool
+        )
 
         # Long-running span interval configurations
         # Only supported for Ray spans for now

tests/internal/test_process_tags.py

@@ -0,0 +1,143 @@
+from unittest.mock import patch
+
+import pytest
+
+from ddtrace.internal import process_tags
+from ddtrace.internal.process_tags import normalize_tag_value
+from ddtrace.internal.settings._config import config
+from tests.subprocesstest import run_in_subprocess
+from tests.utils import TracerTestCase
+from tests.utils import process_tag_reload
+
+
+TEST_SCRIPT_PATH = "/path/to/test_script.py"
+TEST_WORKDIR_PATH = "/path/to/workdir"
+
+
+@pytest.mark.parametrize(
+    "input_tag,expected",
+    [
+        ("#test_starting_hash", "test_starting_hash"),
+        ("TestCAPSandSuch", "testcapsandsuch"),
+        ("Test Conversion Of Weird !@#$%^&**() Characters", "test_conversion_of_weird_characters"),
+        ("$#weird_starting", "weird_starting"),
+        ("allowed:c0l0ns", "allowed:c0l0ns"),
+        ("1love", "1love"),
+        ("/love2", "/love2"),
+        ("ünicöde", "ünicöde"),
+        ("ünicöde:metäl", "ünicöde:metäl"),
+        ("Data🐨dog🐶 繋がっ⛰てて", "data_dog_繋がっ_てて"),
+        (" spaces   ", "spaces"),
+        (" #hashtag!@#spaces #__<>#  ", "hashtag_spaces"),
+        (":testing", ":testing"),
+        ("_foo", "foo"),
+        (":::test", ":::test"),
+        ("contiguous_____underscores", "contiguous_underscores"),
+        ("foo_", "foo"),
+        ("\u017Fodd_\u017Fcase\u017F", "\u017Fodd_\u017Fcase\u017F"),
+        ("", ""),
+        (" ", ""),
+        ("ok", "ok"),
+        ("™Ö™Ö™™Ö™", "ö_ö_ö"),
+        ("AlsO:ök", "also:ök"),
+        (":still_ok", ":still_ok"),
+        ("___trim", "trim"),
+        ("12.:trim@", "12.:trim"),
+        ("12.:trim@@", "12.:trim"),
+        ("fun:ky__tag/1", "fun:ky_tag/1"),
+        ("fun:ky@tag/2", "fun:ky_tag/2"),
+        ("fun:ky@@@tag/3", "fun:ky_tag/3"),
+        ("tag:1/2.3", "tag:1/2.3"),
+        ("---fun:k####y_ta@#g/1_@@#", "---fun:k_y_ta_g/1"),
+        ("AlsO:œ#@ö))œk", "also:œ_ö_œk"),
+        ("test\x99\x8faaa", "test_aaa"),
+        ("test\x99\x8f", "test"),
+        ("a" * 888, "a" * 200),
+        ("a" + "🐶" * 799 + "b", "a"),
+        ("a" + "\ufffd", "a"),
+        ("a" + "\ufffd" + "\ufffd", "a"),
+        ("a" + "\ufffd" + "\ufffd" + "b", "a_b"),
+        (
+            "A00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
+            "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
+            " 000000000000",
+            "a00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
+            "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
+            "_0",
+        ),
+    ],
+)
+def test_normalize_tag(input_tag, expected):
+    assert normalize_tag_value(input_tag) == expected
+
+
+class TestProcessTags(TracerTestCase):
+    def setUp(self):
+        super(TestProcessTags, self).setUp()
+        self._original_process_tags_enabled = config._process_tags_enabled
+        self._original_process_tags = process_tags.process_tags
+
+    def tearDown(self):
+        config._process_tags_enabled = self._original_process_tags_enabled
+        process_tags.process_tags = self._original_process_tags
+        super().tearDown()
+
+    @pytest.mark.snapshot
+    def test_process_tags_deactivated(self):
+        config._process_tags_enabled = False
+        process_tag_reload()
+
+        with self.tracer.trace("test"):
+            pass
+
+    @pytest.mark.snapshot
+    def test_process_tags_activated(self):
+        with patch("sys.argv", [TEST_SCRIPT_PATH]), patch("os.getcwd", return_value=TEST_WORKDIR_PATH):
+            config._process_tags_enabled = True
+            process_tag_reload()
+
+            with self.tracer.trace("parent"):
+                with self.tracer.trace("child"):
+                    pass
+
+    @pytest.mark.snapshot
+    def test_process_tags_edge_case(self):
+        with patch("sys.argv", ["/test_script"]), patch("os.getcwd", return_value=TEST_WORKDIR_PATH):
+            config._process_tags_enabled = True
+            process_tag_reload()
+
+            with self.tracer.trace("span"):
+                pass
+
+    @pytest.mark.snapshot
+    def test_process_tags_error(self):
+        with patch("sys.argv", []), patch("os.getcwd", return_value=TEST_WORKDIR_PATH):
+            config._process_tags_enabled = True
+
+            with self.override_global_config(dict(_telemetry_enabled=False)):
+                with patch("ddtrace.internal.process_tags.log") as mock_log:
+                    process_tag_reload()
+
+                    with self.tracer.trace("span"):
+                        pass
+
+                    # Check if debug log was called
+                    mock_log.debug.assert_called_once()
+                    call_args = mock_log.debug.call_args[0]
+                    assert (
+                        "failed to get process_tags" in call_args[0]
+                    ), f"Expected error message not found. Got: {call_args[0]}"
+
+    @pytest.mark.snapshot
+    @run_in_subprocess(env_overrides=dict(DD_TRACE_PARTIAL_FLUSH_ENABLED="true", DD_TRACE_PARTIAL_FLUSH_MIN_SPANS="2"))
+    def test_process_tags_partial_flush(self):
+        with patch("sys.argv", [TEST_SCRIPT_PATH]), patch("os.getcwd", return_value=TEST_WORKDIR_PATH):
+            config._process_tags_enabled = True
+            process_tag_reload()
+
+            with self.override_global_config(dict(_partial_flush_enabled=True, _partial_flush_min_spans=2)):
+                with self.tracer.trace("parent"):
+                    with self.tracer.trace("child1"):
+                        pass
+                    with self.tracer.trace("child2"):
+                        pass

tests/snapshots/tests.internal.test_process_tags.TestProcessTags.test_edge_case.json

@@ -0,0 +1,26 @@
+[[
+  {
+    "name": "span",
+    "service": "tests.internal",
+    "resource": "span",
+    "trace_id": 0,
+    "span_id": 1,
+    "parent_id": 0,
+    "type": "",
+    "error": 0,
+    "meta": {
+      "_dd.p.dm": "-0",
+      "_dd.p.tid": "6911da3a00000000",
+      "_dd.tags.process": "entrypoint.basedir:,entrypoint.name:test_script,entrypoint.type:script,entrypoint.workdir:workdir",
+      "language": "python",
+      "runtime-id": "c9342b8003de45feb0bf56d32ece46a1"
+    },
+    "metrics": {
+      "_dd.top_level": 1,
+      "_dd.tracer_kr": 1.0,
+      "_sampling_priority_v1": 1,
+      "process_id": 605
+    },
+    "duration": 105292,
+    "start": 1762777658431833668
+  }]]

tests/snapshots/tests.internal.test_process_tags.TestProcessTags.test_process_tags_activated.json

@@ -0,0 +1,38 @@
+[[
+  {
+    "name": "parent",
+    "service": "tests.internal",
+    "resource": "parent",
+    "trace_id": 0,
+    "span_id": 1,
+    "parent_id": 0,
+    "type": "",
+    "error": 0,
+    "meta": {
+      "_dd.p.dm": "-0",
+      "_dd.p.tid": "6911dc5a00000000",
+      "_dd.tags.process": "entrypoint.basedir:to,entrypoint.name:test_script,entrypoint.type:script,entrypoint.workdir:workdir",
+      "language": "python",
+      "runtime-id": "2d5de91f8dd9442cad7faca5554a09f1"
+    },
+    "metrics": {
+      "_dd.top_level": 1,
+      "_dd.tracer_kr": 1.0,
+      "_sampling_priority_v1": 1,
+      "process_id": 605
+    },
+    "duration": 231542,
+    "start": 1762778202287875128
+  },
+     {
+       "name": "child",
+       "service": "tests.internal",
+       "resource": "child",
+       "trace_id": 0,
+       "span_id": 2,
+       "parent_id": 1,
+       "type": "",
+       "error": 0,
+       "duration": 55500,
+       "start": 1762778202287999128
+     }]]

tests/snapshots/tests.internal.test_process_tags.TestProcessTags.test_process_tags_deactivated.json

@@ -0,0 +1,25 @@
+[[
+  {
+    "name": "test",
+    "service": "tests.internal",
+    "resource": "test",
+    "trace_id": 0,
+    "span_id": 1,
+    "parent_id": 0,
+    "type": "",
+    "error": 0,
+    "meta": {
+      "_dd.p.dm": "-0",
+      "_dd.p.tid": "6911dc5a00000000",
+      "language": "python",
+      "runtime-id": "2d5de91f8dd9442cad7faca5554a09f1"
+    },
+    "metrics": {
+      "_dd.top_level": 1,
+      "_dd.tracer_kr": 1.0,
+      "_sampling_priority_v1": 1,
+      "process_id": 605
+    },
+    "duration": 22292,
+    "start": 1762778202327669586
+  }]]

tests/snapshots/tests.internal.test_process_tags.TestProcessTags.test_process_tags_edge_case.json

@@ -0,0 +1,26 @@
+[[
+  {
+    "name": "span",
+    "service": "tests.internal",
+    "resource": "span",
+    "trace_id": 0,
+    "span_id": 1,
+    "parent_id": 0,
+    "type": "",
+    "error": 0,
+    "meta": {
+      "_dd.p.dm": "-0",
+      "_dd.p.tid": "6911dc5a00000000",
+      "_dd.tags.process": "entrypoint.basedir:,entrypoint.name:test_script,entrypoint.type:script,entrypoint.workdir:workdir",
+      "language": "python",
+      "runtime-id": "2d5de91f8dd9442cad7faca5554a09f1"
+    },
+    "metrics": {
+      "_dd.top_level": 1,
+      "_dd.tracer_kr": 1.0,
+      "_sampling_priority_v1": 1,
+      "process_id": 605
+    },
+    "duration": 35458,
+    "start": 1762778202321224878
+  }]]

tests/snapshots/tests.internal.test_process_tags.TestProcessTags.test_process_tags_error.json

@@ -0,0 +1,25 @@
+[[
+  {
+    "name": "span",
+    "service": "tests.internal",
+    "resource": "span",
+    "trace_id": 0,
+    "span_id": 1,
+    "parent_id": 0,
+    "type": "",
+    "error": 0,
+    "meta": {
+      "_dd.p.dm": "-0",
+      "_dd.p.tid": "6911db6e00000000",
+      "language": "python",
+      "runtime-id": "c59cb90aad3246579bc4421d1cca07c8"
+    },
+    "metrics": {
+      "_dd.top_level": 1,
+      "_dd.tracer_kr": 1.0,
+      "_sampling_priority_v1": 1,
+      "process_id": 605
+    },
+    "duration": 102833,
+    "start": 1762777966446950922
+  }]]