chore(telemetry): add process tags (#15227)

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

Add process_tags to telemetry configuration

## Testing

- There is already a test testing the exact config so we know that if we
are not enabling the process tags, it is not in the config
- Check process tags are in the telemetry configuration when enabled.

Author: dubloom

ddtrace/internal/process_tags/__init__.py

@@ -5,7 +5,7 @@
 from typing import Optional
 
 from ddtrace.internal.logger import get_logger
-from ddtrace.internal.settings._config import config
+from ddtrace.internal.settings.process_tags import process_tags_config as config
 
 
 log = get_logger(__name__)
@@ -45,7 +45,7 @@ def is_allowed_char(char: str) -> str:
 
 
 def generate_process_tags() -> Optional[str]:
-    if not config._process_tags_enabled:
+    if not config.enabled:
         return None
 
     try:

ddtrace/internal/settings/process_tags.py

@@ -0,0 +1,15 @@
+from ddtrace.internal.settings._core import DDConfig
+
+
+class ProcessTagsConfig(DDConfig):
+    __prefix__ = "dd"
+
+    enabled = DDConfig.v(
+        bool,
+        "experimental.propagate.process.tags.enabled",
+        default=False,
+        help="Enables process tags in products payload",
+    )
+
+
+process_tags_config = ProcessTagsConfig()

ddtrace/internal/telemetry/data.py

@@ -7,6 +7,7 @@
 from typing import List  # noqa:F401
 from typing import Tuple  # noqa:F401
 
+from ddtrace.internal import process_tags
 from ddtrace.internal.constants import DEFAULT_SERVICE_NAME
 from ddtrace.internal.packages import get_module_distribution_versions
 from ddtrace.internal.runtime.container import get_container_info
@@ -58,7 +59,7 @@ def _get_application(key):
     """
     service, version, env = key
 
-    return {
+    application = {
         "service_name": service or DEFAULT_SERVICE_NAME,  # mandatory field, can not be empty
         "service_version": version or "",
         "env": env or "",
@@ -69,6 +70,11 @@ def _get_application(key):
         "runtime_version": _format_version_info(sys.implementation.version),
     }
 
+    if p_tags := process_tags.process_tags:
+        application["process_tags"] = p_tags
+
+    return application
+
 
 def update_imported_dependencies(already_imported: Dict[str, str], new_modules: Iterable[str]) -> List[Dict[str, str]]:
     deps = []

tests/internal/test_process_tags.py

@@ -9,7 +9,7 @@
 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 ddtrace.internal.settings.process_tags import process_tags_config as config
 from tests.subprocesstest import run_in_subprocess
 from tests.utils import TracerTestCase
 from tests.utils import process_tag_reload
@@ -79,17 +79,17 @@ def test_normalize_tag(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_enabled = config.enabled
         self._original_process_tags = process_tags.process_tags
 
     def tearDown(self):
-        config._process_tags_enabled = self._original_process_tags_enabled
+        config.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
+        config.enabled = False  # type: ignore[assignment]
         process_tag_reload()
 
         with self.tracer.trace("test"):
@@ -98,7 +98,7 @@ def test_process_tags_deactivated(self):
     @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
+            config.enabled = True  # type: ignore[assignment]
             process_tag_reload()
 
             with self.tracer.trace("parent"):
@@ -108,7 +108,7 @@ def test_process_tags_activated(self):
     @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
+            config.enabled = True  # type: ignore[assignment]
             process_tag_reload()
 
             with self.tracer.trace("span"):
@@ -117,7 +117,7 @@ def test_process_tags_edge_case(self):
     @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
+            config.enabled = True  # type: ignore[assignment]
 
             with self.override_global_config(dict(_telemetry_enabled=False)):
                 with patch("ddtrace.internal.process_tags.log") as mock_log:
@@ -137,7 +137,7 @@ def test_process_tags_error(self):
     @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
+            config.enabled = True  # type: ignore[assignment]
             process_tag_reload()
 
             with self.override_global_config(dict(_partial_flush_enabled=True, _partial_flush_min_spans=2)):

tests/telemetry/test_data.py

@@ -43,6 +43,14 @@ def test_get_application_with_values():
     assert application["env"] == "staging"
 
 
+@pytest.mark.subprocess(env={"DD_EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED": "True"})
+def test_get_application_with_process_tags():
+    from ddtrace.internal.telemetry.data import get_application
+
+    application = get_application("", "", "")
+    assert "process_tags" in application
+
+
 def test_application_with_setenv(run_python_code_in_subprocess, monkeypatch):
     """
     validates the return value of get_application when DD_SERVICE, DD_VERSION, and DD_ENV environment variables are set