chore(internal): add process_tags to APM payload (#15146)

This PR implements this
[RFC](https://docs.google.com/document/d/1AFdLUuVk70i0bJd5335-RxqsvwAV9ovAqcO2z5mEMbA/edit?pli=1&tab=t.0#heading=h.s9l1lctqlg11)
for the tracing product.

Add process_tags in APM payload. According to the RFC and as the python
tracer supports only v0.4 and v0.5 the tag `_dd.tags.process` is set in
the first first span of each trace chunk set.

The set tag are:
- `entrypoint.workdir`. The working dir name, should be the last path
segment.
- `entrypoint.name`. The entrypoint of the application. It is the the
script name.
- `entrypoint.type`. I'd assume that it can be only script for Python.
- `entrypoint.basedir`. The base dir is the directory where the called
executable resides. Should be the last path segment.

Author: dubloom

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/debugging/_encoding.py

@@ -20,6 +20,7 @@
 from ddtrace.debugging._signal.log import LogSignal
 from ddtrace.debugging._signal.snapshot import Snapshot
 from ddtrace.internal import forksafe
+from ddtrace.internal import process_tags
 from ddtrace.internal._encoding import BufferFull
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.utils.formats import format_trace_id
@@ -113,6 +114,9 @@ def _build_log_track_payload(
         "timestamp": int(signal.timestamp * 1e3),  # milliseconds,
     }
 
+    if p_tags := process_tags.process_tags:
+        payload["process_tags"] = p_tags
+
     # Add the correlation IDs if available
     if context is not None and context.trace_id is not None:
         payload["dd"] = {

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/debugging/test_encoding.py

@@ -250,6 +250,61 @@ def test_batch_json_encoder():
     assert queue.count == 0
 
 
+def test_process_tags_are_not_included_by_default():
+    s = Snapshot(
+        probe=create_snapshot_line_probe(probe_id="batch-test", source_file="foo.py", line=42),
+        frame=inspect.currentframe(),
+        thread=threading.current_thread(),
+    )
+    buffer_size = 30 * (1 << 20)
+    queue = SignalQueue(encoder=LogSignalJsonEncoder(None), buffer_size=buffer_size)
+
+    s.line({})
+
+    queue = SignalQueue(encoder=LogSignalJsonEncoder("test-service"))
+    queue.put(s)
+    data = queue.flush()
+    assert data is not None
+    payload, _ = data
+    decoded = json.loads(payload.decode())
+    assert "process_tags" not in decoded[0]
+
+
+@pytest.mark.subprocess(
+    env=dict(
+        DD_EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED="true",
+    )
+)
+def test_process_tags_are_included():
+    import inspect
+    import json
+    import threading
+
+    from ddtrace.debugging._encoding import LogSignalJsonEncoder
+    from ddtrace.debugging._encoding import SignalQueue
+    from ddtrace.debugging._signal.snapshot import Snapshot
+    from tests.debugging.utils import create_snapshot_line_probe
+
+    s = Snapshot(
+        probe=create_snapshot_line_probe(probe_id="batch-test", source_file="foo.py", line=42),
+        frame=inspect.currentframe(),
+        thread=threading.current_thread(),
+    )
+    buffer_size = 30 * (1 << 20)
+    queue = SignalQueue(encoder=LogSignalJsonEncoder(None), buffer_size=buffer_size)
+
+    s.line({})
+
+    queue = SignalQueue(encoder=LogSignalJsonEncoder("test-service"))
+    queue.put(s)
+    data = queue.flush()
+    assert data is not None
+    payload, _ = data
+    decoded = json.loads(payload.decode())
+
+    assert "process_tags" in decoded[0]
+
+
 def test_batch_flush_reencode():
     s = Snapshot(
         probe=create_snapshot_line_probe(probe_id="batch-test", source_file="foo.py", line=42),

tests/internal/test_process_tags.py

@@ -0,0 +1,164 @@
+from unittest.mock import patch
+
+import pytest
+
+from ddtrace.internal import process_tags
+from ddtrace.internal.constants import PROCESS_TAGS
+from ddtrace.internal.process_tags import ENTRYPOINT_BASEDIR_TAG
+from ddtrace.internal.process_tags import ENTRYPOINT_NAME_TAG
+from ddtrace.internal.process_tags import ENTRYPOINT_TYPE_TAG
+from ddtrace.internal.process_tags import ENTRYPOINT_WORKDIR_TAG
+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
+
+    @run_in_subprocess(env_overrides=dict(DD_EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED="True"))
+    def test_process_tags_activated_with_env(self):
+        with self.tracer.trace("test"):
+            pass
+
+        span = self.get_spans()[0]
+
+        assert span is not None
+        assert PROCESS_TAGS in span._meta
+
+        process_tags = span._meta[PROCESS_TAGS]
+        assert ENTRYPOINT_BASEDIR_TAG in process_tags
+        assert ENTRYPOINT_NAME_TAG in process_tags
+        assert ENTRYPOINT_TYPE_TAG in process_tags
+        assert ENTRYPOINT_WORKDIR_TAG in process_tags

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
+  }]]