Add /evp_proxy/v4 support for openai-java tests

Author: ygree

utils/_context/_scenarios/integration_frameworks.py

@@ -30,7 +30,8 @@ def __init__(self, name: str, doc: str) -> None:
             name,
             doc=doc,
             github_workflow="endtoend",
-            agent_image="ghcr.io/datadog/dd-apm-test-agent/ddapm-test-agent:v1.38.0",
+            # TODO update "local" -> "v1.42.0" once https://github.com/DataDog/dd-apm-test-agent/pull/280 merged and released
+            agent_image="ghcr.io/datadog/dd-apm-test-agent/ddapm-test-agent:local",
             scenario_groups=(groups.integration_frameworks,),
         )
 
@@ -166,4 +167,4 @@ def _set_dd_trace_integrations_enabled(self, library: str) -> None:
             # Reduce telemetry intervals for faster metric reporting in tests
             self.environment["DD_TELEMETRY_HEARTBEAT_INTERVAL"] = "1"
             self.environment["DD_TELEMETRY_METRICS_INTERVAL"] = "1"
-            self.environment["DD_TRACE_DEBUG"] = "true"
+            self.environment["DD_TRACE_DEBUG"] = "true"

utils/build/docker/java/openai_app/.gitignore

@@ -0,0 +1,2 @@
+.gradle/
+build/

utils/build/docker/java/openai_app/src/main/java/SingleFileServer.java

@@ -144,14 +144,14 @@ private static String doOpenAIChatCompletion(Context ctx, OpenAIClient openaiCli
 
     var builder = ChatCompletionCreateParams.builder();
     builder.model(payload.optString("model"));
-    
+
     // Parse messages array
     JSONArray messages = payload.getJSONArray("messages");
     for (int i = 0; i < messages.length(); i++) {
       JSONObject message = messages.getJSONObject(i);
       String role = message.getString("role");
       String content = message.getString("content");
-      
+
       if ("user".equals(role)) {
         builder.addUserMessage(content);
       } else if ("system".equals(role)) {
@@ -166,13 +166,13 @@ private static String doOpenAIChatCompletion(Context ctx, OpenAIClient openaiCli
       // maxCompletionTokens is preferable
       builder.maxTokens(parameters.optLong("max_tokens"));
     }
-    if (!Double.isNaN(parameters.optDouble("temperature"))) { 
+    if (!Double.isNaN(parameters.optDouble("temperature"))) {
       builder.temperature(parameters.optDouble("temperature"));
     }
     if (tools != null) {
       for (int i = 0; i < tools.length(); i++) {
         JSONObject tool = tools.getJSONObject(i);
-        
+
          JSONObject functionObj = tool.getJSONObject("function");
          Map<String, Object> functionMap = functionObj.toMap();
 
@@ -212,7 +212,7 @@ private static String doOpenAIEmbedding (Context ctx, OpenAIClient openaiClient)
     builder.input(input);
 
     openaiClient.embeddings().create(builder.build());
-    
+
     return toJson(new HashMap<String, String>());
   }
 
@@ -242,7 +242,7 @@ private static String doOpenAIResponsesCreate (Context ctx, OpenAIClient openaiC
       builder.maxOutputTokens((long) parameters.getDouble("max_output_tokens"));
     }
 
-    if (!Double.isNaN(parameters.optDouble("temperature"))) { 
+    if (!Double.isNaN(parameters.optDouble("temperature"))) {
       builder.temperature(parameters.getDouble("temperature"));
     }
 
@@ -257,7 +257,7 @@ private static String doOpenAIResponsesCreate (Context ctx, OpenAIClient openaiC
 
     if (tools != null) {
       List<Object> toolsList = (List<Object>) deepConvertJsonToJava(tools);
-      
+
       builder.tools(JsonValue.from(toolsList));
       builder.toolChoice(JsonValue.from("auto"));
     }
@@ -267,7 +267,7 @@ private static String doOpenAIResponsesCreate (Context ctx, OpenAIClient openaiC
         streamResponse.stream().forEach(chunk -> {
             // consume the stream
         });
-      } 
+      }
     } else {
       openaiClient.responses().create(builder.build());
     }
@@ -303,7 +303,7 @@ private static String doAnthropicCreate (Context ctx, AnthropicClient anthropicC
       List<Object> systemList = (List<Object>) deepConvertJsonToJava(system);
       builder.system(com.anthropic.core.JsonValue.from(systemList));
     }
-    
+
     builder.model(model);
 
     if (messages != null) {
@@ -402,8 +402,8 @@ private static String doCreateTrace (JSONObject traceStructure) {
   }
 
   private static void doTraceChildren(JSONArray children) {
-    if (children == null) { 
-      return; 
+    if (children == null) {
+      return;
     }
 
     for (int i = 0; i < children.length(); i++) {

utils/docker_fixtures/_test_agent.py

@@ -2,6 +2,7 @@
 from collections.abc import Generator
 import contextlib
 import datetime
+import gzip
 import hashlib
 from http import HTTPStatus
 import json
@@ -11,6 +12,7 @@
 from typing import TypedDict, Any, cast
 import urllib.parse
 
+import msgpack
 import pytest
 import requests
 from retry import retry
@@ -386,24 +388,47 @@ def info(self):
         self._write_log("info", resp_json)
         return resp_json
 
+    def _decode_llmobs_body(self, body_b64: str) -> list[Any]:
+        """Decode base64 body; handle gzip (Java), then JSON or MessagePack (Java) encoding.
+
+        Returns a list of events (each event is a dict with 'spans'). Java can send multiple
+        concatenated msgpack objects in one request; we use Unpacker to decode all of them
+        (same as llm-obs test/conftest.py).
+        """
+        decoded = base64.b64decode(body_b64)
+        if decoded[:2] == b"\x1f\x8b":
+            decoded = gzip.decompress(decoded)
+        # JSON (Python/Node tracer): starts with { or [
+        if decoded.lstrip().startswith((b"{", b"[")):
+            parsed = json.loads(decoded)
+            return [parsed] if isinstance(parsed, dict) else parsed
+        # MessagePack (Java tracer): binary format; may be multiple concatenated objects
+        unpacker = msgpack.Unpacker(unicode_errors="replace", strict_map_key=False)
+        unpacker.feed(decoded)
+        return list(unpacker)
+
     def llmobs_requests(self) -> list[Any]:
-        reqs = [r for r in self.requests() if r["url"].endswith("/evp_proxy/v2/api/v2/llmobs")]
+        reqs = [
+            r
+            for r in self.requests()
+            if r["url"].endswith("/evp_proxy/v2/api/v2/llmobs") or r["url"].endswith("/evp_proxy/v4/api/v2/llmobs")
+        ]
 
         events = []
         for r in reqs:
-            decoded_body = base64.b64decode(r["body"])
-            events.append(json.loads(decoded_body))
+            events.append(self._decode_llmobs_body(r["body"]))
         return events
 
-    def llmobs_evaluations_requests(self):
+    def llmobs_evaluations_requests(self) -> list[Any]:
         reqs = [
             r
             for r in self.requests()
             if r["url"].endswith("/evp_proxy/v2/api/intake/llm-obs/v1/eval-metric")
             or r["url"].endswith("/evp_proxy/v2/api/intake/llm-obs/v2/eval-metric")
         ]
-
-        return [json.loads(base64.b64decode(r["body"])) for r in reqs]
+        # One decoded body per request (evaluations are typically single JSON per request)
+        decoded_per_request = [self._decode_llmobs_body(r["body"]) for r in reqs]
+        return [events[0] for events in decoded_per_request]
 
     @contextlib.contextmanager
     def snapshot_context(self, token: str, ignores: list[str] | None = None):