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..1eacd903b96 --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV1PayloadTest.java @@ -0,0 +1,979 @@ +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)); + + /** 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 + 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); + List stringTable = newStringTable(); + + 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); + } + } + + 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)); + List stringTable = newStringTable(); + + Long actualSpanId = null; + Long actualParentId = null; + 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 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); + + Map decodedMetaStruct = new HashMap<>(); + 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); + } + } + } + + 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); + 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); + } + } 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(EXPECTED_SPAN_FIELD_IDS.size(), spanFieldCount); + + Set spanFieldsSeen = new HashSet<>(); + 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(); + spanFieldsSeen.add(fieldId); + 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); + } + } + + // 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); + 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 + "." + 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 most tests; 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;