fix(core): handle failed encoding (#1946)

Looks like there was a typo/oversight with the error handling for encoding in which we'd continue to try to send the trace after encoding failed.

* add regression test

* continue only when encoding is successful

Author: Kyle-Verhoog

ddtrace/internal/writer.py

@@ -256,31 +256,31 @@ def write(self, spans):
         try:
             encoded = self._encoder.encode_trace(spans)
         except Exception:
-            log.warning("failed to encode trace with encoder %r", self._encoder, exc_info=True)
-
-        try:
-            self._buffer.put(encoded)
-        except BufferItemTooLarge:
-            log.warning(
-                "trace (%db) larger than payload limit (%db), dropping",
-                len(encoded),
-                self._max_payload_size,
-            )
-            self._metrics_dist("buffer.dropped.traces", 1, tags=["reason:t_too_big"])
-            self._metrics_dist("buffer.dropped.bytes", len(encoded), tags=["reason:t_too_big"])
-        except BufferFull:
-            log.warning(
-                "trace buffer (%s traces %db/%db) cannot fit trace of size %db, dropping",
-                len(self._buffer),
-                self._buffer.size,
-                self._buffer.max_size,
-                len(encoded),
-            )
-            self._metrics_dist("buffer.dropped.traces", 1, tags=["reason:full"])
-            self._metrics_dist("buffer.dropped.bytes", len(encoded), tags=["reason:full"])
+            log.error("failed to encode trace with encoder %r", self._encoder, exc_info=True)
         else:
-            self._metrics_dist("buffer.accepted.traces", 1)
-            self._metrics_dist("buffer.accepted.spans", len(spans))
+            try:
+                self._buffer.put(encoded)
+            except BufferItemTooLarge:
+                log.warning(
+                    "trace (%db) larger than payload limit (%db), dropping",
+                    len(encoded),
+                    self._max_payload_size,
+                )
+                self._metrics_dist("buffer.dropped.traces", 1, tags=["reason:t_too_big"])
+                self._metrics_dist("buffer.dropped.bytes", len(encoded), tags=["reason:t_too_big"])
+            except BufferFull:
+                log.warning(
+                    "trace buffer (%s traces %db/%db) cannot fit trace of size %db, dropping",
+                    len(self._buffer),
+                    self._buffer.size,
+                    self._buffer.max_size,
+                    len(encoded),
+                )
+                self._metrics_dist("buffer.dropped.traces", 1, tags=["reason:full"])
+                self._metrics_dist("buffer.dropped.bytes", len(encoded), tags=["reason:full"])
+            else:
+                self._metrics_dist("buffer.accepted.traces", 1)
+                self._metrics_dist("buffer.accepted.spans", len(spans))
 
     def flush_queue(self):
         enc_traces = self._buffer.get()

tests/integration/test_integration.py

@@ -341,6 +341,24 @@ def join_encoded(self, traces):
     log.error.assert_has_calls(calls)
 
 
+def test_bad_encoder():
+    t = Tracer()
+
+    class BadEncoder:
+        def encode_trace(self, spans):
+            raise Exception()
+
+        def join_encoded(self, traces):
+            pass
+
+    t.writer._encoder = BadEncoder()
+    with mock.patch("ddtrace.internal.writer.log") as log:
+        t.trace("asdf").finish()
+        t.shutdown()
+    calls = [mock.call("failed to encode trace with encoder %r", t.writer._encoder, exc_info=True)]
+    log.error.assert_has_calls(calls)
+
+
 @pytest.mark.skipif(AGENT_VERSION == "testagent", reason="Test agent doesn't support v0.3")
 def test_downgrade():
     t = Tracer()