chore(tracing): improve debug logging for v05 encoding errors [backport 1.20] (#7173)

Backport 85d5ec5 from #7099 to 1.20.

There is an edge case were the string table used by the v0.5 encoder
swaps entries. This corrupts trace payloads. In this PR I will add debug
logs that will help us reproduce the behavior that triggered
incident-24455.

## RISK

This PR maybe increase memory usage when v0.5 is enabled. With this
change we will hold a reference to a all encoded spans until the encoded
bytes are flushed and we will log trace payloads that can have a size of
up to 50MB when we traces are misencoded.

## Checklist

- [x] Change(s) are motivated and described in the PR description.
- [x] Testing strategy is described if automated tests are not included
in the PR.
- [x] Risk is outlined (performance impact, potential for breakage,
maintainability, etc).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] [Library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
are followed. If no release note is required, add label
`changelog/no-changelog`.
- [x] Documentation is included (in-code, generated user docs, [public
corp docs](https://github.com/DataDog/documentation/)).
- [x] Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist

- [x] Title is accurate.
- [x] No unnecessary changes are introduced.
- [x] Description motivates each change.
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes unless absolutely necessary.
- [x] Testing strategy adequately addresses listed risk(s).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] Release note makes sense to a user of the library.
- [x] Reviewer has explicitly acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment.
- [x] Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)
- [x] If this PR touches code that signs or publishes builds or
packages, or handles credentials of any kind, I've requested a review
from `@DataDog/security-design-and-guidance`.
- [x] This PR doesn't touch any of that.

---------

Co-authored-by: Munir Abdinur <[email protected]>

Author: github-actions[bot]

ddtrace/internal/_encoding.pyi

@@ -16,6 +16,9 @@ class BufferFull(Exception):
 class BufferItemTooLarge(Exception):
     pass
 
+class EncodingValidationError(Exception):
+    pass
+
 class BufferedEncoder(object):
     max_size: int
     max_item_size: int

ddtrace/internal/_encoding.pyx

@@ -67,6 +67,10 @@ class BufferItemTooLarge(Exception):
     pass
 
 
+class EncodingValidationError(Exception):
+    pass
+
+
 cdef inline const char * string_to_buff(str s):
     IF PY_MAJOR_VERSION >= 3:
         return PyUnicode_AsUTF8(s)
@@ -268,6 +272,12 @@ cdef class MsgpackStringTable(StringTable):
         self._sp_len = self.pk.length
         self._sp_id = self._next_id
 
+    @property
+    def table_values(self):
+        return "_sp_len:{}\n_sp_id:{}\npk_buf:{}\npk_buff_size:{}\nmax_size:{}\ntable:{}".format(
+            self._sp_len, self._sp_id, self.get_bytes(), self.pk.buf_size, self.max_size, self._table
+        )
+
     cdef rollback(self):
         if self._sp_len > 0:
             self.pk.length = self._sp_len
@@ -711,9 +721,11 @@ cdef class MsgpackEncoderV03(MsgpackEncoderBase):
 
 cdef class MsgpackEncoderV05(MsgpackEncoderBase):
     cdef MsgpackStringTable _st
+    cdef dict _encoded_spans
 
     def __cinit__(self, size_t max_size, size_t max_item_size):
         self._st = MsgpackStringTable(max_size)
+        self._encoded_spans = {}
 
     cpdef flush(self):
         with self._lock:
@@ -722,9 +734,16 @@ cdef class MsgpackEncoderV05(MsgpackEncoderBase):
                     PyLong_FromLong(<long> self.get_buffer()),
                     <Py_ssize_t> super(MsgpackEncoderV05, self).size,
                 )
-                return self._st.flush()
+                v05bytes = self._st.flush()
+                self._verify_encoding(v05bytes)
+                return v05bytes
             finally:
                 self._reset_buffer()
+                self._encoded_spans = {}
+
+    @property
+    def stable(self):
+        return self._st
 
     @property
     def size(self):
@@ -737,6 +756,8 @@ cdef class MsgpackEncoderV05(MsgpackEncoderBase):
             try:
                 self._st.savepoint()
                 super(MsgpackEncoderV05, self).put(trace)
+                for span in trace:
+                    self._encoded_spans[span.span_id] = span
             except Exception:
                 self._st.rollback()
                 raise
@@ -831,6 +852,80 @@ cdef class MsgpackEncoderV05(MsgpackEncoderBase):
 
         return 0
 
+    cpdef _verify_encoding(self, encoded_bytes):
+        import msgpack
+
+        from ddtrace import config
+
+        if not config._trace_writer_log_err_payload:
+            return
+
+        unpacked = msgpack.unpackb(encoded_bytes, raw=True, strict_map_key=False)
+        if not unpacked or not unpacked[0]:
+            return unpacked
+
+        table, packed_traces = unpacked
+        for trace in packed_traces:
+            for span in trace:
+                og_span = self._encoded_spans.get(span[4])  # correlate encoded span to a span in span_dict
+                # structure of v0.5 encoded spans
+                # 		 0: Service   (uint32)
+                # 		 1: Name      (uint32)
+                # 		 2: Resource  (uint32)
+                # 		 3: TraceID   (uint64)
+                # 		 4: SpanID    (uint64)
+                # 		 5: ParentID  (uint64)
+                # 		 6: Start     (int64)
+                # 		 7: Duration  (int64)
+                # 		 8: Error     (int32)
+                # 		 9: Meta      (map[uint32]uint32)
+                # 		10: Metrics   (map[uint32]float64)
+                # 		11: Type      (uint32)
+                try:
+                    assert og_span.service == (table[span[0]].decode() or None), "misencoded service: {}".format(
+                        table[span[0]]
+                    )
+                    assert og_span.name == (table[span[1]].decode() or None), "misencoded name: {}".format(
+                        table[span[1]]
+                    )
+                    assert og_span.resource == (table[span[2]].decode() or None), "misencoded resource: {}".format(
+                        table[span[2]]
+                    )
+                    assert og_span._trace_id_64bits == (span[3] or None), "misencoded trace id: {}".format(span[3])
+                    assert og_span.span_id == (span[4] or None), "misencoded span id: {}".format(span[4])
+                    assert og_span.parent_id == (span[5] or None), "misencoded parent id: {}".format(span[5])
+                    assert og_span.start_ns == span[6], "misencoded start: {}".format(span[6])
+                    assert og_span.duration_ns == (span[7] or None), "misencoded duration: {}".format(span[7])
+                    assert og_span.error == span[8], "misencoded error type: {}".format(span[8])
+
+                    for k, v in span[9].items():
+                        k = table[k].decode()
+                        v = table[v].decode()
+                        if "dropped string of length" in k or "dropped string of length" in v:
+                            continue
+                        assert og_span._meta[k] == v,  "misencoded tag: k={} v={}".format(k, v)
+
+                    for k, v in span[10].items():
+                        k = table[k].decode()
+                        if "dropped string of length" in k:
+                            continue
+                        assert og_span._metrics[k] == v, "misencoded metric: k={} v={}".format(k, v)
+
+                    assert og_span.span_type == (table[span[11]].decode() or None), "misencoded span type: {}".format(
+                        table[span[11]]
+                    )
+                except Exception as e:
+                    eve = EncodingValidationError(
+                        str(e) + "\nDecoded Span does not match encoded span: {}".format(og_span._pprint())
+                    )
+                    setattr(
+                        eve,
+                        "_debug_message",
+                        "Malformed String table values\ntable:{}\ntraces:{}\nencoded_bytes:{}\nspans:{}".format(
+                            table, packed_traces, encoded_bytes, self._encoded_spans
+                        ),
+                    )
+                    raise eve
 
 cdef class Packer(object):
     """Slightly modified version of the v0.6.2 msgpack Packer

ddtrace/internal/writer/writer.py

@@ -30,6 +30,7 @@
 from ...internal.utils.time import StopWatch
 from .._encoding import BufferFull
 from .._encoding import BufferItemTooLarge
+from .._encoding import EncodingValidationError
 from ..agent import get_connection
 from ..constants import _HTTPLIB_NO_TRACE_REQUEST
 from ..encoding import JSONEncoderV2
@@ -395,6 +396,11 @@ def _flush_queue_with_client(self, client, raise_exc=False):
             encoded = client.encoder.encode()
             if encoded is None:
                 return
+        except EncodingValidationError as e:
+            log.error("Encoding Error (or span was modified after finish): %s", str(e))
+            if hasattr(e, "_debug_message"):
+                log.debug(e._debug_message)
+            return
         except Exception:
             log.error("failed to encode trace with encoder %r", client.encoder, exc_info=True)
             self._metrics_dist("encoder.dropped.traces", n_traces)

tests/integration/test_integration.py

@@ -529,6 +529,31 @@ def encode_traces(self, traces):
     )
 
 
+@skip_if_testagent
+@pytest.mark.subprocess(
+    env={
+        "_DD_TRACE_WRITER_LOG_ERROR_PAYLOADS": "true",
+        "DD_TRACE_API_VERSION": "v0.5",
+        "DD_TRACE_WRITER_INTERVAL_SECONDS": "1000",
+    }
+)
+def test_trace_with_invalid_encoding_for_v05_payload():
+    import mock
+
+    from ddtrace import tracer
+    from tests.utils import AnyStr
+    from tests.utils import AnyStringWithText
+
+    with mock.patch("ddtrace.internal.writer.writer.log") as log:
+        span = tracer.trace("name")
+        span.finish()
+        span.name = "new_name"
+        tracer.flush()
+
+    log.error.assert_has_calls([mock.call("Encoding Error (or span was modified after finish): %s", AnyStr())])
+    log.debug.assert_has_calls([mock.call(AnyStringWithText("Malformed String table values"))])
+
+
 def test_trace_with_failing_encoder_generates_error_log():
     class ExceptionBadEncoder(BadEncoder):
         def encode(self):

tests/tracer/test_encoders.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 import contextlib
 import json
+import os
 import random
 import string
 import threading
@@ -21,6 +22,7 @@
 from ddtrace.ext.ci import CI_APP_TEST_ORIGIN
 from ddtrace.internal._encoding import BufferFull
 from ddtrace.internal._encoding import BufferItemTooLarge
+from ddtrace.internal._encoding import EncodingValidationError
 from ddtrace.internal._encoding import ListStringTable
 from ddtrace.internal._encoding import MsgpackStringTable
 from ddtrace.internal.compat import msgpack_type
@@ -33,6 +35,7 @@
 from ddtrace.internal.encoding import _EncoderBase
 from ddtrace.span import Span
 from tests.utils import DummyTracer
+from tests.utils import override_global_config
 
 
 _ORIGIN_KEY = ORIGIN_KEY.encode()
@@ -700,3 +703,139 @@ def test_json_encoder_traces_bytes():
         assert u"\ufffdspan.a" == span_a["name"], span_a["name"]
         assert u"\x80span.b" == span_b["name"]
         assert u"\ufffdspan.b" == span_c["name"]
+
+
+@pytest.mark.skipif(
+    os.getenv("PYTHONOPTIMIZE", "").lower() in ("1", "t", "true") or six.PY2,
+    reason="Python optimize removes assertions from cython code. PY2 byte string formatting fails assertions",
+)
+def test_verifying_v05_payloads():
+    string_table_size = 4 * (1 << 12)
+    encoder = MsgpackEncoderV05(string_table_size, string_table_size)
+
+    # Ensure EncodingValidationError is not raised when trace fields are encoded as expected
+    with override_global_config({"_trace_writer_log_err_payload": True}):
+        traces = [[Span("name", "service", "resource", "type") for _ in range(5)] for _ in range(100)]
+        for trace in traces:
+            for s in trace:
+                s._meta = {
+                    "app": "ac_query",
+                    "language": "python",
+                    "key_for_long_string": "very_long_string_that_will_be_dropped" * int(string_table_size * 0.1),
+                }
+                s._metrics = {
+                    "zqSCqAYiBgjmqYKoBiohcCKwCagB": 7,
+                    "_sampling_priority_v1": 0,
+                    "very_long_string_that_will_be_dropped" * int(string_table_size * 0.1): 1,
+                }
+
+        encoded = []
+        for trace in traces:
+            try:
+                encoder.put(trace)
+                encoded.append(trace)
+            except BufferFull:
+                pass
+        assert 0 < len(encoded) < len(traces), "Ensures BufferFull is raised and only a subset of traces are encoded"
+        assert encoder.encode()
+
+    # Ensure EncodingValidationError is raised when the encoded span name does not match the span name
+    with override_global_config({"_trace_writer_log_err_payload": True}):
+        with pytest.raises(EncodingValidationError) as e:
+            og_span = Span("name", "service", "resource", "type")
+            encoder.put([og_span])
+            og_span.name = "new_name"
+            encoder.encode()
+        assert "misencoded name: b'name'" in e.value.args[0]
+
+    # Ensure EncodingValidationError is raised when the encoded service does not match the span service
+    with override_global_config({"_trace_writer_log_err_payload": True}):
+        with pytest.raises(EncodingValidationError) as e:
+            og_span = Span("name", "service", "resource", "type")
+            encoder.put([og_span])
+            og_span.service = "new_service"
+            encoder.encode()
+        assert "misencoded service: b'service'" in e.value.args[0]
+
+    # Ensure EncodingValidationError is raised when the encoded resource does not match the span resource
+    with override_global_config({"_trace_writer_log_err_payload": True}):
+        with pytest.raises(EncodingValidationError) as e:
+            og_span = Span("name", "service", "resource", "type")
+            encoder.put([og_span])
+            og_span.resource = "new_resource"
+            encoder.encode()
+        assert "misencoded resource: b'resource'" in e.value.args[0]
+
+    # Ensure EncodingValidationError is raised when the encoded duration does not match
+    with override_global_config({"_trace_writer_log_err_payload": True}):
+        with pytest.raises(EncodingValidationError) as e:
+            og_span = Span("name", "service", "resource", "type")
+            encoder.put([og_span])
+            og_span.duration_ns = 55
+            encoder.encode()
+        assert "misencoded duration: 0" in e.value.args[0]
+
+    # Ensure EncodingValidationError is raised when the encoded start does not match
+    with override_global_config({"_trace_writer_log_err_payload": True}):
+        with pytest.raises(EncodingValidationError) as e:
+            og_span = Span("name", "service", "resource", "type", start=10)
+            encoder.put([og_span])
+            og_span.start_ns = 100000001
+            encoder.encode()
+        assert "misencoded start: 10" in e.value.args[0]
+
+    # Ensure EncodingValidationError is raised when the encoded parent_id does not match
+    with override_global_config({"_trace_writer_log_err_payload": True}):
+        with pytest.raises(EncodingValidationError) as e:
+            og_span = Span("name", "service", "resource", "type", parent_id=1)
+            encoder.put([og_span])
+            og_span.parent_id = 2
+            encoder.encode()
+        assert "misencoded parent id: 1" in e.value.args[0]
+
+    # Ensure EncodingValidationError is raised when the encoded trace id does not match
+    with override_global_config({"_trace_writer_log_err_payload": True}):
+        with pytest.raises(EncodingValidationError) as e:
+            og_span = Span("name", "service", "resource", "type", trace_id=1)
+            encoder.put([og_span])
+            og_span.trace_id = 2
+            encoder.encode()
+        assert "misencoded trace id: 1" in e.value.args[0]
+
+    # Ensure EncodingValidationError is raised when the encoded span id does not match
+    with override_global_config({"_trace_writer_log_err_payload": True}):
+        with pytest.raises(EncodingValidationError) as e:
+            og_span = Span("name", "service", "resource", "type", span_id=1)
+            encoder.put([og_span])
+            og_span.span_id = 2
+            encoder.encode()
+        assert "misencoded span id: 1" in e.value.args[0]
+
+    # Ensure EncodingValidationError is raised when the encoded tags do not match the span tag's
+    with override_global_config({"_trace_writer_log_err_payload": True}):
+        with pytest.raises(EncodingValidationError) as e:
+            og_span = Span("name", "service", "resource", "type")
+            og_span._meta["hi"] = "tag"
+            encoder.put([og_span])
+            og_span._meta["hi"] = "new tag"
+            encoder.encode()
+        assert "misencoded tag: k=hi v=tag" in e.value.args[0]
+
+    # Ensure EncodingValidationError is raised when the encoded metrics do not match the metrics set on the span
+    with override_global_config({"_trace_writer_log_err_payload": True}):
+        with pytest.raises(EncodingValidationError) as e:
+            og_span = Span("name", "service", "resource", "type")
+            og_span._metrics["hi"] = 1
+            encoder.put([og_span])
+            og_span._metrics["hi"] = 2
+            encoder.encode()
+        assert "misencoded metric: k=hi v=1" in e.value.args[0]
+
+    # Ensure EncodingValidationError is raised when the encoded span type does not match the span type on the span
+    with override_global_config({"_trace_writer_log_err_payload": True}):
+        with pytest.raises(EncodingValidationError) as e:
+            og_span = Span("name", "service", "resource", "type")
+            encoder.put([og_span])
+            og_span.span_type = "new_span_type"
+            encoder.encode()
+        assert "misencoded span type: b'type'" in e.value.args[0]

tests/utils.py

@@ -1098,6 +1098,11 @@ def __eq__(self, other):
         return isinstance(other, str)
 
 
+class AnyStringWithText(str):
+    def __eq__(self, other):
+        return self in other
+
+
 class AnyInt(object):
     def __eq__(self, other):
         return isinstance(other, int)