chore(runtime_metrics): add process_tags (#15465)

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

Add process_tags to Runtime Metrics

## Testing

- Check that process tags are empty when deactivated
- Check the process tags are created with the right values when
activated

Author: dubloom

ddtrace/internal/runtime/runtime_metrics.py

@@ -18,6 +18,7 @@
 from .metric_collectors import PSUtilRuntimeMetricCollector
 from .tag_collectors import PlatformTagCollector
 from .tag_collectors import PlatformTagCollectorV2
+from .tag_collectors import ProcessTagCollector
 from .tag_collectors import TracerTagCollector
 
 
@@ -59,6 +60,12 @@ class TracerTags(RuntimeCollectorsIterable):
     COLLECTORS = [TracerTagCollector]
 
 
+class ProcessTags(RuntimeCollectorsIterable):
+    # DEV: `None` means to allow all tags generated by ProcessTagsCollector
+    ENABLED = None
+    COLLECTORS = [ProcessTagCollector]
+
+
 class RuntimeMetrics(RuntimeCollectorsIterable):
     ENABLED = DEFAULT_RUNTIME_METRICS
     COLLECTORS = [
@@ -94,6 +101,8 @@ def __init__(self, interval=DEFAULT_RUNTIME_METRICS_INTERVAL, tracer=None, dogst
         else:
             self._platform_tags = self._format_tags(PlatformTags())
 
+        self._process_tags: List[str] = ProcessTags()  # type: ignore[assignment]
+
     @classmethod
     def disable(cls) -> None:
         with cls._lock:
@@ -140,9 +149,9 @@ def enable(
 
     def flush(self) -> None:
         # Ensure runtime metrics have up-to-date tags (ex: service, env, version)
-        rumtime_tags = self._format_tags(TracerTags()) + self._platform_tags
-        log.debug("Sending runtime metrics with the following tags: %s", rumtime_tags)
-        self._dogstatsd_client.constant_tags = rumtime_tags
+        runtime_tags = self._format_tags(TracerTags()) + self._platform_tags + self._process_tags
+        log.debug("Sending runtime metrics with the following tags: %s", runtime_tags)
+        self._dogstatsd_client.constant_tags = runtime_tags
 
         with self._dogstatsd_client:
             for key, value in self._runtime_metrics:

ddtrace/internal/runtime/tag_collectors.py

@@ -5,6 +5,7 @@
 
 from ...constants import ENV_KEY
 from ...constants import VERSION_KEY
+from .. import process_tags
 from ..constants import DEFAULT_SERVICE_NAME
 from .collector import ValueCollector
 from .constants import LANG
@@ -97,3 +98,13 @@ def collect_fn(self, keys):
         tags = super(PlatformTagCollectorV2, self).collect_fn(keys)
         tags.append(("runtime-id", get_runtime_id()))
         return tags
+
+
+class ProcessTagCollector(RuntimeTagCollector):
+    """Tag collector for process tags."""
+
+    def collect_fn(self, keys):
+        # DEV: we do not access direct process_tags_list so we can
+        # reload it in the tests
+        process_tags_list = process_tags.process_tags_list
+        return process_tags_list or []

tests/tracer/runtime/test_tag_collectors.py

@@ -104,3 +104,39 @@ def process_trace(self, _):
     assert values == [
         ("service", "my-service"),
     ], values
+
+
+def test_process_tags_disabled_by_default():
+    ptc = tag_collectors.ProcessTagCollector()
+    tags = list(ptc.collect())
+    assert len(tags) == 0, "Process tags should be empty when not enabled"
+
+
+@pytest.mark.subprocess(env={"DD_EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED": "true"})
+def test_process_tags_enabled():
+    from unittest.mock import patch
+
+    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.runtime import tag_collectors
+    from tests.utils import process_tag_reload
+
+    with patch("sys.argv", ["/path/to/test_script.py"]), patch("os.getcwd", return_value="/path/to/workdir"):
+        process_tag_reload()
+
+        ptc = tag_collectors.ProcessTagCollector()
+        tags: list[str] = ptc.collect()
+        assert len(tags) == 4, f"Expected 4 process tags, got {len(tags)}: {tags}"
+
+        tags_dict = {k: v for k, v in (s.split(":") for s in tags)}
+        assert ENTRYPOINT_NAME_TAG in tags_dict
+        assert ENTRYPOINT_WORKDIR_TAG in tags_dict
+        assert ENTRYPOINT_BASEDIR_TAG in tags_dict
+        assert ENTRYPOINT_TYPE_TAG in tags_dict
+
+        assert tags_dict[ENTRYPOINT_NAME_TAG] == "test_script"
+        assert tags_dict[ENTRYPOINT_WORKDIR_TAG] == "workdir"
+        assert tags_dict[ENTRYPOINT_BASEDIR_TAG] == "to"
+        assert tags_dict[ENTRYPOINT_TYPE_TAG] == "script"