fix(llmobs): send batches of span events [backport 3.3] (#12980)

Backport 76d06b6 from #12890 to 3.3.

Update spans to be sent as a batch of events. Each event contains one
span, allowing us to send spans up to 1MB (the event size limit). This
was a limit we assumed previously, but we were actually under a 1MB
limit _per payload/batch of spans_ limit since the payload we sent to
EVP intake was being treated as a single event.

For script
```
for i in range(200):
    with LLMObs.agent(
        "hi"
    ):
        LLMObs.annotate(
            input_data = "b" * 900_000,
            tags = {
                "test_run": "after_fix"
            }
        )
        pass
```


### Before 
**3 out of 200 sent**
<img width="920" alt="image"
src="https://github.com/user-attachments/assets/8a9d681c-87d3-4d86-a969-e4458d6413d6"
/>



### After

**200 out of 200 sent**
<img width="771" alt="image"
src="https://github.com/user-attachments/assets/bffce791-9aaa-4882-a4b2-bce370c895be"
/>


## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- 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)

Co-authored-by: lievan <[email protected]>

Author: github-actions[bot]

ddtrace/llmobs/_writer.py

@@ -212,7 +212,14 @@ def encode(self):
                 return None, 0
             events = self._buffer
             self._init_buffer()
-        data = {"_dd.stage": "raw", "_dd.tracer_version": ddtrace.__version__, "event_type": "span", "spans": events}
+        """
+        Send a batch of events, where each event contains a single span in the `spans` field. This allows us to
+        fully take advantage of EVP event/payload size limits.
+        """
+        data = [
+            {"_dd.stage": "raw", "_dd.tracer_version": ddtrace.__version__, "event_type": "span", "spans": [event]}
+            for event in events
+        ]
         try:
             enc_llm_events = safe_json(data)
             logger.debug("encode %d LLMObs span events to be sent", len(events))

releasenotes/notes/batch-spans-1120505146330ff6.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    LLM Observability: This fix resolves an issue where large spans traced within a short time interval were dropped despite being under the 1 MB limit.

tests/llmobs/test_llmobs.py

@@ -240,8 +240,8 @@ def test_utf_non_ascii_io(llmobs, llmobs_backend):
             llmobs.annotate(workflow_span, input_data="안녕, 지금 몇 시야?")
     events = llmobs_backend.wait_for_num_events(num=1)
     assert len(events) == 1
-    assert events[0]["spans"][0]["meta"]["input"]["messages"][0]["content"] == "안녕, 지금 몇 시야?"
-    assert events[0]["spans"][1]["meta"]["input"]["value"] == "안녕, 지금 몇 시야?"
+    assert events[0][0]["spans"][0]["meta"]["input"]["messages"][0]["content"] == "안녕, 지금 몇 시야?"
+    assert events[0][1]["spans"][0]["meta"]["input"]["value"] == "안녕, 지금 몇 시야?"
 
 
 def test_non_utf8_inputs_outputs(llmobs, llmobs_backend):
@@ -255,7 +255,7 @@ def test_non_utf8_inputs_outputs(llmobs, llmobs_backend):
     events = llmobs_backend.wait_for_num_events(num=1)
     assert len(events) == 1
     assert (
-        events[0]["spans"][0]["meta"]["input"]["messages"][0]["content"]
+        events[0][0]["spans"][0]["meta"]["input"]["messages"][0]["content"]
         == "The first Super Bowl (aka First AFL–NFL World Championship Game), was played in 1967."
     )
 
@@ -267,16 +267,16 @@ def test_structured_io_data(llmobs, llmobs_backend):
             llmobs.annotate(span, input_data={"data": "test1"}, output_data={"data": "test2"})
         events = llmobs_backend.wait_for_num_events(num=1)
         assert len(events) == 1
-        assert events[0]["spans"][0]["meta"]["input"]["value"] == '{"data": "test1"}'
-        assert events[0]["spans"][0]["meta"]["output"]["value"] == '{"data": "test2"}'
+        assert events[0][0]["spans"][0]["meta"]["input"]["value"] == '{"data": "test1"}'
+        assert events[0][0]["spans"][0]["meta"]["output"]["value"] == '{"data": "test2"}'
 
 
 def test_structured_prompt_data(llmobs, llmobs_backend):
     with llmobs.llm() as span:
         llmobs.annotate(span, prompt={"template": "test {{value}}"})
     events = llmobs_backend.wait_for_num_events(num=1)
     assert len(events) == 1
-    assert events[0]["spans"][0]["meta"]["input"] == {
+    assert events[0][0]["spans"][0]["meta"]["input"] == {
         "prompt": {
             "template": "test {{value}}",
             "_dd_context_variable_keys": ["context"],
@@ -295,5 +295,5 @@ class CustomObj:
             llmobs.annotate(span, input_data=CustomObj(), output_data=CustomObj())
         events = llmobs_backend.wait_for_num_events(num=1)
         assert len(events) == 1
-        assert expected_repr in events[0]["spans"][0]["meta"]["input"]["value"]
-        assert expected_repr in events[0]["spans"][0]["meta"]["output"]["value"]
+        assert expected_repr in events[0][0]["spans"][0]["meta"]["input"]["value"]
+        assert expected_repr in events[0][0]["spans"][0]["meta"]["output"]["value"]

tests/llmobs/test_llmobs_span_encoder.py

@@ -15,12 +15,14 @@ def test_encode_span(mock_writer_logs):
     encoder.put([span])
     encoded_llm_events, n_spans = encoder.encode()
 
-    expected_llm_events = {
-        "_dd.stage": "raw",
-        "_dd.tracer_version": ddtrace.__version__,
-        "event_type": "span",
-        "spans": [span],
-    }
+    expected_llm_events = [
+        {
+            "_dd.stage": "raw",
+            "_dd.tracer_version": ddtrace.__version__,
+            "event_type": "span",
+            "spans": [span],
+        }
+    ]
 
     assert n_spans == 1
     decoded_llm_events = json.loads(encoded_llm_events)
@@ -34,12 +36,20 @@ def test_encode_multiple_spans(mock_writer_logs):
     encoder.put(trace)
     encoded_llm_events, n_spans = encoder.encode()
 
-    expected_llm_events = {
-        "_dd.stage": "raw",
-        "_dd.tracer_version": ddtrace.__version__,
-        "event_type": "span",
-        "spans": trace,
-    }
+    expected_llm_events = [
+        {
+            "_dd.stage": "raw",
+            "_dd.tracer_version": ddtrace.__version__,
+            "event_type": "span",
+            "spans": [trace[0]],
+        },
+        {
+            "_dd.stage": "raw",
+            "_dd.tracer_version": ddtrace.__version__,
+            "event_type": "span",
+            "spans": [trace[1]],
+        },
+    ]
 
     assert n_spans == 2
     decoded_llm_events = json.loads(encoded_llm_events)
@@ -53,16 +63,18 @@ def test_encode_span_with_unserializable_fields():
     encoder.put([span])
     encoded_llm_events, n_spans = encoder.encode()
 
-    expected_llm_events = {
-        "_dd.stage": "raw",
-        "_dd.tracer_version": ddtrace.__version__,
-        "event_type": "span",
-        "spans": [mock.ANY],
-    }
+    expected_llm_events = [
+        {
+            "_dd.stage": "raw",
+            "_dd.tracer_version": ddtrace.__version__,
+            "event_type": "span",
+            "spans": [mock.ANY],
+        }
+    ]
 
     assert n_spans == 1
     decoded_llm_events = json.loads(encoded_llm_events)
     assert decoded_llm_events == expected_llm_events
-    decoded_llm_span = decoded_llm_events["spans"][0]
+    decoded_llm_span = decoded_llm_events[0]["spans"][0]
     assert decoded_llm_span["meta"]["metadata"]["unserializable"] is not None
     assert "<object object at 0x" in decoded_llm_span["meta"]["metadata"]["unserializable"]