chore(di): add process_tags (#15225)

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

Add process_tags to dynamic instrumentation payload

## Testing

- Check that process tags are not included if deactivated
- Check the right process tags are set in the payload when activated

Author: dubloom

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"] = {

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),