fix(ci-visibility): add dd_origin propagation to CIVisibilityEncoder (#5541)

This PR builds on #5489 by adding dd_origin propagation to the
CIVisibilityEncoder, which is necessary for CI Visibility. This is not a
new feature as this functionality is already present in the regular
encoder.

This doesn't need a changelog since this is internal and is a fix to an
unreleased change.

Author: Yun-Kim

ddtrace/internal/ci_visibility/encoder.py

@@ -22,13 +22,14 @@ def __init__(self, *args):
         super(CIVisibilityEncoderV01, self).__init__()
         self._lock = threading.RLock()
         self._init_buffer()
+        self._metadata = {}
 
     def __len__(self):
         with self._lock:
             return len(self.buffer)
 
     def set_metadata(self, metadata):
-        self._metadata = metadata or dict()
+        self._metadata.update(metadata)
 
     def _init_buffer(self):
         with self._lock:
@@ -48,21 +49,25 @@ def encode(self):
             return payload
 
     def _build_payload(self, traces):
-        normalized_spans = [CIVisibilityEncoderV01._convert_span(span) for trace in traces for span in trace]
+        normalized_spans = [
+            CIVisibilityEncoderV01._convert_span(span, trace[0].context.dd_origin) for trace in traces for span in trace
+        ]
         self._metadata = {k: v for k, v in self._metadata.items() if k in self.ALLOWED_METADATA_KEYS}
         # TODO: Split the events in several payloads as needed to avoid hitting the intake's maximum payload size.
         return msgpack_packb(
             {"version": self.PAYLOAD_FORMAT_VERSION, "metadata": {"*": self._metadata}, "events": normalized_spans}
         )
 
     @staticmethod
-    def _convert_span(span):
-        # type: (Span) -> Dict[str, Any]
+    def _convert_span(span, dd_origin):
+        # type: (Span, str) -> Dict[str, Any]
         sp = JSONEncoderV2._convert_span(span)
         sp["type"] = span.span_type
         sp["duration"] = span.duration_ns
         sp["meta"] = dict(sorted(span._meta.items()))
         sp["metrics"] = dict(sorted(span._metrics.items()))
+        if dd_origin is not None:
+            sp["meta"].update({"_dd.origin": dd_origin})
         if span.span_type == "test":
             event_type = "test"
         else:

tests/contrib/pytest/test_pytest.py

@@ -12,6 +12,7 @@
 from ddtrace.ext import ci
 from ddtrace.ext import test
 from ddtrace.internal.ci_visibility import CIVisibility
+from ddtrace.internal.ci_visibility.encoder import CIVisibilityEncoderV01
 from tests.utils import DummyCIVisibilityWriter
 from tests.utils import TracerTestCase
 from tests.utils import override_env
@@ -547,7 +548,6 @@ def test_service(ddtracer):
         file_name = os.path.basename(py_file.strpath)
         rec = self.inline_run("--ddtrace", file_name)
         rec.assertoutcome(passed=1)
-
         spans = self.pop_spans()
         # Check if spans tagged with dd_origin after encoding and decoding as the tagging occurs at encode time
         encoder = self.tracer.encoder
@@ -558,6 +558,15 @@ def test_service(ddtracer):
         for span in decoded_trace:
             assert span[b"meta"][b"_dd.origin"] == b"ciapp-test"
 
+        ci_agentless_encoder = CIVisibilityEncoderV01(0, 0)
+        ci_agentless_encoder.put(spans)
+        trace = ci_agentless_encoder.encode()
+        decoded_trace = self.tracer.encoder._decode(trace)
+        assert len(decoded_trace[b"events"]) == 4
+        for event in decoded_trace[b"events"]:
+            assert event[b"content"][b"meta"][b"_dd.origin"] == b"ciapp-test"
+        pass
+
     def test_pytest_doctest_module(self):
         """Test that pytest with doctest works as expected."""
         py_file = self.testdir.makepyfile(