From 44203624240da135fab59d6851c7c71a4f3840ca Mon Sep 17 00:00:00 2001 From: Alexey Kuznetsov Date: Fri, 31 Jul 2026 14:30:52 -0400 Subject: [PATCH 1/3] Port TraceMapperV1 payload test from Spock to JUnit 5 Co-Authored-By: Claude Opus 5 (1M context) --- .../ddagent/TraceMapperV1PayloadTest.groovy | 1208 ----------------- .../ddagent/TraceMapperV1PayloadTest.java | 946 +++++++++++++ .../writer/ddagent/V1PayloadReader.java | 6 +- 3 files changed, 949 insertions(+), 1211 deletions(-) delete mode 100644 dd-trace-core/src/test/groovy/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.groovy create mode 100644 dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.java diff --git a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.groovy b/dd-trace-core/src/test/groovy/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.groovy deleted file mode 100644 index 5ad21853679..00000000000 --- a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.groovy +++ /dev/null @@ -1,1208 +0,0 @@ -package datadog.trace.common.writer.ddagent - -import static datadog.trace.common.writer.TraceGenerator.generateRandomTraces -import static datadog.trace.common.writer.ddagent.V1PayloadReader.readAttributes -import static datadog.trace.common.writer.ddagent.V1PayloadReader.readFirstSpan -import static datadog.trace.common.writer.ddagent.V1PayloadReader.readStreamingString -import static datadog.trace.common.writer.ddagent.V1PayloadReader.skipChunkField -import static datadog.trace.common.writer.ddagent.V1PayloadReader.skipPayloadField -import static datadog.trace.common.writer.ddagent.V1PayloadReader.skipSpanField -import static datadog.trace.common.writer.ddagent.V1PayloadReader.unpackUnsignedLong -import static org.junit.jupiter.api.Assertions.assertArrayEquals -import static org.junit.jupiter.api.Assertions.assertEquals -import static org.junit.jupiter.api.Assertions.assertNotNull -import static org.junit.jupiter.api.Assertions.assertTrue - -import datadog.communication.serialization.ByteBufferConsumer -import datadog.communication.serialization.FlushingBuffer -import datadog.communication.serialization.msgpack.MsgPackWriter -import datadog.trace.api.DDTags -import datadog.trace.api.DDTraceId -import datadog.trace.api.DDSpanId -import datadog.trace.api.ProcessTags -import datadog.trace.api.sampling.PrioritySampling -import datadog.trace.api.sampling.SamplingMechanism -import datadog.trace.bootstrap.instrumentation.api.InstrumentationTags -import datadog.trace.bootstrap.instrumentation.api.SpanAttributes -import datadog.trace.bootstrap.instrumentation.api.SpanLink -import datadog.trace.bootstrap.instrumentation.api.Tags -import datadog.trace.common.writer.Payload -import datadog.trace.common.writer.TraceGenerator -import datadog.trace.core.MetadataConsumer -import datadog.trace.test.util.DDSpecification -import java.nio.ByteBuffer -import java.nio.channels.WritableByteChannel -import org.junit.jupiter.api.Assertions -import org.msgpack.core.MessageFormat -import org.msgpack.core.MessagePack -import org.msgpack.core.MessageUnpacker - -class TraceMapperV1PayloadTest extends DDSpecification { - - def "test traces written correctly"() { - setup: - List> traces = generateRandomTraces(traceCount, lowCardinality) - TraceMapperV1 traceMapper = new TraceMapperV1() - PayloadVerifier verifier = new PayloadVerifier(traces, traceMapper) - MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(bufferSize, verifier)) - - when: - boolean tracesFitInBuffer = true - for (List trace : traces) { - if (!packer.format(trace, traceMapper)) { - verifier.skipLargeTrace() - tracesFitInBuffer = false - traceMapper.reset() - } - } - packer.flush() - - then: - if (tracesFitInBuffer) { - verifier.verifyTracesConsumed() - } - - where: - bufferSize | traceCount | lowCardinality - 20 << 10 | 0 | true - 20 << 10 | 1 | true - 30 << 10 | 2 | true - 20 << 10 | 0 | false - 20 << 10 | 1 | false - 30 << 10 | 2 | false - 100 << 10 | 10 | true - 100 << 10 | 100 | false - } - - def "test endpoint returns v1.0"() { - expect: - new TraceMapperV1().endpoint() == "v1.0" - } - - def "test span kind value conversion"() { - expect: - TraceMapperV1.getSpanKindValue(null) == TraceMapperV1.SPAN_KIND_UNSPECIFIED - TraceMapperV1.getSpanKindValue(Tags.SPAN_KIND_INTERNAL) == TraceMapperV1.SPAN_KIND_INTERNAL - TraceMapperV1.getSpanKindValue(Tags.SPAN_KIND_SERVER) == TraceMapperV1.SPAN_KIND_SERVER - TraceMapperV1.getSpanKindValue(Tags.SPAN_KIND_CLIENT) == TraceMapperV1.SPAN_KIND_CLIENT - TraceMapperV1.getSpanKindValue(Tags.SPAN_KIND_PRODUCER) == TraceMapperV1.SPAN_KIND_PRODUCER - TraceMapperV1.getSpanKindValue(Tags.SPAN_KIND_CONSUMER) == TraceMapperV1.SPAN_KIND_CONSUMER - TraceMapperV1.getSpanKindValue("unknown") == TraceMapperV1.SPAN_KIND_INTERNAL - } - - def "test payload contains expected header and chunk fields"() { - setup: - Map tags = [ - (Tags.ENV): "prod", - (Tags.VERSION): "1.2.3", - (Tags.COMPONENT): "http-client", - (Tags.SPAN_KIND): Tags.SPAN_KIND_CLIENT, - "attr.string": "value", - "attr.bool" : true, - "attr.number": 12.5d - ] - def span = new TraceGenerator.PojoSpan( - "service-a", - "operation-a", - "resource-a", - DDTraceId.ONE, - 123L, - 0L, - 1000L, - 2000L, - 1, - ["_dd.p.dm": "-3"], - tags, - "web", - false, - PrioritySampling.SAMPLER_KEEP, - 200, - "rum") - - TraceMapperV1 mapper = new TraceMapperV1() - byte[] encoded = serializeMappedPayload(mapper, [[span]]) - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded) - List stringTable = new ArrayList<>() - stringTable.add("") - - when: - int payloadFieldCount = unpacker.unpackMapHeader() - Set payloadFieldsSeen = new HashSet<>() - int chunkCount = -1 - Map payloadAttributes = null - - for (int i = 0; i < payloadFieldCount; i++) { - int fieldId = unpacker.unpackInt() - payloadFieldsSeen.add(fieldId) - switch (fieldId) { - case 2: - case 3: - case 4: - case 5: - case 6: - case 7: - case 8: - case 9: - readStreamingString(unpacker, stringTable) - break - case 10: - payloadAttributes = readAttributes(unpacker, stringTable) - break - case 11: - chunkCount = unpacker.unpackArrayHeader() - assertEquals(1, chunkCount) - verifyChunk(unpacker, [span], stringTable) - break - default: - Assertions.fail("Unexpected payload field id: " + fieldId) - } - } - - then: - assertEquals(10, payloadFieldCount) - assertEquals((2..11).toSet(), payloadFieldsSeen) - assertEquals(1, chunkCount) - assertNotNull(payloadAttributes) - if (ProcessTags.tagsForSerialization == null) { - assertEquals(0, payloadAttributes.size()) - } else { - assertEquals(1, payloadAttributes.size()) - assertEquals(ProcessTags.tagsForSerialization.toString(), payloadAttributes.get(DDTags.PROCESS_TAGS)) - } - } - - def "test sampling mechanism normalization from _dd.p.dm"() { - setup: - def span = new TraceGenerator.PojoSpan( - "service-a", - "operation-a", - "resource-a", - DDTraceId.ONE, - 321L, - 0L, - 1000L, - 2000L, - 0, - decisionMakerTag == null ? [:] : ["_dd.p.dm": decisionMakerTag], - [:], - "custom", - false, - PrioritySampling.SAMPLER_KEEP, - 200, - null) - - TraceMapperV1 mapper = new TraceMapperV1() - byte[] encoded = serializeMappedPayload(mapper, [[span]]) - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded) - List stringTable = new ArrayList<>() - stringTable.add("") - - when: - unpacker.unpackMapHeader() - int samplingMechanism = -1 - - for (int i = 0; i < 10; i++) { - int payloadFieldId = unpacker.unpackInt() - if (payloadFieldId == 11) { - int chunkCount = unpacker.unpackArrayHeader() - assertEquals(1, chunkCount) - int chunkFieldCount = unpacker.unpackMapHeader() - for (int j = 0; j < chunkFieldCount; j++) { - int chunkFieldId = unpacker.unpackInt() - if (chunkFieldId == 7) { - samplingMechanism = unpacker.unpackInt() - } else { - skipChunkField(unpacker, chunkFieldId, stringTable) - } - } - } else { - skipPayloadField(unpacker, payloadFieldId, stringTable) - } - } - - then: - assertEquals(expectedSamplingMechanism, samplingMechanism) - - where: - decisionMakerTag | expectedSamplingMechanism - null | SamplingMechanism.DEFAULT - "-3" | 3 - "934086a686-7" | 7 - "invalid" | SamplingMechanism.DEFAULT - } - - def "test span ids are encoded as unsigned values in v1 payloads"() { - setup: - long spanId = Long.MIN_VALUE + 123L - long parentId = Long.MIN_VALUE + 456L - def span = new TraceGenerator.PojoSpan( - "service-a", - "operation-a", - "resource-a", - DDTraceId.ONE, - spanId, - parentId, - 1000L, - 2000L, - 0, - [:], - [:], - "web", - false, - PrioritySampling.SAMPLER_KEEP, - 200, - null) - - TraceMapperV1 mapper = new TraceMapperV1() - byte[] encoded = serializeMappedPayload(mapper, [[span]]) - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded) - List stringTable = new ArrayList<>() - stringTable.add("") - - when: - unpacker.unpackMapHeader() - Long actualSpanId = null - Long actualParentId = null - - for (int i = 0; i < 10; i++) { - int payloadFieldId = unpacker.unpackInt() - if (payloadFieldId == 11) { - int chunkCount = unpacker.unpackArrayHeader() - assertEquals(1, chunkCount) - int chunkFieldCount = unpacker.unpackMapHeader() - for (int j = 0; j < chunkFieldCount; j++) { - int chunkFieldId = unpacker.unpackInt() - if (chunkFieldId == 4) { - int spanCount = unpacker.unpackArrayHeader() - assertEquals(1, spanCount) - int spanFieldCount = unpacker.unpackMapHeader() - for (int k = 0; k < spanFieldCount; k++) { - int spanFieldId = unpacker.unpackInt() - switch (spanFieldId) { - case 4: - assertEquals(MessageFormat.UINT64, unpacker.nextFormat) - actualSpanId = DDSpanId.from("${unpacker.unpackBigInteger()}") - break - case 5: - assertEquals(MessageFormat.UINT64, unpacker.nextFormat) - actualParentId = DDSpanId.from("${unpacker.unpackBigInteger()}") - break - default: - skipSpanField(unpacker, spanFieldId, stringTable) - } - } - } else { - skipChunkField(unpacker, chunkFieldId, stringTable) - } - } - } else { - skipPayloadField(unpacker, payloadFieldId, stringTable) - } - } - - then: - assertEquals(spanId, actualSpanId) - assertEquals(parentId, actualParentId) - } - - def "test span links are encoded from structured span links"() { - setup: - List spanLinks = [ - new SpanLink( - DDTraceId.fromHex("11223344556677889900aabbccddeeff"), - DDSpanId.fromHex("000000000000002a"), - (byte) 1, - "dd=s:1", - SpanAttributes.fromMap(["link.kind": "follows_from", "context_headers": "tracecontext"])), - new SpanLink( - DDTraceId.fromHex("00000000000000000000000000000001"), - DDSpanId.fromHex("0000000000000002"), - (byte) 0, - "", - SpanAttributes.EMPTY) - ] - def span = new TraceGenerator.PojoSpan( - "service-a", - "operation-a", - "resource-a", - DDTraceId.ONE, - 123L, - 0L, - 1000L, - 2000L, - 0, - [:], - [:], - "web", - false, - PrioritySampling.SAMPLER_KEEP, - 200, - null, - spanLinks) - - TraceMapperV1 mapper = new TraceMapperV1() - byte[] encoded = serializeMappedPayload(mapper, [[span]]) - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded) - List stringTable = new ArrayList<>() - stringTable.add("") - - when: - List> links = readFirstSpan(unpacker, stringTable).links - - then: - assertEquals(2, links.size()) - assertArrayEquals(traceIdBytes(DDTraceId.fromHex("11223344556677889900aabbccddeeff")), links[0].traceId as byte[]) - assertEquals(DDSpanId.fromHex("000000000000002a"), links[0].spanId) - assertEquals("dd=s:1", links[0].traceState) - assertEquals(1L, links[0].traceFlags) - assertEquals(["link.kind": "follows_from", "context_headers": "tracecontext"], links[0].attributes) - - assertArrayEquals(traceIdBytes(DDTraceId.fromHex("00000000000000000000000000000001")), links[1].traceId as byte[]) - assertEquals(DDSpanId.fromHex("0000000000000002"), links[1].spanId) - assertEquals("", links[1].traceState) - assertEquals(0L, links[1].traceFlags) - assertEquals([:], links[1].attributes) - } - - def "test first span tags are processed once"() { - setup: - def firstSpan = new CountingPojoSpan( - "service-a", - "operation-a", - "resource-a", - DDTraceId.ONE, - 123L, - 0L, - 1000L, - 2000L, - 0, - [:], - [(Tags.HTTP_URL): "http://localhost:7777/"], - "web", - false, - PrioritySampling.SAMPLER_KEEP, - 200, - null) - - def secondSpan = new CountingPojoSpan( - "service-a", - "operation-b", - "resource-b", - DDTraceId.ONE, - 456L, - 123L, - 1000L, - 2000L, - 0, - [:], - [(Tags.HTTP_URL): "http://localhost:7777/"], - "web", - false, - PrioritySampling.SAMPLER_KEEP, - 200, - null) - - TraceMapperV1 mapper = new TraceMapperV1() - - when: - serializeMappedPayload(mapper, [[firstSpan, secondSpan]]) - - then: - assertEquals(1, firstSpan.processTagsAndBaggageCount) - assertEquals(1, secondSpan.processTagsAndBaggageCount) - } - - def "test missing span links encode empty links"() { - setup: - def span = new TraceGenerator.PojoSpan( - "service-a", - "operation-a", - "resource-a", - DDTraceId.ONE, - 123L, - 0L, - 1000L, - 2000L, - 0, - [:], - [:], - "web", - false, - PrioritySampling.SAMPLER_KEEP, - 200, - null) - - TraceMapperV1 mapper = new TraceMapperV1() - byte[] encoded = serializeMappedPayload(mapper, [[span]]) - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded) - List stringTable = new ArrayList<>() - stringTable.add("") - - when: - List> links = readFirstSpan(unpacker, stringTable).links - - then: - assertTrue(links.isEmpty()) - } - - def "test span events are encoded from events tag"() { - setup: - List> eventPayload = [ - [ - time_unix_nano: 1234567890L, - name : "event.one", - attributes : [ - str : "v", - int : 42L, - double: 12.5d, - bool : true, - arr : ["x", 7L, 2.5d, false] - ] - ], - [ - time_unix_nano: 1234567891L, - name : "event.two" - ] - ] - def span = new TraceGenerator.PojoSpan( - "service-a", - "operation-a", - "resource-a", - DDTraceId.ONE, - 123L, - 0L, - 1000L, - 2000L, - 0, - [:], - ["events": eventPayload], - "web", - false, - PrioritySampling.SAMPLER_KEEP, - 200, - null) - - TraceMapperV1 mapper = new TraceMapperV1() - byte[] encoded = serializeMappedPayload(mapper, [[span]]) - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded) - List stringTable = new ArrayList<>() - stringTable.add("") - - when: - List> events = readFirstSpan(unpacker, stringTable).events - - then: - assertEquals(2, events.size()) - assertEquals(1234567890L, events[0].timeUnixNano) - assertEquals("event.one", events[0].name) - assertEquals("v", events[0].attributes["str"]) - assertEquals(42L, events[0].attributes["int"]) - assertEquals(12.5d, (events[0].attributes["double"] as Number).doubleValue(), 0.000001d) - assertEquals(true, events[0].attributes["bool"]) - assertEquals(["x", 7L, 2.5d, false], events[0].attributes["arr"]) - - assertEquals(1234567891L, events[1].timeUnixNano) - assertEquals("event.two", events[1].name) - assertEquals([:], events[1].attributes) - } - - def "test malformed span events fall back to empty events"() { - setup: - def span = new TraceGenerator.PojoSpan( - "service-a", - "operation-a", - "resource-a", - DDTraceId.ONE, - 123L, - 0L, - 1000L, - 2000L, - 0, - [:], - ["events": [foo: "bar"]], - "web", - false, - PrioritySampling.SAMPLER_KEEP, - 200, - null) - - TraceMapperV1 mapper = new TraceMapperV1() - byte[] encoded = serializeMappedPayload(mapper, [[span]]) - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded) - List stringTable = new ArrayList<>() - stringTable.add("") - - when: - List> events = readFirstSpan(unpacker, stringTable).events - - then: - assertTrue(events.isEmpty()) - } - - def "test meta struct is encoded as bytes attribute"() { - setup: - def span = new TraceGenerator.PojoSpan( - "service-a", - "operation-a", - "resource-a", - DDTraceId.ONE, - 123L, - 0L, - 1000L, - 2000L, - 0, - [:], - [:], - "web", - false, - PrioritySampling.SAMPLER_KEEP, - 200, - null) - span.setMetaStruct("meta_key", [foo: "bar", answer: 42L]) - - TraceMapperV1 mapper = new TraceMapperV1() - byte[] encoded = serializeMappedPayload(mapper, [[span]]) - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded) - List stringTable = new ArrayList<>() - stringTable.add("") - - when: - Map attributes = readFirstSpan(unpacker, stringTable).attributes - byte[] metaStructBytes = attributes["meta_key"] as byte[] - MessageUnpacker metaStructUnpacker = MessagePack.newDefaultUnpacker(metaStructBytes) - int metaStructFieldCount = metaStructUnpacker.unpackMapHeader() - Map decodedMetaStruct = [:] - for (int i = 0; i < metaStructFieldCount; i++) { - String key = metaStructUnpacker.unpackString() - switch (metaStructUnpacker.getNextFormat().getValueType()) { - case org.msgpack.value.ValueType.INTEGER: - decodedMetaStruct[key] = metaStructUnpacker.unpackLong() - break - case org.msgpack.value.ValueType.STRING: - decodedMetaStruct[key] = metaStructUnpacker.unpackString() - break - default: - Assertions.fail("Unexpected meta_struct value type for key " + key) - } - } - - then: - assertNotNull(metaStructBytes) - assertEquals("bar", decodedMetaStruct["foo"]) - assertEquals(42L, decodedMetaStruct["answer"]) - } - - def "test map-valued span tags are flattened in v1 attributes"() { - setup: - def span = new TraceGenerator.PojoSpan( - "service-a", - "operation-a", - "resource-a", - DDTraceId.ONE, - 123L, - 0L, - 1000L, - 2000L, - 0, - [:], - [ - "usr": [ - "id" : "123", - "name" : "alice", - "authenticated": true, - "profile" : [ - "age": 30L - ] - ], - "appsec.events.users.login.success": [ - "metadata0": [ - "event" : "login", - "attempts": 1L - ], - "metadata1": [ - "blocked": false - ] - ] - ], - "web", - false, - PrioritySampling.SAMPLER_KEEP, - 0, - null) - - TraceMapperV1 mapper = new TraceMapperV1() - byte[] encoded = serializeMappedPayload(mapper, [[span]]) - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded) - List stringTable = new ArrayList<>() - stringTable.add("") - - when: - Map attributes = readFirstSpan(unpacker, stringTable).attributes - - then: - assertTrue(attributes.containsKey("usr.id")) - assertTrue(attributes.containsKey("usr.name")) - assertTrue(attributes.containsKey("usr.authenticated")) - assertTrue(attributes.containsKey("usr.profile.age")) - assertTrue(attributes.containsKey("appsec.events.users.login.success.metadata0.event")) - assertTrue(attributes.containsKey("appsec.events.users.login.success.metadata0.attempts")) - assertTrue(attributes.containsKey("appsec.events.users.login.success.metadata1.blocked")) - - assertEquals("123", attributes.get("usr.id")) - assertEquals("alice", attributes.get("usr.name")) - assertEquals(true, attributes.get("usr.authenticated")) - assertEquals(30d, (attributes.get("usr.profile.age") as Number).doubleValue(), 0.000001d) - assertEquals("login", attributes.get("appsec.events.users.login.success.metadata0.event")) - assertEquals(1d, (attributes.get("appsec.events.users.login.success.metadata0.attempts") as Number).doubleValue(), 0.000001d) - assertEquals(false, attributes.get("appsec.events.users.login.success.metadata1.blocked")) - - assertTrue(!attributes.containsKey("usr")) - assertTrue(!attributes.containsKey("appsec.events.users.login.success")) - } - - def "test primitive span tags are encoded in v1 attributes"() { - setup: - def span = new TraceGenerator.PojoSpan( - "service-a", - "operation-a", - "resource-a", - DDTraceId.ONE, - 123L, - 0L, - 1000L, - 2000L, - 0, - [:], - [ - "tag.bool" : true, - "tag.int" : 7, - "tag.long" : 9L, - "tag.float" : 3.5f, - "tag.double": 4.25d - ], - "web", - false, - PrioritySampling.SAMPLER_KEEP, - 0, - null) - - TraceMapperV1 mapper = new TraceMapperV1() - byte[] encoded = serializeMappedPayload(mapper, [[span]]) - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded) - List stringTable = new ArrayList<>() - stringTable.add("") - - when: - Map attributes = readFirstSpan(unpacker, stringTable).attributes - - then: - assertEquals(true, attributes.get("tag.bool")) - assertEquals(7d, (attributes.get("tag.int") as Number).doubleValue(), 0.000001d) - assertEquals(9d, (attributes.get("tag.long") as Number).doubleValue(), 0.000001d) - assertEquals(3.5d, (attributes.get("tag.float") as Number).doubleValue(), 0.000001d) - assertEquals(4.25d, (attributes.get("tag.double") as Number).doubleValue(), 0.000001d) - } - - def "test thread metadata is encoded in v1 attributes"() { - setup: - def span = new TraceGenerator.PojoSpan( - "service-a", - "operation-a", - "resource-a", - DDTraceId.ONE, - 123L, - 0L, - 1000L, - 2000L, - 0, - [:], - [:], - "web", - false, - PrioritySampling.SAMPLER_KEEP, - 0, - null) - - TraceMapperV1 mapper = new TraceMapperV1() - byte[] encoded = serializeMappedPayload(mapper, [[span]]) - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded) - List stringTable = new ArrayList<>() - stringTable.add("") - - when: - Map attributes = readFirstSpan(unpacker, stringTable).attributes - - then: - assertAttributeValueEquals(span.getTag(DDTags.THREAD_ID), attributes.get(DDTags.THREAD_ID), DDTags.THREAD_ID) - assertEquals(span.getTag(DDTags.THREAD_NAME).toString(), attributes.get(DDTags.THREAD_NAME)) - } - - private static final class PayloadVerifier implements ByteBufferConsumer, WritableByteChannel { - - private final List> expectedTraces - private final TraceMapperV1 mapper - private ByteBuffer captured = ByteBuffer.allocate(200 << 10) - private int position = 0 - - private PayloadVerifier(List> expectedTraces, TraceMapperV1 mapper) { - this.expectedTraces = expectedTraces - this.mapper = mapper - } - - void skipLargeTrace() { - ++position - } - - @Override - void accept(int messageCount, ByteBuffer buffer) { - if (expectedTraces.isEmpty() && messageCount == 0) { - return - } - try { - Payload payload = mapper.newPayload().withBody(messageCount, buffer) - payload.writeTo(this) - captured.flip() - - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(captured) - if (messageCount == 0) { - assertEquals(0, unpacker.unpackMapHeader()) - return - } - - List stringTable = new ArrayList<>() - stringTable.add("") - - int payloadFieldCount = unpacker.unpackMapHeader() - assertEquals(10, payloadFieldCount) - - boolean seenChunks = false - for (int i = 0; i < payloadFieldCount; i++) { - int fieldId = unpacker.unpackInt() - if (fieldId == 11) { - int traceCount = unpacker.unpackArrayHeader() - assertEquals(messageCount, traceCount) - seenChunks = true - for (int traceIndex = 0; traceIndex < traceCount; traceIndex++) { - List expectedTrace = expectedTraces.get(position++) - verifyChunk(unpacker, expectedTrace, stringTable) - } - } else { - skipPayloadField(unpacker, fieldId, stringTable) - } - } - - assertTrue(seenChunks) - } catch (IOException e) { - Assertions.fail(e.getMessage()) - } finally { - mapper.reset() - captured.position(0) - captured.limit(captured.capacity()) - } - } - - @Override - int write(ByteBuffer src) { - if (captured.remaining() < src.remaining()) { - ByteBuffer newBuffer = ByteBuffer.allocate(captured.capacity() + src.remaining()) - captured.flip() - newBuffer.put(captured) - captured = newBuffer - return write(src) - } - captured.put(src) - return src.position() - } - - void verifyTracesConsumed() { - assertEquals(expectedTraces.size(), position) - } - - @Override - boolean isOpen() { - return true - } - - @Override - void close() { - } - } - - private static void verifyChunk( - MessageUnpacker unpacker, - List expectedTrace, - List stringTable) { - int chunkFieldCount = unpacker.unpackMapHeader() - assertEquals(6, chunkFieldCount) - - Integer priority = null - String origin = null - Map chunkAttributes = null - byte[] traceId = null - Integer samplingMechanism = null - List decodedSpans = null - - for (int i = 0; i < chunkFieldCount; i++) { - int fieldId = unpacker.unpackInt() - switch (fieldId) { - case 1: - priority = unpacker.unpackInt() - break - case 2: - origin = readStreamingString(unpacker, stringTable) - break - case 3: - chunkAttributes = readAttributes(unpacker, stringTable) - break - case 4: - decodedSpans = verifySpans(unpacker, expectedTrace, stringTable) - break - case 6: - int traceIdLen = unpacker.unpackBinaryHeader() - traceId = new byte[traceIdLen] - unpacker.readPayload(traceId) - break - case 7: - samplingMechanism = unpacker.unpackInt() - break - default: - Assertions.fail("Unexpected chunk field id: " + fieldId) - } - } - - assertNotNull(priority) - assertNotNull(origin) - assertNotNull(chunkAttributes) - assertNotNull(decodedSpans) - assertNotNull(traceId) - assertNotNull(samplingMechanism) - - TraceGenerator.PojoSpan firstSpan = expectedTrace.get(0) - assertEquals(firstSpan.samplingPriority(), priority) - assertEqualsWithNullAsEmpty(firstSpan.getOrigin(), origin) - assertEquals(1, chunkAttributes.size()) - assertEqualsWithNullAsEmpty(firstSpan.getLocalRootSpan().getServiceName(), chunkAttributes.get("service")) - assertArrayEquals(traceIdBytes(firstSpan.getTraceId()), traceId) - assertEquals(expectedSamplingMechanism(firstSpan.getBaggage()), samplingMechanism) - } - - private static byte[] traceIdBytes(DDTraceId traceId) { - ByteBuffer.allocate(16) - .putLong(traceId.toHighOrderLong()) - .putLong(traceId.toLong()) - .array() - } - - private static List verifySpans( - MessageUnpacker unpacker, - List expectedTrace, - List stringTable) { - int spanCount = unpacker.unpackArrayHeader() - assertEquals(expectedTrace.size(), spanCount) - - for (int i = 0; i < spanCount; i++) { - verifySpan(unpacker, expectedTrace.get(i), stringTable) - } - return expectedTrace - } - - private static void verifySpan( - MessageUnpacker unpacker, - TraceGenerator.PojoSpan expectedSpan, - List stringTable) { - int spanFieldCount = unpacker.unpackMapHeader() - assertEquals(16, spanFieldCount) - - String service = null - String name = null - String resource = null - Long spanId = null - Long parentId = null - Long start = null - Long duration = null - Boolean error = null - Map attributes = null - String type = null - int linksCount = -1 - int eventsCount = -1 - String env = null - String version = null - String component = null - Integer spanKind = null - - for (int i = 0; i < spanFieldCount; i++) { - int fieldId = unpacker.unpackInt() - switch (fieldId) { - case 1: - service = readStreamingString(unpacker, stringTable) - break - case 2: - name = readStreamingString(unpacker, stringTable) - break - case 3: - resource = readStreamingString(unpacker, stringTable) - break - case 4: - spanId = unpackUnsignedLong(unpacker) - break - case 5: - parentId = unpackUnsignedLong(unpacker) - break - case 6: - start = unpacker.unpackLong() - break - case 7: - duration = unpacker.unpackLong() - break - case 8: - error = unpacker.unpackBoolean() - break - case 9: - attributes = readAttributes(unpacker, stringTable) - break - case 10: - type = readStreamingString(unpacker, stringTable) - break - case 11: - linksCount = unpacker.unpackArrayHeader() - break - case 12: - eventsCount = unpacker.unpackArrayHeader() - break - case 13: - env = readStreamingString(unpacker, stringTable) - break - case 14: - version = readStreamingString(unpacker, stringTable) - break - case 15: - component = readStreamingString(unpacker, stringTable) - break - case 16: - spanKind = unpacker.unpackInt() - break - default: - Assertions.fail("Unexpected span field id: " + fieldId) - } - } - - assertEqualsWithNullAsEmpty(expectedSpan.getServiceName(), service) - assertEqualsWithNullAsEmpty(expectedSpan.getOperationName(), name) - assertEqualsWithNullAsEmpty(expectedSpan.getResourceName(), resource) - assertEquals(expectedSpan.getSpanId(), spanId) - assertEquals(expectedSpan.getParentId(), parentId) - assertEquals(expectedSpan.getStartTime(), start) - assertEquals(expectedSpan.getDurationNano(), duration) - assertEquals(expectedSpan.getError() != 0, error) - assertEqualsWithNullAsEmpty(expectedSpan.getType(), type) - assertEquals(0, linksCount) - assertEquals(0, eventsCount) - assertEqualsWithNullAsEmpty(expectedSpan.getTag(Tags.ENV), env) - assertEqualsWithNullAsEmpty(expectedSpan.getTag(Tags.VERSION), version) - assertEqualsWithNullAsEmpty(expectedSpan.getTag(Tags.COMPONENT), component) - assertEquals(TraceMapperV1.getSpanKindValue(expectedSpan.getTag(Tags.SPAN_KIND)), spanKind) - - assertNotNull(attributes) - int expectedHttpStatusCode = expectedSpan.getHttpStatusCode() - boolean shouldContainHttpStatus = expectedHttpStatusCode != 0 && !expectedSpan.getTags().containsKey("http.status_code") - Map expectedAttributes = [:] - for (Map.Entry entry : expectedSpan.getBaggage().entrySet()) { - expectedAttributes.put(entry.getKey(), entry.getValue()) - } - expectedAttributes.put(DDTags.THREAD_ID, expectedSpan.getTag(DDTags.THREAD_ID)) - expectedAttributes.put(DDTags.THREAD_NAME, expectedSpan.getTag(DDTags.THREAD_NAME)) - for (Map.Entry entry : expectedSpan.getTags().entrySet()) { - if (DDTags.SPAN_EVENTS == entry.getKey()) { - continue - } - addFlattenedExpectedAttribute(expectedAttributes, entry.getKey(), entry.getValue()) - } - if (shouldContainHttpStatus) { - expectedAttributes.put("http.status_code", Integer.toString(expectedHttpStatusCode)) - } - if (expectedSpan.isTopLevel()) { - expectedAttributes.put(InstrumentationTags.DD_TOP_LEVEL.toString(), 1d) - } - - assertEquals(expectedAttributes.size(), attributes.size()) - for (Map.Entry entry : expectedAttributes.entrySet()) { - String key = entry.getKey() - Object expectedValue = entry.getValue() - assertTrue(attributes.containsKey(key), "Missing attribute key: $key") - assertAttributeValueEquals(expectedValue, attributes.get(key), key) - } - } - - private static void assertAttributeValueEquals(Object expected, Object actual, String key) { - if (expected instanceof Number) { - assertTrue(actual instanceof Number, "Attribute $key should be numeric") - double expectedValue = ((Number) expected).doubleValue() - double actualValue = ((Number) actual).doubleValue() - double delta = Math.max(0.000001d, Math.abs(expectedValue) * 0.000000000001d) - assertEquals(expectedValue, actualValue, delta, "Numeric mismatch for $key") - } else if (expected instanceof Boolean) { - assertEquals(expected, actual, "Boolean mismatch for $key") - } else { - assertEquals(String.valueOf(expected), String.valueOf(actual), "String mismatch for $key") - } - } - - private static void addFlattenedExpectedAttribute( - Map expectedAttributes, - String key, - Object value) { - if (!(value instanceof Map)) { - expectedAttributes.put(key, value) - return - } - for (Map.Entry entry : ((Map) value).entrySet()) { - addFlattenedExpectedAttribute( - expectedAttributes, - key + "." + String.valueOf(entry.getKey()), - entry.getValue()) - } - } - - private static int expectedSamplingMechanism(Map propagationMetadata) { - Object decisionMakerRaw = propagationMetadata.get("_dd.p.dm") - if (decisionMakerRaw == null) { - return SamplingMechanism.DEFAULT - } - - String decisionMaker = String.valueOf(decisionMakerRaw) - try { - int value = Integer.parseInt(decisionMaker) - return value < 0 ? -value : value - } catch (NumberFormatException ignored) { - int separator = decisionMaker.lastIndexOf('-') - if (separator >= 0 && separator + 1 < decisionMaker.length()) { - try { - int value = Integer.parseInt(decisionMaker.substring(separator + 1)) - return value < 0 ? -value : value - } catch (NumberFormatException ignoredAgain) { - } - } - return SamplingMechanism.DEFAULT - } - } - - private static byte[] serializeMappedPayload( - TraceMapperV1 mapper, - List> traces) { - CapturedBody capturedBody = new CapturedBody(mapper) - MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(2 << 20, capturedBody)) - - for (List trace : traces) { - assertTrue(packer.format(trace, mapper)) - } - packer.flush() - - assertNotNull(capturedBody.payloadBytes) - return capturedBody.payloadBytes - } - - private static byte[] serializePayload(Payload payload) { - ByteArrayChannel channel = new ByteArrayChannel() - payload.writeTo(channel) - return channel.bytes() - } - - private static class CapturedBody implements ByteBufferConsumer { - private final TraceMapperV1 mapper - private byte[] payloadBytes - - private CapturedBody(TraceMapperV1 mapper) { - this.mapper = mapper - } - - @Override - void accept(int messageCount, ByteBuffer buffer) { - Payload payload = mapper.newPayload().withBody(messageCount, buffer) - payloadBytes = serializePayload(payload) - mapper.reset() - } - } - - private static class CountingPojoSpan extends TraceGenerator.PojoSpan { - int processTagsAndBaggageCount = 0 - - CountingPojoSpan( - String serviceName, - String operationName, - CharSequence resourceName, - DDTraceId traceId, - long spanId, - long parentId, - long start, - long duration, - int error, - Map baggage, - Map tags, - CharSequence type, - boolean measured, - int samplingPriority, - int statusCode, - CharSequence origin) { - super( - serviceName, - operationName, - resourceName, - traceId, - spanId, - parentId, - start, - duration, - error, - baggage, - tags, - type, - measured, - samplingPriority, - statusCode, - origin) - } - - @Override - void processTagsAndBaggage(MetadataConsumer consumer) { - processTagsAndBaggageCount++ - super.processTagsAndBaggage(consumer) - } - } - - private static class ByteArrayChannel implements WritableByteChannel { - private byte[] data = new byte[0] - - @Override - int write(ByteBuffer src) { - int len = src.remaining() - byte[] incoming = new byte[len] - src.get(incoming) - byte[] combined = new byte[data.length + incoming.length] - System.arraycopy(data, 0, combined, 0, data.length) - System.arraycopy(incoming, 0, combined, data.length, incoming.length) - data = combined - return len - } - - byte[] bytes() { - return data - } - - @Override - boolean isOpen() { - return true - } - - @Override - void close() { - } - } - - private static void assertEqualsWithNullAsEmpty(CharSequence expected, CharSequence actual) { - if (expected == null) { - assertEquals("", actual) - } else { - assertEquals(expected.toString(), actual.toString()) - } - } -} diff --git a/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.java b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.java new file mode 100644 index 00000000000..0c92df7f0b6 --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.java @@ -0,0 +1,946 @@ +package datadog.trace.common.writer.ddagent; + +import static datadog.trace.api.DDTags.PROCESS_TAGS; +import static datadog.trace.api.DDTags.SPAN_EVENTS; +import static datadog.trace.api.DDTags.THREAD_ID; +import static datadog.trace.api.DDTags.THREAD_NAME; +import static datadog.trace.api.sampling.PrioritySampling.SAMPLER_KEEP; +import static datadog.trace.bootstrap.instrumentation.api.Tags.COMPONENT; +import static datadog.trace.bootstrap.instrumentation.api.Tags.ENV; +import static datadog.trace.bootstrap.instrumentation.api.Tags.HTTP_STATUS; +import static datadog.trace.bootstrap.instrumentation.api.Tags.HTTP_URL; +import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND; +import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND_CLIENT; +import static datadog.trace.bootstrap.instrumentation.api.Tags.VERSION; +import static datadog.trace.common.writer.TraceGenerator.generateRandomTraces; +import static datadog.trace.common.writer.ddagent.PayloadVerifiers.assertEqualsWithNullAsEmpty; +import static datadog.trace.common.writer.ddagent.V1PayloadReader.newStringTable; +import static datadog.trace.common.writer.ddagent.V1PayloadReader.readAttributes; +import static datadog.trace.common.writer.ddagent.V1PayloadReader.readBinary; +import static datadog.trace.common.writer.ddagent.V1PayloadReader.readFirstChunk; +import static datadog.trace.common.writer.ddagent.V1PayloadReader.readFirstSpan; +import static datadog.trace.common.writer.ddagent.V1PayloadReader.readStreamingString; +import static datadog.trace.common.writer.ddagent.V1PayloadReader.skipChunkField; +import static datadog.trace.common.writer.ddagent.V1PayloadReader.skipPayloadField; +import static datadog.trace.common.writer.ddagent.V1PayloadReader.skipSpanField; +import static datadog.trace.common.writer.ddagent.V1PayloadReader.traceIdBytes; +import static datadog.trace.common.writer.ddagent.V1PayloadReader.unpackUnsignedLong; +import static java.util.Arrays.asList; +import static java.util.Collections.emptyList; +import static java.util.Collections.emptyMap; +import static java.util.Collections.singletonList; +import static java.util.Collections.singletonMap; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; + +import datadog.communication.serialization.ByteBufferConsumer; +import datadog.communication.serialization.FlushingBuffer; +import datadog.communication.serialization.msgpack.MsgPackWriter; +import datadog.trace.api.Config; +import datadog.trace.api.DDSpanId; +import datadog.trace.api.DDTraceId; +import datadog.trace.api.ProcessTags; +import datadog.trace.api.sampling.SamplingMechanism; +import datadog.trace.bootstrap.instrumentation.api.AgentSpanLink; +import datadog.trace.bootstrap.instrumentation.api.InstrumentationTags; +import datadog.trace.bootstrap.instrumentation.api.SpanAttributes; +import datadog.trace.bootstrap.instrumentation.api.SpanLink; +import datadog.trace.common.writer.Payload; +import datadog.trace.common.writer.TraceGenerator.PojoSpan; +import datadog.trace.common.writer.ddagent.V1PayloadReader.ChunkField; +import datadog.trace.common.writer.ddagent.V1PayloadReader.PayloadField; +import datadog.trace.common.writer.ddagent.V1PayloadReader.SpanField; +import datadog.trace.common.writer.ddagent.V1PayloadReader.V1SpanEvent; +import datadog.trace.common.writer.ddagent.V1PayloadReader.V1SpanLink; +import datadog.trace.core.MetadataConsumer; +import datadog.trace.test.junit.utils.config.WithConfigExtension; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.Channels; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.msgpack.core.MessageFormat; +import org.msgpack.core.MessagePack; +import org.msgpack.core.MessageUnpacker; +import org.msgpack.value.ValueType; +import org.tabletest.junit.TableTest; + +@ExtendWith(WithConfigExtension.class) +class TraceMapperV1PayloadTest { + + private static final String DECISION_MAKER_TAG = "_dd.p.dm"; + + /** Every field the v1 payload header carries, mirroring {@code TraceMapperV1.buildHeader}. */ + private static final Set EXPECTED_PAYLOAD_FIELD_IDS = + new HashSet<>( + asList( + PayloadField.CONTAINER_ID, + PayloadField.LANGUAGE_NAME, + PayloadField.LANGUAGE_VERSION, + PayloadField.TRACER_VERSION, + PayloadField.RUNTIME_ID, + PayloadField.ENV, + PayloadField.HOSTNAME, + PayloadField.APP_VERSION, + PayloadField.ATTRIBUTES, + PayloadField.CHUNKS)); + + // Keep the ProcessTags static in sync with the (per-test rebuilt) Config, the way DDSpecification + // did for the original Spock tests. Runs after WithConfigExtension has rebuilt Config. + @BeforeEach + void syncProcessTags() { + ProcessTags.reset(Config.get()); + } + + @TableTest({ + "scenario | bufferSizeKb | traceCount | lowCardinality", + "no traces, low cardinality | 20 | 0 | true ", + "one trace, low cardinality | 20 | 1 | true ", + "two traces, low cardinality | 30 | 2 | true ", + "no traces, high cardinality | 20 | 0 | false ", + "one trace, high cardinality | 20 | 1 | false ", + "two traces, high cardinality | 30 | 2 | false ", + "ten traces, low cardinality | 100 | 10 | true ", + "hundred traces, high cardinality | 100 | 100 | false " + }) + void tracesWrittenCorrectly( + String scenario, int bufferSizeKb, int traceCount, boolean lowCardinality) { + List> traces = generateRandomTraces(traceCount, lowCardinality); + TraceMapperV1 traceMapper = new TraceMapperV1(); + PayloadVerifier verifier = new PayloadVerifier(traces, traceMapper); + MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(bufferSizeKb << 10, verifier)); + + boolean tracesFitInBuffer = true; + for (List trace : traces) { + if (!packer.format(trace, traceMapper)) { + verifier.skipLargeTrace(); + tracesFitInBuffer = false; + traceMapper.reset(); + } + } + packer.flush(); + + if (tracesFitInBuffer) { + verifier.verifyTracesConsumed(); + } + } + + @Test + void endpointReturnsV1() { + assertEquals("v1.0", new TraceMapperV1().endpoint()); + } + + // expectedKind holds the wire values, which must stay in sync with TraceMapperV1.SPAN_KIND_*. + @TableTest({ + "scenario | spanKind | expectedKind", + "no span.kind tag | | 0 ", + "internal span kind | internal | 1 ", + "server span kind | server | 2 ", + "client span kind | client | 3 ", + "producer span kind | producer | 4 ", + "consumer span kind | consumer | 5 ", + "unrecognized span kind | unknown | 1 " + }) + void spanKindValueConversion(String scenario, String spanKind, int expectedKind) { + assertEquals(expectedKind, TraceMapperV1.getSpanKindValue(spanKind)); + } + + @Test + void payloadContainsExpectedHeaderAndChunkFields() throws IOException { + Map tags = new HashMap<>(); + tags.put(ENV, "prod"); + tags.put(VERSION, "1.2.3"); + tags.put(COMPONENT, "http-client"); + tags.put(SPAN_KIND, SPAN_KIND_CLIENT); + tags.put("attr.string", "value"); + tags.put("attr.bool", true); + tags.put("attr.number", 12.5d); + PojoSpan span = + new PojoSpan( + "service-a", + "operation-a", + "resource-a", + DDTraceId.ONE, + 123L, + 0L, + 1000L, + 2000L, + 1, + singletonMap(DECISION_MAKER_TAG, "-3"), + tags, + "web", + false, + SAMPLER_KEEP, + 200, + "rum"); + + byte[] encoded = serializeV1Payload(span); + MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded); + List stringTable = newStringTable(); + + int payloadFieldCount = unpacker.unpackMapHeader(); + Set payloadFieldsSeen = new HashSet<>(); + int chunkCount = -1; + Map payloadAttributes = null; + + for (int i = 0; i < payloadFieldCount; i++) { + int fieldId = unpacker.unpackInt(); + payloadFieldsSeen.add(fieldId); + if (fieldId == PayloadField.CHUNKS) { + chunkCount = unpacker.unpackArrayHeader(); + assertEquals(1, chunkCount); + verifyChunk(unpacker, singletonList(span), stringTable); + } else if (fieldId == PayloadField.ATTRIBUTES) { + payloadAttributes = readAttributes(unpacker, stringTable); + } else { + skipPayloadField(unpacker, fieldId, stringTable); + } + } + + assertEquals(EXPECTED_PAYLOAD_FIELD_IDS.size(), payloadFieldCount); + assertEquals(EXPECTED_PAYLOAD_FIELD_IDS, payloadFieldsSeen); + assertEquals(1, chunkCount); + assertNotNull(payloadAttributes); + CharSequence processTags = ProcessTags.getTagsForSerialization(); + if (processTags == null) { + assertEquals(0, payloadAttributes.size()); + } else { + assertEquals(1, payloadAttributes.size()); + assertEquals(processTags.toString(), payloadAttributes.get(PROCESS_TAGS)); + } + } + + // expectedSamplingMechanism 0 is SamplingMechanism.DEFAULT. + @TableTest({ + "scenario | decisionMakerTag | expectedSamplingMechanism", + "no _dd.p.dm tag | | 0 ", + "negative numeric value | '-3' | 3 ", + "hashed prefix before the dash | '934086a686-7' | 7 ", + "unparseable value | invalid | 0 " + }) + void samplingMechanismNormalizationFromDecisionMaker( + String scenario, String decisionMakerTag, int expectedSamplingMechanism) throws IOException { + Map baggage = new HashMap<>(); + if (decisionMakerTag != null) { + baggage.put(DECISION_MAKER_TAG, decisionMakerTag); + } + + byte[] encoded = serializeV1Payload(spanWithBaggage(baggage)); + + assertEquals(expectedSamplingMechanism, readFirstChunk(encoded).getSamplingMechanism()); + } + + @Test + void spanIdsAreEncodedAsUnsignedValues() throws IOException { + long spanId = Long.MIN_VALUE + 123L; + long parentId = Long.MIN_VALUE + 456L; + + byte[] encoded = serializeV1Payload(spanWithIds(spanId, parentId)); + MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded); + List stringTable = newStringTable(); + + Long actualSpanId = null; + Long actualParentId = null; + int payloadFieldCount = unpacker.unpackMapHeader(); + for (int i = 0; i < payloadFieldCount; i++) { + int payloadFieldId = unpacker.unpackInt(); + if (payloadFieldId != PayloadField.CHUNKS) { + skipPayloadField(unpacker, payloadFieldId, stringTable); + continue; + } + assertEquals(1, unpacker.unpackArrayHeader()); + int chunkFieldCount = unpacker.unpackMapHeader(); + for (int j = 0; j < chunkFieldCount; j++) { + int chunkFieldId = unpacker.unpackInt(); + if (chunkFieldId != ChunkField.SPANS) { + skipChunkField(unpacker, chunkFieldId, stringTable); + continue; + } + assertEquals(1, unpacker.unpackArrayHeader()); + int spanFieldCount = unpacker.unpackMapHeader(); + for (int k = 0; k < spanFieldCount; k++) { + int spanFieldId = unpacker.unpackInt(); + switch (spanFieldId) { + case SpanField.SPAN_ID: + actualSpanId = unpackUint64(unpacker); + break; + case SpanField.PARENT_ID: + actualParentId = unpackUint64(unpacker); + break; + default: + skipSpanField(unpacker, spanFieldId, stringTable); + } + } + } + } + + assertEquals(spanId, actualSpanId); + assertEquals(parentId, actualParentId); + } + + @Test + void spanLinksAreEncodedFromStructuredSpanLinks() throws IOException { + DDTraceId firstLinkTraceId = DDTraceId.fromHex("11223344556677889900aabbccddeeff"); + long firstLinkSpanId = DDSpanId.fromHex("000000000000002a"); + Map firstLinkAttributes = new HashMap<>(); + firstLinkAttributes.put("link.kind", "follows_from"); + firstLinkAttributes.put("context_headers", "tracecontext"); + DDTraceId secondLinkTraceId = DDTraceId.fromHex("00000000000000000000000000000001"); + long secondLinkSpanId = DDSpanId.fromHex("0000000000000002"); + List spanLinks = new ArrayList<>(); + spanLinks.add( + new TestSpanLink( + firstLinkTraceId, + firstLinkSpanId, + (byte) 1, + "dd=s:1", + SpanAttributes.fromMap(firstLinkAttributes))); + spanLinks.add( + new TestSpanLink(secondLinkTraceId, secondLinkSpanId, (byte) 0, "", SpanAttributes.EMPTY)); + + List links = readFirstSpan(serializeV1Payload(spanWithLinks(spanLinks))).getLinks(); + + assertEquals(2, links.size()); + V1SpanLink firstLink = links.get(0); + assertArrayEquals(traceIdBytes(firstLinkTraceId), firstLink.getTraceId()); + assertEquals(firstLinkSpanId, firstLink.getSpanId()); + assertEquals("dd=s:1", firstLink.getTraceState()); + assertEquals(1L, firstLink.getTraceFlags()); + assertEquals(firstLinkAttributes, firstLink.getAttributes()); + + V1SpanLink secondLink = links.get(1); + assertArrayEquals(traceIdBytes(secondLinkTraceId), secondLink.getTraceId()); + assertEquals(secondLinkSpanId, secondLink.getSpanId()); + assertEquals("", secondLink.getTraceState()); + assertEquals(0L, secondLink.getTraceFlags()); + assertEquals(emptyMap(), secondLink.getAttributes()); + } + + @Test + void firstSpanTagsAreProcessedOnce() { + CountingPojoSpan firstSpan = new CountingPojoSpan("operation-a", "resource-a", 123L, 0L); + CountingPojoSpan secondSpan = new CountingPojoSpan("operation-b", "resource-b", 456L, 123L); + List trace = new ArrayList<>(); + trace.add(firstSpan); + trace.add(secondSpan); + + serializeV1Payload(trace); + + assertEquals(1, firstSpan.processTagsAndBaggageCount); + assertEquals(1, secondSpan.processTagsAndBaggageCount); + } + + @Test + void missingSpanLinksEncodeEmptyLinks() throws IOException { + byte[] encoded = serializeV1Payload(span(emptyMap())); + + assertTrue(readFirstSpan(encoded).getLinks().isEmpty()); + } + + @Test + void spanEventsAreEncodedFromEventsTag() throws IOException { + Map firstEventAttributes = new HashMap<>(); + firstEventAttributes.put("str", "v"); + firstEventAttributes.put("int", 42L); + firstEventAttributes.put("double", 12.5d); + firstEventAttributes.put("bool", true); + firstEventAttributes.put("arr", asList("x", 7L, 2.5d, false)); + Map firstEvent = new HashMap<>(); + firstEvent.put("time_unix_nano", 1234567890L); + firstEvent.put("name", "event.one"); + firstEvent.put("attributes", firstEventAttributes); + Map secondEvent = new HashMap<>(); + secondEvent.put("time_unix_nano", 1234567891L); + secondEvent.put("name", "event.two"); + PojoSpan span = span(singletonMap(SPAN_EVENTS, asList(firstEvent, secondEvent))); + + List events = readFirstSpan(serializeV1Payload(span)).getEvents(); + + assertEquals(2, events.size()); + V1SpanEvent firstDecodedEvent = events.get(0); + assertEquals(1234567890L, firstDecodedEvent.getTimeUnixNano()); + assertEquals("event.one", firstDecodedEvent.getName()); + Map firstDecodedAttributes = firstDecodedEvent.getAttributes(); + assertEquals("v", firstDecodedAttributes.get("str")); + assertEquals(42L, firstDecodedAttributes.get("int")); + assertAttributeValueEquals(12.5d, firstDecodedAttributes.get("double"), "double"); + assertEquals(true, firstDecodedAttributes.get("bool")); + assertEquals(asList("x", 7L, 2.5d, false), firstDecodedAttributes.get("arr")); + + V1SpanEvent secondDecodedEvent = events.get(1); + assertEquals(1234567891L, secondDecodedEvent.getTimeUnixNano()); + assertEquals("event.two", secondDecodedEvent.getName()); + assertEquals(emptyMap(), secondDecodedEvent.getAttributes()); + } + + @Test + void malformedSpanEventsFallBackToEmptyEvents() throws IOException { + // a map instead of the expected list of events + PojoSpan span = span(singletonMap(SPAN_EVENTS, singletonMap("foo", "bar"))); + + assertTrue(readFirstSpan(serializeV1Payload(span)).getEvents().isEmpty()); + } + + @Test + void metaStructIsEncodedAsBytesAttribute() throws IOException { + PojoSpan span = span(emptyMap()); + Map metaStructValue = new HashMap<>(); + metaStructValue.put("foo", "bar"); + metaStructValue.put("answer", 42L); + span.setMetaStruct("meta_key", metaStructValue); + + Map attributes = readFirstSpan(serializeV1Payload(span)).getAttributes(); + byte[] metaStructBytes = (byte[]) attributes.get("meta_key"); + assertNotNull(metaStructBytes); + + MessageUnpacker metaStructUnpacker = MessagePack.newDefaultUnpacker(metaStructBytes); + int metaStructFieldCount = metaStructUnpacker.unpackMapHeader(); + Map decodedMetaStruct = new HashMap<>(); + for (int i = 0; i < metaStructFieldCount; i++) { + String key = metaStructUnpacker.unpackString(); + ValueType valueType = metaStructUnpacker.getNextFormat().getValueType(); + switch (valueType) { + case INTEGER: + decodedMetaStruct.put(key, metaStructUnpacker.unpackLong()); + break; + case STRING: + decodedMetaStruct.put(key, metaStructUnpacker.unpackString()); + break; + default: + fail("Unexpected meta_struct value type for key " + key); + } + } + + assertEquals("bar", decodedMetaStruct.get("foo")); + assertEquals(42L, decodedMetaStruct.get("answer")); + } + + @Test + void mapValuedSpanTagsAreFlattenedInAttributes() throws IOException { + Map user = new HashMap<>(); + user.put("id", "123"); + user.put("name", "alice"); + user.put("authenticated", true); + user.put("profile", singletonMap("age", 30L)); + Map loginMetadata = new HashMap<>(); + loginMetadata.put("event", "login"); + loginMetadata.put("attempts", 1L); + Map loginSuccess = new HashMap<>(); + loginSuccess.put("metadata0", loginMetadata); + loginSuccess.put("metadata1", singletonMap("blocked", false)); + Map tags = new HashMap<>(); + tags.put("usr", user); + String loginSuccessTag = "appsec.events.users.login.success"; + tags.put(loginSuccessTag, loginSuccess); + + // status code 0 keeps the encoder from adding an http.status_code attribute + Map attributes = + readFirstSpan(serializeV1Payload(span(tags, 0))).getAttributes(); + + assertEquals("123", attributes.get("usr.id")); + assertEquals("alice", attributes.get("usr.name")); + assertEquals(true, attributes.get("usr.authenticated")); + assertAttributeValueEquals(30L, attributes.get("usr.profile.age"), "usr.profile.age"); + assertEquals("login", attributes.get(loginSuccessTag + ".metadata0.event")); + assertAttributeValueEquals( + 1L, attributes.get(loginSuccessTag + ".metadata0.attempts"), "attempts"); + assertEquals(false, attributes.get(loginSuccessTag + ".metadata1.blocked")); + // the 7 flattened entries plus thread.id and thread.name, and nothing else + assertEquals(9, attributes.size()); + + // the map-valued tags themselves are replaced by their flattened entries + assertFalse(attributes.containsKey("usr")); + assertFalse(attributes.containsKey(loginSuccessTag)); + } + + @Test + void primitiveSpanTagsAreEncodedInAttributes() throws IOException { + Map tags = new HashMap<>(); + tags.put("tag.bool", true); + tags.put("tag.int", 7); + tags.put("tag.long", 9L); + tags.put("tag.float", 3.5f); + tags.put("tag.double", 4.25d); + + // status code 0 keeps the encoder from adding an http.status_code attribute + Map attributes = + readFirstSpan(serializeV1Payload(span(tags, 0))).getAttributes(); + + assertEquals(true, attributes.get("tag.bool")); + assertAttributeValueEquals(7, attributes.get("tag.int"), "tag.int"); + assertAttributeValueEquals(9L, attributes.get("tag.long"), "tag.long"); + assertAttributeValueEquals(3.5f, attributes.get("tag.float"), "tag.float"); + assertAttributeValueEquals(4.25d, attributes.get("tag.double"), "tag.double"); + // the 5 tags plus thread.id and thread.name, and nothing else + assertEquals(7, attributes.size()); + } + + @Test + void threadMetadataIsEncodedInAttributes() throws IOException { + // status code 0 keeps the encoder from adding an http.status_code attribute + PojoSpan span = span(emptyMap(), 0); + + Map attributes = readFirstSpan(serializeV1Payload(span)).getAttributes(); + + assertAttributeValueEquals(span.getTag(THREAD_ID), attributes.get(THREAD_ID), THREAD_ID); + Object expectedThreadName = span.getTag(THREAD_NAME); + assertEquals(expectedThreadName.toString(), attributes.get(THREAD_NAME)); + // thread metadata is all a tagless span encodes + assertEquals(2, attributes.size()); + } + + private static final class PayloadVerifier implements ByteBufferConsumer { + + private final List> expectedTraces; + private final TraceMapperV1 mapper; + private final PayloadVerifiers.CapturingChannel channel = + new PayloadVerifiers.CapturingChannel(200 << 10); + + private int position = 0; + + private PayloadVerifier(List> expectedTraces, TraceMapperV1 mapper) { + this.expectedTraces = expectedTraces; + this.mapper = mapper; + } + + void skipLargeTrace() { + ++position; + } + + @Override + public void accept(int messageCount, ByteBuffer buffer) { + if (expectedTraces.isEmpty() && messageCount == 0) { + return; + } + try { + Payload payload = mapper.newPayload().withBody(messageCount, buffer); + payload.writeTo(channel); + MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(channel.flipForReading()); + if (messageCount == 0) { + assertEquals(0, unpacker.unpackMapHeader()); + return; + } + + List stringTable = newStringTable(); + int payloadFieldCount = unpacker.unpackMapHeader(); + assertEquals(EXPECTED_PAYLOAD_FIELD_IDS.size(), payloadFieldCount); + + boolean seenChunks = false; + for (int i = 0; i < payloadFieldCount; i++) { + int fieldId = unpacker.unpackInt(); + if (fieldId == PayloadField.CHUNKS) { + int traceCount = unpacker.unpackArrayHeader(); + assertEquals(messageCount, traceCount); + seenChunks = true; + for (int traceIndex = 0; traceIndex < traceCount; traceIndex++) { + verifyChunk(unpacker, expectedTraces.get(position++), stringTable); + } + } else { + skipPayloadField(unpacker, fieldId, stringTable); + } + } + + assertTrue(seenChunks); + } catch (IOException e) { + fail(e.getMessage()); + } finally { + mapper.reset(); + channel.resetForWriting(); + } + } + + void verifyTracesConsumed() { + assertEquals(expectedTraces.size(), position); + } + } + + private static void verifyChunk( + MessageUnpacker unpacker, List expectedTrace, List stringTable) + throws IOException { + int chunkFieldCount = unpacker.unpackMapHeader(); + assertEquals(6, chunkFieldCount); + + Integer priority = null; + String origin = null; + Map chunkAttributes = null; + byte[] traceId = null; + Integer samplingMechanism = null; + boolean seenSpans = false; + + for (int i = 0; i < chunkFieldCount; i++) { + int fieldId = unpacker.unpackInt(); + switch (fieldId) { + case ChunkField.PRIORITY: + priority = unpacker.unpackInt(); + break; + case ChunkField.ORIGIN: + origin = readStreamingString(unpacker, stringTable); + break; + case ChunkField.ATTRIBUTES: + chunkAttributes = readAttributes(unpacker, stringTable); + break; + case ChunkField.SPANS: + verifySpans(unpacker, expectedTrace, stringTable); + seenSpans = true; + break; + case ChunkField.TRACE_ID: + traceId = readBinary(unpacker); + break; + case ChunkField.SAMPLING_MECHANISM: + samplingMechanism = unpacker.unpackInt(); + break; + default: + fail("Unexpected chunk field id: " + fieldId); + } + } + + assertNotNull(priority); + assertNotNull(origin); + assertNotNull(chunkAttributes); + assertTrue(seenSpans); + assertNotNull(traceId); + assertNotNull(samplingMechanism); + + PojoSpan firstSpan = expectedTrace.get(0); + assertEquals(firstSpan.samplingPriority(), priority.intValue()); + assertEqualsWithNullAsEmpty(firstSpan.getOrigin(), origin); + assertEquals(1, chunkAttributes.size()); + assertEqualsWithNullAsEmpty( + firstSpan.getLocalRootSpan().getServiceName(), (String) chunkAttributes.get("service")); + assertArrayEquals(traceIdBytes(firstSpan.getTraceId()), traceId); + assertEquals(expectedSamplingMechanism(firstSpan.getBaggage()), samplingMechanism.intValue()); + } + + private static void verifySpans( + MessageUnpacker unpacker, List expectedTrace, List stringTable) + throws IOException { + int spanCount = unpacker.unpackArrayHeader(); + assertEquals(expectedTrace.size(), spanCount); + + for (int i = 0; i < spanCount; i++) { + verifySpan(unpacker, expectedTrace.get(i), stringTable); + } + } + + private static void verifySpan( + MessageUnpacker unpacker, PojoSpan expectedSpan, List stringTable) + throws IOException { + int spanFieldCount = unpacker.unpackMapHeader(); + assertEquals(16, spanFieldCount); + + String service = null; + String name = null; + String resource = null; + long spanId = 0; + long parentId = 0; + long start = 0; + long duration = 0; + boolean error = false; + Map attributes = null; + String type = null; + int linksCount = -1; + int eventsCount = -1; + String env = null; + String version = null; + String component = null; + int spanKind = -1; + + for (int i = 0; i < spanFieldCount; i++) { + int fieldId = unpacker.unpackInt(); + switch (fieldId) { + case SpanField.SERVICE: + service = readStreamingString(unpacker, stringTable); + break; + case SpanField.NAME: + name = readStreamingString(unpacker, stringTable); + break; + case SpanField.RESOURCE: + resource = readStreamingString(unpacker, stringTable); + break; + case SpanField.SPAN_ID: + spanId = unpackUnsignedLong(unpacker); + break; + case SpanField.PARENT_ID: + parentId = unpackUnsignedLong(unpacker); + break; + case SpanField.START: + start = unpacker.unpackLong(); + break; + case SpanField.DURATION: + duration = unpacker.unpackLong(); + break; + case SpanField.ERROR: + error = unpacker.unpackBoolean(); + break; + case SpanField.ATTRIBUTES: + attributes = readAttributes(unpacker, stringTable); + break; + case SpanField.TYPE: + type = readStreamingString(unpacker, stringTable); + break; + case SpanField.LINKS: + linksCount = unpacker.unpackArrayHeader(); + break; + case SpanField.EVENTS: + eventsCount = unpacker.unpackArrayHeader(); + break; + case SpanField.ENV: + env = readStreamingString(unpacker, stringTable); + break; + case SpanField.VERSION: + version = readStreamingString(unpacker, stringTable); + break; + case SpanField.COMPONENT: + component = readStreamingString(unpacker, stringTable); + break; + case SpanField.KIND: + spanKind = unpacker.unpackInt(); + break; + default: + fail("Unexpected span field id: " + fieldId); + } + } + + assertEqualsWithNullAsEmpty(expectedSpan.getServiceName(), service); + assertEqualsWithNullAsEmpty(expectedSpan.getOperationName(), name); + assertEqualsWithNullAsEmpty(expectedSpan.getResourceName(), resource); + assertEquals(expectedSpan.getSpanId(), spanId); + assertEquals(expectedSpan.getParentId(), parentId); + assertEquals(expectedSpan.getStartTime(), start); + assertEquals(expectedSpan.getDurationNano(), duration); + assertEquals(expectedSpan.getError() != 0, error); + assertEqualsWithNullAsEmpty(expectedSpan.getType(), type); + assertEquals(0, linksCount); + assertEquals(0, eventsCount); + assertEqualsWithNullAsEmpty(expectedSpan.getTag(ENV), env); + assertEqualsWithNullAsEmpty(expectedSpan.getTag(VERSION), version); + assertEqualsWithNullAsEmpty(expectedSpan.getTag(COMPONENT), component); + assertEquals( + TraceMapperV1.getSpanKindValue(expectedSpan.getTag(SPAN_KIND)), spanKind); + + assertNotNull(attributes); + int expectedHttpStatusCode = expectedSpan.getHttpStatusCode(); + boolean shouldContainHttpStatus = + expectedHttpStatusCode != 0 && !expectedSpan.getTags().containsKey(HTTP_STATUS); + Map expectedAttributes = new HashMap<>(expectedSpan.getBaggage()); + expectedAttributes.put(THREAD_ID, expectedSpan.getTag(THREAD_ID)); + expectedAttributes.put(THREAD_NAME, expectedSpan.getTag(THREAD_NAME)); + for (Map.Entry entry : expectedSpan.getTags().entrySet()) { + if (SPAN_EVENTS.equals(entry.getKey())) { + continue; + } + addFlattenedExpectedAttribute(expectedAttributes, entry.getKey(), entry.getValue()); + } + if (shouldContainHttpStatus) { + expectedAttributes.put(HTTP_STATUS, Integer.toString(expectedHttpStatusCode)); + } + if (expectedSpan.isTopLevel()) { + expectedAttributes.put(InstrumentationTags.DD_TOP_LEVEL.toString(), 1d); + } + + assertEquals(expectedAttributes.size(), attributes.size()); + for (Map.Entry entry : expectedAttributes.entrySet()) { + String key = entry.getKey(); + assertTrue(attributes.containsKey(key), "Missing attribute key: " + key); + assertAttributeValueEquals(entry.getValue(), attributes.get(key), key); + } + } + + private static void assertAttributeValueEquals(Object expected, Object actual, String key) { + if (expected instanceof Number) { + assertInstanceOf(Number.class, actual, "Attribute " + key + " should be numeric"); + double expectedValue = ((Number) expected).doubleValue(); + double actualValue = ((Number) actual).doubleValue(); + double delta = Math.max(0.000001d, Math.abs(expectedValue) * 0.000000000001d); + assertEquals(expectedValue, actualValue, delta, "Numeric mismatch for " + key); + } else if (expected instanceof Boolean) { + assertEquals(expected, actual, "Boolean mismatch for " + key); + } else { + assertEquals(String.valueOf(expected), String.valueOf(actual), "String mismatch for " + key); + } + } + + private static void addFlattenedExpectedAttribute( + Map expectedAttributes, String key, Object value) { + if (!(value instanceof Map)) { + expectedAttributes.put(key, value); + return; + } + for (Map.Entry entry : ((Map) value).entrySet()) { + addFlattenedExpectedAttribute( + expectedAttributes, key + "." + String.valueOf(entry.getKey()), entry.getValue()); + } + } + + private static int expectedSamplingMechanism(Map propagationMetadata) { + String decisionMaker = propagationMetadata.get(DECISION_MAKER_TAG); + if (decisionMaker == null) { + return SamplingMechanism.DEFAULT; + } + + try { + return Math.abs(Integer.parseInt(decisionMaker)); + } catch (NumberFormatException ignored) { + int separator = decisionMaker.lastIndexOf('-'); + if (separator >= 0 && separator + 1 < decisionMaker.length()) { + try { + return Math.abs(Integer.parseInt(decisionMaker.substring(separator + 1))); + } catch (NumberFormatException ignoredAgain) { + // not a mechanism suffix either, fall through to the default + } + } + return SamplingMechanism.DEFAULT; + } + } + + /** Asserts the next value is encoded as an unsigned 64-bit integer, and returns it. */ + private static long unpackUint64(MessageUnpacker unpacker) throws IOException { + assertEquals(MessageFormat.UINT64, unpacker.getNextFormat()); + return unpackUnsignedLong(unpacker); + } + + private static byte[] serializeV1Payload(PojoSpan span) { + return serializeV1Payload(singletonList(span)); + } + + /** Maps a single trace and returns the complete v1 payload bytes. */ + private static byte[] serializeV1Payload(List trace) { + TraceMapperV1 mapper = new TraceMapperV1(); + CapturedBody capturedBody = new CapturedBody(mapper); + MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(2 << 20, capturedBody)); + + assertTrue(packer.format(trace, mapper)); + packer.flush(); + + assertNotNull(capturedBody.payloadBytes); + return capturedBody.payloadBytes; + } + + private static PojoSpan span(Map tags) { + return span(123L, 0L, emptyMap(), tags, 200, emptyList()); + } + + private static PojoSpan span(Map tags, int statusCode) { + return span(123L, 0L, emptyMap(), tags, statusCode, emptyList()); + } + + private static PojoSpan spanWithBaggage(Map baggage) { + return span(123L, 0L, baggage, emptyMap(), 200, emptyList()); + } + + private static PojoSpan spanWithIds(long spanId, long parentId) { + return span(spanId, parentId, emptyMap(), emptyMap(), 200, emptyList()); + } + + private static PojoSpan spanWithLinks(List spanLinks) { + return span(123L, 0L, emptyMap(), emptyMap(), 200, spanLinks); + } + + /** A span carrying the fields shared by every test; only the varying pieces are parameters. */ + private static PojoSpan span( + long spanId, + long parentId, + Map baggage, + Map tags, + int statusCode, + List spanLinks) { + return new PojoSpan( + "service-a", + "operation-a", + "resource-a", + DDTraceId.ONE, + spanId, + parentId, + 1000L, + 2000L, + 0, + baggage, + tags, + "web", + false, + SAMPLER_KEEP, + statusCode, + null, + spanLinks); + } + + private static final class CapturedBody implements ByteBufferConsumer { + private final TraceMapperV1 mapper; + private byte[] payloadBytes; + + private CapturedBody(TraceMapperV1 mapper) { + this.mapper = mapper; + } + + @Override + public void accept(int messageCount, ByteBuffer buffer) { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + try { + Payload payload = mapper.newPayload().withBody(messageCount, buffer); + payload.writeTo(Channels.newChannel(out)); + payloadBytes = out.toByteArray(); + } catch (IOException e) { + fail("Failed to serialize the v1 payload: " + e.getMessage()); + } finally { + mapper.reset(); + } + } + } + + /** A span counting how many times its tags and baggage were handed to the mapper. */ + private static final class CountingPojoSpan extends PojoSpan { + private int processTagsAndBaggageCount = 0; + + private CountingPojoSpan( + String operationName, CharSequence resourceName, long spanId, long parentId) { + super( + "service-a", + operationName, + resourceName, + DDTraceId.ONE, + spanId, + parentId, + 1000L, + 2000L, + 0, + emptyMap(), + singletonMap(HTTP_URL, "http://localhost:7777/"), + "web", + false, + SAMPLER_KEEP, + 200, + null); + } + + @Override + public void processTagsAndBaggage(MetadataConsumer consumer) { + processTagsAndBaggageCount++; + super.processTagsAndBaggage(consumer); + } + } + + /** {@link SpanLink}'s constructor is protected, so tests reach it through a subclass. */ + private static final class TestSpanLink extends SpanLink { + private TestSpanLink( + DDTraceId traceId, + long spanId, + byte traceFlags, + String traceState, + SpanAttributes attributes) { + super(traceId, spanId, traceFlags, traceState, attributes); + } + } +} diff --git a/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/V1PayloadReader.java b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/V1PayloadReader.java index b2e357bc80a..4dfdaf4bbd7 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/V1PayloadReader.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/V1PayloadReader.java @@ -37,7 +37,7 @@ public final class V1PayloadReader { /** msgpack field ids of the top-level payload (header) map, mirroring {@code buildHeader}. */ - private static final class PayloadField { + static final class PayloadField { static final int CONTAINER_ID = 2; static final int LANGUAGE_NAME = 3; static final int LANGUAGE_VERSION = 4; @@ -53,7 +53,7 @@ private PayloadField() {} } /** msgpack field ids of a trace chunk map, mirroring {@code TraceMapperV1.map} (no field 5). */ - private static final class ChunkField { + static final class ChunkField { static final int PRIORITY = 1; static final int ORIGIN = 2; static final int ATTRIBUTES = 3; @@ -65,7 +65,7 @@ private ChunkField() {} } /** msgpack field ids of a span map, mirroring {@code encodeSpans} (16 fields). */ - private static final class SpanField { + static final class SpanField { static final int SERVICE = 1; static final int NAME = 2; static final int RESOURCE = 3; From 607440569bc019221d58ded1f2b9de1ad7e9dc2e Mon Sep 17 00:00:00 2001 From: Alexey Kuznetsov Date: Fri, 31 Jul 2026 14:59:47 -0400 Subject: [PATCH 2/3] Code cleanup. --- .../ddagent/TraceMapperV1PayloadTest.java | 197 +++++++++--------- 1 file changed, 101 insertions(+), 96 deletions(-) diff --git a/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.java b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.java index 0c92df7f0b6..5676bc88716 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.java @@ -188,39 +188,40 @@ void payloadContainsExpectedHeaderAndChunkFields() throws IOException { "rum"); byte[] encoded = serializeV1Payload(span); - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded); List stringTable = newStringTable(); - int payloadFieldCount = unpacker.unpackMapHeader(); - Set payloadFieldsSeen = new HashSet<>(); - int chunkCount = -1; - Map payloadAttributes = null; + try (MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded)) { + int payloadFieldCount = unpacker.unpackMapHeader(); + Set payloadFieldsSeen = new HashSet<>(); + int chunkCount = -1; + Map payloadAttributes = null; + + for (int i = 0; i < payloadFieldCount; i++) { + int fieldId = unpacker.unpackInt(); + payloadFieldsSeen.add(fieldId); + if (fieldId == PayloadField.CHUNKS) { + chunkCount = unpacker.unpackArrayHeader(); + assertEquals(1, chunkCount); + verifyChunk(unpacker, singletonList(span), stringTable); + } else if (fieldId == PayloadField.ATTRIBUTES) { + payloadAttributes = readAttributes(unpacker, stringTable); + } else { + skipPayloadField(unpacker, fieldId, stringTable); + } + } - for (int i = 0; i < payloadFieldCount; i++) { - int fieldId = unpacker.unpackInt(); - payloadFieldsSeen.add(fieldId); - if (fieldId == PayloadField.CHUNKS) { - chunkCount = unpacker.unpackArrayHeader(); - assertEquals(1, chunkCount); - verifyChunk(unpacker, singletonList(span), stringTable); - } else if (fieldId == PayloadField.ATTRIBUTES) { - payloadAttributes = readAttributes(unpacker, stringTable); + assertEquals(EXPECTED_PAYLOAD_FIELD_IDS.size(), payloadFieldCount); + assertEquals(EXPECTED_PAYLOAD_FIELD_IDS, payloadFieldsSeen); + assertEquals(1, chunkCount); + assertNotNull(payloadAttributes); + CharSequence processTags = ProcessTags.getTagsForSerialization(); + if (processTags == null) { + assertEquals(0, payloadAttributes.size()); } else { - skipPayloadField(unpacker, fieldId, stringTable); + assertEquals(1, payloadAttributes.size()); + assertEquals(processTags.toString(), payloadAttributes.get(PROCESS_TAGS)); } } - - assertEquals(EXPECTED_PAYLOAD_FIELD_IDS.size(), payloadFieldCount); - assertEquals(EXPECTED_PAYLOAD_FIELD_IDS, payloadFieldsSeen); - assertEquals(1, chunkCount); - assertNotNull(payloadAttributes); - CharSequence processTags = ProcessTags.getTagsForSerialization(); - if (processTags == null) { - assertEquals(0, payloadAttributes.size()); - } else { - assertEquals(1, payloadAttributes.size()); - assertEquals(processTags.toString(), payloadAttributes.get(PROCESS_TAGS)); - } } // expectedSamplingMechanism 0 is SamplingMechanism.DEFAULT. @@ -249,39 +250,42 @@ void spanIdsAreEncodedAsUnsignedValues() throws IOException { long parentId = Long.MIN_VALUE + 456L; byte[] encoded = serializeV1Payload(spanWithIds(spanId, parentId)); - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded); List stringTable = newStringTable(); Long actualSpanId = null; Long actualParentId = null; - int payloadFieldCount = unpacker.unpackMapHeader(); - for (int i = 0; i < payloadFieldCount; i++) { - int payloadFieldId = unpacker.unpackInt(); - if (payloadFieldId != PayloadField.CHUNKS) { - skipPayloadField(unpacker, payloadFieldId, stringTable); - continue; - } - assertEquals(1, unpacker.unpackArrayHeader()); - int chunkFieldCount = unpacker.unpackMapHeader(); - for (int j = 0; j < chunkFieldCount; j++) { - int chunkFieldId = unpacker.unpackInt(); - if (chunkFieldId != ChunkField.SPANS) { - skipChunkField(unpacker, chunkFieldId, stringTable); + try (MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(encoded)) { + int payloadFieldCount = unpacker.unpackMapHeader(); + for (int i = 0; i < payloadFieldCount; i++) { + int payloadFieldId = unpacker.unpackInt(); + if (payloadFieldId != PayloadField.CHUNKS) { + skipPayloadField(unpacker, payloadFieldId, stringTable); continue; } assertEquals(1, unpacker.unpackArrayHeader()); - int spanFieldCount = unpacker.unpackMapHeader(); - for (int k = 0; k < spanFieldCount; k++) { - int spanFieldId = unpacker.unpackInt(); - switch (spanFieldId) { - case SpanField.SPAN_ID: - actualSpanId = unpackUint64(unpacker); - break; - case SpanField.PARENT_ID: - actualParentId = unpackUint64(unpacker); - break; - default: - skipSpanField(unpacker, spanFieldId, stringTable); + + int chunkFieldCount = unpacker.unpackMapHeader(); + for (int j = 0; j < chunkFieldCount; j++) { + int chunkFieldId = unpacker.unpackInt(); + if (chunkFieldId != ChunkField.SPANS) { + skipChunkField(unpacker, chunkFieldId, stringTable); + continue; + } + assertEquals(1, unpacker.unpackArrayHeader()); + + int spanFieldCount = unpacker.unpackMapHeader(); + for (int k = 0; k < spanFieldCount; k++) { + int spanFieldId = unpacker.unpackInt(); + switch (spanFieldId) { + case SpanField.SPAN_ID: + actualSpanId = unpackUint64(unpacker); + break; + case SpanField.PARENT_ID: + actualParentId = unpackUint64(unpacker); + break; + default: + skipSpanField(unpacker, spanFieldId, stringTable); + } } } } @@ -406,21 +410,22 @@ void metaStructIsEncodedAsBytesAttribute() throws IOException { byte[] metaStructBytes = (byte[]) attributes.get("meta_key"); assertNotNull(metaStructBytes); - MessageUnpacker metaStructUnpacker = MessagePack.newDefaultUnpacker(metaStructBytes); - int metaStructFieldCount = metaStructUnpacker.unpackMapHeader(); Map decodedMetaStruct = new HashMap<>(); - for (int i = 0; i < metaStructFieldCount; i++) { - String key = metaStructUnpacker.unpackString(); - ValueType valueType = metaStructUnpacker.getNextFormat().getValueType(); - switch (valueType) { - case INTEGER: - decodedMetaStruct.put(key, metaStructUnpacker.unpackLong()); - break; - case STRING: - decodedMetaStruct.put(key, metaStructUnpacker.unpackString()); - break; - default: - fail("Unexpected meta_struct value type for key " + key); + try (MessageUnpacker metaStructUnpacker = MessagePack.newDefaultUnpacker(metaStructBytes)) { + int metaStructFieldCount = metaStructUnpacker.unpackMapHeader(); + for (int i = 0; i < metaStructFieldCount; i++) { + String key = metaStructUnpacker.unpackString(); + ValueType valueType = metaStructUnpacker.getNextFormat().getValueType(); + switch (valueType) { + case INTEGER: + decodedMetaStruct.put(key, metaStructUnpacker.unpackLong()); + break; + case STRING: + decodedMetaStruct.put(key, metaStructUnpacker.unpackString()); + break; + default: + fail("Unexpected meta_struct value type for key " + key); + } } } @@ -528,32 +533,33 @@ public void accept(int messageCount, ByteBuffer buffer) { try { Payload payload = mapper.newPayload().withBody(messageCount, buffer); payload.writeTo(channel); - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(channel.flipForReading()); - if (messageCount == 0) { - assertEquals(0, unpacker.unpackMapHeader()); - return; - } + try (MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(channel.flipForReading())) { + if (messageCount == 0) { + assertEquals(0, unpacker.unpackMapHeader()); + return; + } - List stringTable = newStringTable(); - int payloadFieldCount = unpacker.unpackMapHeader(); - assertEquals(EXPECTED_PAYLOAD_FIELD_IDS.size(), payloadFieldCount); - - boolean seenChunks = false; - for (int i = 0; i < payloadFieldCount; i++) { - int fieldId = unpacker.unpackInt(); - if (fieldId == PayloadField.CHUNKS) { - int traceCount = unpacker.unpackArrayHeader(); - assertEquals(messageCount, traceCount); - seenChunks = true; - for (int traceIndex = 0; traceIndex < traceCount; traceIndex++) { - verifyChunk(unpacker, expectedTraces.get(position++), stringTable); + List stringTable = newStringTable(); + int payloadFieldCount = unpacker.unpackMapHeader(); + assertEquals(EXPECTED_PAYLOAD_FIELD_IDS.size(), payloadFieldCount); + + boolean seenChunks = false; + for (int i = 0; i < payloadFieldCount; i++) { + int fieldId = unpacker.unpackInt(); + if (fieldId == PayloadField.CHUNKS) { + int traceCount = unpacker.unpackArrayHeader(); + assertEquals(messageCount, traceCount); + seenChunks = true; + for (int traceIndex = 0; traceIndex < traceCount; traceIndex++) { + verifyChunk(unpacker, expectedTraces.get(position++), stringTable); + } + } else { + skipPayloadField(unpacker, fieldId, stringTable); } - } else { - skipPayloadField(unpacker, fieldId, stringTable); } - } - assertTrue(seenChunks); + assertTrue(seenChunks); + } } catch (IOException e) { fail(e.getMessage()); } finally { @@ -725,11 +731,10 @@ private static void verifySpan( assertEqualsWithNullAsEmpty(expectedSpan.getType(), type); assertEquals(0, linksCount); assertEquals(0, eventsCount); - assertEqualsWithNullAsEmpty(expectedSpan.getTag(ENV), env); - assertEqualsWithNullAsEmpty(expectedSpan.getTag(VERSION), version); - assertEqualsWithNullAsEmpty(expectedSpan.getTag(COMPONENT), component); - assertEquals( - TraceMapperV1.getSpanKindValue(expectedSpan.getTag(SPAN_KIND)), spanKind); + assertEqualsWithNullAsEmpty(expectedSpan.getTag(ENV), env); + assertEqualsWithNullAsEmpty(expectedSpan.getTag(VERSION), version); + assertEqualsWithNullAsEmpty(expectedSpan.getTag(COMPONENT), component); + assertEquals(TraceMapperV1.getSpanKindValue(expectedSpan.getTag(SPAN_KIND)), spanKind); assertNotNull(attributes); int expectedHttpStatusCode = expectedSpan.getHttpStatusCode(); @@ -781,7 +786,7 @@ private static void addFlattenedExpectedAttribute( } for (Map.Entry entry : ((Map) value).entrySet()) { addFlattenedExpectedAttribute( - expectedAttributes, key + "." + String.valueOf(entry.getKey()), entry.getValue()); + expectedAttributes, key + "." + entry.getKey(), entry.getValue()); } } @@ -849,7 +854,7 @@ private static PojoSpan spanWithLinks(List spanLinks) { return span(123L, 0L, emptyMap(), emptyMap(), 200, spanLinks); } - /** A span carrying the fields shared by every test; only the varying pieces are parameters. */ + /** A span carrying the fields shared by most tests; only the varying pieces are parameters. */ private static PojoSpan span( long spanId, long parentId, From d6bf61ba7bbf9de2c9f22d5093d692e2bceac659 Mon Sep 17 00:00:00 2001 From: Alexey Kuznetsov Date: Fri, 31 Jul 2026 16:11:49 -0400 Subject: [PATCH 3/3] Pin decoded span field ids in the v1 payload test verifySpan asserted a 16-entry span map but not that each field id appeared exactly once, so a repeated id plus an omitted one left the omitted field's decoded value at its initialiser. For parentId (0) and error (false) those sentinels equal the expected values, and TraceGenerator always uses DDSpanId.ZERO as the parent id, so the parent-id check was inert. Assert the decoded field-id set instead, which covers all 16 fields and names the missing id on failure. Co-Authored-By: Claude Opus 5 (1M context) --- .../ddagent/TraceMapperV1PayloadTest.java | 30 ++++++++++++++++++- 1 file changed, 29 insertions(+), 1 deletion(-) diff --git a/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.java b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.java index 5676bc88716..1eacd903b96 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.java @@ -98,6 +98,27 @@ class TraceMapperV1PayloadTest { PayloadField.ATTRIBUTES, PayloadField.CHUNKS)); + /** Every field a v1 span carries, mirroring {@code TraceMapperV1.encodeSpans}. */ + private static final Set EXPECTED_SPAN_FIELD_IDS = + new HashSet<>( + asList( + SpanField.SERVICE, + SpanField.NAME, + SpanField.RESOURCE, + SpanField.SPAN_ID, + SpanField.PARENT_ID, + SpanField.START, + SpanField.DURATION, + SpanField.ERROR, + SpanField.ATTRIBUTES, + SpanField.TYPE, + SpanField.LINKS, + SpanField.EVENTS, + SpanField.ENV, + SpanField.VERSION, + SpanField.COMPONENT, + SpanField.KIND)); + // Keep the ProcessTags static in sync with the (per-test rebuilt) Config, the way DDSpecification // did for the original Spock tests. Runs after WithConfigExtension has rebuilt Config. @BeforeEach @@ -645,8 +666,9 @@ private static void verifySpan( MessageUnpacker unpacker, PojoSpan expectedSpan, List stringTable) throws IOException { int spanFieldCount = unpacker.unpackMapHeader(); - assertEquals(16, spanFieldCount); + assertEquals(EXPECTED_SPAN_FIELD_IDS.size(), spanFieldCount); + Set spanFieldsSeen = new HashSet<>(); String service = null; String name = null; String resource = null; @@ -666,6 +688,7 @@ private static void verifySpan( for (int i = 0; i < spanFieldCount; i++) { int fieldId = unpacker.unpackInt(); + spanFieldsSeen.add(fieldId); switch (fieldId) { case SpanField.SERVICE: service = readStreamingString(unpacker, stringTable); @@ -720,6 +743,11 @@ private static void verifySpan( } } + // A 16-entry map could still repeat one field id and omit another, which would leave a decoded + // value at its initial sentinel. Pinning the id set makes each field present exactly once, so + // the value assertions below cannot pass on an unwritten field (e.g. parentId 0, error false). + assertEquals(EXPECTED_SPAN_FIELD_IDS, spanFieldsSeen); + assertEqualsWithNullAsEmpty(expectedSpan.getServiceName(), service); assertEqualsWithNullAsEmpty(expectedSpan.getOperationName(), name); assertEqualsWithNullAsEmpty(expectedSpan.getResourceName(), resource);