diff --git a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/ddagent/TraceMapperV04PayloadTest.groovy b/dd-trace-core/src/test/groovy/datadog/trace/common/writer/ddagent/TraceMapperV04PayloadTest.groovy deleted file mode 100644 index 650d846a360..00000000000 --- a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/ddagent/TraceMapperV04PayloadTest.groovy +++ /dev/null @@ -1,425 +0,0 @@ -package datadog.trace.common.writer.ddagent - -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.DD64bTraceId -import datadog.trace.api.DDTags -import datadog.trace.api.DDTraceId -import datadog.trace.api.ProcessTags -import datadog.trace.bootstrap.instrumentation.api.Tags -import datadog.trace.common.writer.Payload -import datadog.trace.common.writer.TraceGenerator -import datadog.trace.core.DDSpanContext -import datadog.trace.test.util.DDSpecification -import org.junit.jupiter.api.Assertions -import org.msgpack.core.MessageFormat -import org.msgpack.core.MessagePack -import org.msgpack.core.MessageUnpacker - -import java.nio.ByteBuffer -import java.nio.channels.WritableByteChannel - -import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.DD_MEASURED -import static datadog.trace.common.writer.TraceGenerator.generateRandomTraces -import static org.junit.jupiter.api.Assertions.assertEquals -import static org.junit.jupiter.api.Assertions.assertFalse -import static org.junit.jupiter.api.Assertions.assertNotNull -import static org.junit.jupiter.api.Assertions.assertTrue -import static org.msgpack.core.MessageFormat.FLOAT32 -import static org.msgpack.core.MessageFormat.FLOAT64 -import static org.msgpack.core.MessageFormat.INT16 -import static org.msgpack.core.MessageFormat.INT32 -import static org.msgpack.core.MessageFormat.INT64 -import static org.msgpack.core.MessageFormat.INT8 -import static org.msgpack.core.MessageFormat.NEGFIXINT -import static org.msgpack.core.MessageFormat.POSFIXINT -import static org.msgpack.core.MessageFormat.UINT16 -import static org.msgpack.core.MessageFormat.UINT32 -import static org.msgpack.core.MessageFormat.UINT64 -import static org.msgpack.core.MessageFormat.UINT8 - -class TraceMapperV04PayloadTest extends DDSpecification { - - def "test traces written correctly"() { - setup: - List> traces = generateRandomTraces(traceCount, lowCardinality) - TraceMapperV0_4 traceMapper = new TraceMapperV0_4() - 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 - // in the real like the mapper is always reset each trace. - // here we need to force it when we fail since the buffer will be reset as well - traceMapper.reset() - } - } - packer.flush() - - then: - if (tracesFitInBuffer) { - verifier.verifyTracesConsumed() - } - - where: - bufferSize | traceCount | lowCardinality - 20 << 10 | 0 | true - 20 << 10 | 1 | true - 30 << 10 | 1 | true - 30 << 10 | 2 | true - 20 << 10 | 0 | false - 20 << 10 | 1 | false - 30 << 10 | 1 | false - 30 << 10 | 2 | false - 100 << 10 | 0 | true - 100 << 10 | 1 | true - 100 << 10 | 10 | true - 100 << 10 | 100 | true - 100 << 10 | 1000 | true - 100 << 10 | 0 | false - 100 << 10 | 1 | false - 100 << 10 | 10 | false - 100 << 10 | 100 | false - 100 << 10 | 1000 | false - } - - def "test full 64-bit trace and span identifiers"() { - setup: - def span = new TraceGenerator.PojoSpan( - "service", - "operation", - "resource", - traceId, - spanId, - parentId, - 123L, - 456L, - 0, - [:], - [:], - "type", - false, - 0, - 0, - "origin") - def traces = [[span]] - TraceMapperV0_4 traceMapper = new TraceMapperV0_4() - PayloadVerifier verifier = new PayloadVerifier(traces, traceMapper) - MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(20 << 10, verifier)) - - when: - packer.format([span], traceMapper) - packer.flush() - - then: - verifier.verifyTracesConsumed() - - where: - traceId | spanId | parentId - DD64bTraceId.ONE | 2L | 3L - DD64bTraceId.MAX | 2L | 3L - DD64bTraceId.from(-10) | -11L | -12L - } - - void 'test metaStruct support'() { - given: - def span = new TraceGenerator.PojoSpan( - 'service', - 'operation', - 'resource', - DDTraceId.ONE, - 1L, - -1L, - 123L, - 456L, - 0, - [:], - [:], - 'type', - false, - 0, - 0, - 'origin') - span.setMetaStruct('stack', Thread.currentThread().stackTrace.toList().collect { - [ - file: it.fileName ?: '', - class_name: it.className ?: '', - function: it.methodName ?: '' - ] - }) - def traces = [[span]] - TraceMapperV0_4 traceMapper = new TraceMapperV0_4() - PayloadVerifier verifier = new PayloadVerifier(traces, traceMapper, (List expected, byte[] received) -> { - def unpacker = MessagePack.newDefaultUnpacker(received) - def size = unpacker.unpackArrayHeader() - assertEquals(expected.size(), size) - expected.eachWithIndex { - def stackEntry, int i -> - int fields = unpacker.unpackMapHeader() - (0..> expectedTraces - private final TraceMapperV0_4 mapper - private ByteBuffer captured = ByteBuffer.allocate(200 << 10) - private MetaStructVerifier metaStructVerifier - - private int position = 0 - - private PayloadVerifier(List> traces, TraceMapperV0_4 mapper, MetaStructVerifier metaStructVerifier = null) { - this.expectedTraces = traces - this.mapper = mapper - this.metaStructVerifier = metaStructVerifier - } - - void skipLargeTrace() { - ++position - } - - @Override - void accept(int messageCount, ByteBuffer buffer) { - if (expectedTraces.isEmpty() && messageCount == 0) { - return - } - int processTagsCount = 0 - try { - Payload payload = mapper.newPayload().withBody(messageCount, buffer) - payload.writeTo(this) - captured.flip() - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(captured) - int traceCount = unpacker.unpackArrayHeader() - for (int i = 0; i < traceCount; ++i) { - List expectedTrace = expectedTraces.get(position++) - int spanCount = unpacker.unpackArrayHeader() - assertEquals(expectedTrace.size(), spanCount) - for (int k = 0; k < spanCount; ++k) { - TraceGenerator.PojoSpan expectedSpan = expectedTrace.get(k) - int elementCount = unpacker.unpackMapHeader() - boolean hasMetaStruct = !expectedSpan.getMetaStruct().isEmpty() - assertEquals(hasMetaStruct ? 13 : 12, elementCount) - assertEquals("service", unpacker.unpackString()) - String serviceName = unpacker.unpackString() - assertEqualsWithNullAsEmpty(expectedSpan.getServiceName(), serviceName) - assertEquals("name", unpacker.unpackString()) - String operationName = unpacker.unpackString() - assertEqualsWithNullAsEmpty(expectedSpan.getOperationName(), operationName) - assertEquals("resource", unpacker.unpackString()) - String resourceName = unpacker.unpackString() - assertEqualsWithNullAsEmpty(expectedSpan.getResourceName(), resourceName) - assertEquals("trace_id", unpacker.unpackString()) - long traceId = unpacker.unpackValue().asNumberValue().toLong() - assertEquals(expectedSpan.getTraceId().toLong(), traceId) - assertEquals("span_id", unpacker.unpackString()) - long spanId = unpacker.unpackValue().asNumberValue().toLong() - assertEquals(expectedSpan.getSpanId(), spanId) - assertEquals("parent_id", unpacker.unpackString()) - long parentId = unpacker.unpackValue().asNumberValue().toLong() - assertEquals(expectedSpan.getParentId(), parentId) - assertEquals("start", unpacker.unpackString()) - long startTime = unpacker.unpackLong() - assertEquals(expectedSpan.getStartTime(), startTime) - assertEquals("duration", unpacker.unpackString()) - long duration = unpacker.unpackLong() - assertEquals(expectedSpan.getDurationNano(), duration) - assertEquals("type", unpacker.unpackString()) - String type = unpacker.unpackString() - assertEquals(expectedSpan.getType(), type) - assertEquals("error", unpacker.unpackString()) - int error = unpacker.unpackInt() - assertEquals(expectedSpan.getError(), error) - assertEquals("metrics", unpacker.unpackString()) - int metricsSize = unpacker.unpackMapHeader() - HashMap metrics = new HashMap<>() - for (int j = 0; j < metricsSize; ++j) { - String key = unpacker.unpackString() - Number n = null - MessageFormat format = unpacker.getNextFormat() - switch (format) { - case NEGFIXINT: - case POSFIXINT: - case INT8: - case UINT8: - case INT16: - case UINT16: - case INT32: - case UINT32: - n = unpacker.unpackInt() - break - case INT64: - case UINT64: - n = unpacker.unpackLong() - break - case FLOAT32: - n = unpacker.unpackFloat() - break - case FLOAT64: - n = unpacker.unpackDouble() - break - default: - Assertions.fail("Unexpected type in metrics values: " + format) - } - if (DD_MEASURED.toString() == key) { - assert ((n == 1) && expectedSpan.isMeasured()) || !expectedSpan.isMeasured() - } else if (DDSpanContext.PRIORITY_SAMPLING_KEY == key) { - //check that priority sampling is only on first and last span - if (k == 0 || k == spanCount - 1) { - assertEquals(expectedSpan.samplingPriority(), n.intValue()) - } else { - assertFalse(expectedSpan.hasSamplingPriority()) - } - } else { - metrics.put(key, n) - } - } - for (Map.Entry metric : metrics.entrySet()) { - if (metric.getValue() instanceof Double || metric.getValue() instanceof Float) { - assertEquals(((Number) expectedSpan.getTag(metric.getKey())).doubleValue(), metric.getValue().doubleValue(), 0.001) - } else { - assertEquals(expectedSpan.getTag(metric.getKey()), metric.getValue()) - } - } - assertEquals("meta", unpacker.unpackString()) - int metaSize = unpacker.unpackMapHeader() - HashMap meta = new HashMap<>() - for (int j = 0; j < metaSize; ++j) { - meta.put(unpacker.unpackString(), unpacker.unpackString()) - } - for (Map.Entry entry : meta.entrySet()) { - if (Tags.HTTP_STATUS.equals(entry.getKey())) { - assertEquals(String.valueOf(expectedSpan.getHttpStatusCode()), entry.getValue()) - } else if (DDTags.ORIGIN_KEY.equals(entry.getKey())) { - assertEquals(expectedSpan.getOrigin(), entry.getValue()) - } else if (DDTags.PROCESS_TAGS.equals(entry.getKey())) { - assertTrue(Config.get().isExperimentalPropagateProcessTagsEnabled()) - assertEquals(0, k) - assertEquals(ProcessTags.tagsForSerialization.toString(), entry.getValue()) - processTagsCount++ - } else { - Object tag = expectedSpan.getTag(entry.getKey()) - if (null != tag) { - assertEquals(String.valueOf(tag), entry.getValue()) - } else { - assertEquals(expectedSpan.getBaggage().get(entry.getKey()), entry.getValue()) - } - } - } - if (hasMetaStruct) { - Map metaStruct = expectedSpan.getMetaStruct() - assertEquals("meta_struct", unpacker.unpackString()) - int metaStructSize = unpacker.unpackMapHeader() - for (int j = 0; j < metaStructSize; ++j) { - String field = unpacker.unpackString() - if (metaStructVerifier != null) { - byte[] binary = new byte[unpacker.unpackBinaryHeader()] - unpacker.readPayload(binary) - metaStructVerifier.verify(metaStruct.get(field), binary) - } - } - } - } - } - } catch (IOException e) { - Assertions.fail(e.getMessage()) - } finally { - mapper.reset() - captured.position(0) - captured.limit(captured.capacity()) - assert processTagsCount == (Config.get().isExperimentalPropagateProcessTagsEnabled() ? 1 : 0) - } - } - - @Override - int write(ByteBuffer src) { - if (captured.remaining() < src.remaining()) { - ByteBuffer newBuffer = ByteBuffer.allocate(captured.capacity() + src.capacity()) - 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 assertEqualsWithNullAsEmpty(CharSequence expected, CharSequence actual) { - if (null == expected) { - assertEquals("", actual) - } else { - assertEquals(expected.toString(), actual.toString()) - } - } - - private static interface MetaStructVerifier { - void verify(final E expected, final byte[] received) - } -} diff --git a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/ddagent/TraceMapperV05PayloadTest.groovy b/dd-trace-core/src/test/groovy/datadog/trace/common/writer/ddagent/TraceMapperV05PayloadTest.groovy deleted file mode 100644 index e03bf6e48eb..00000000000 --- a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/ddagent/TraceMapperV05PayloadTest.groovy +++ /dev/null @@ -1,389 +0,0 @@ -package datadog.trace.common.writer.ddagent - -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.DDTags -import datadog.trace.api.DDTraceId -import datadog.trace.api.ProcessTags -import datadog.trace.api.sampling.PrioritySampling -import datadog.trace.bootstrap.instrumentation.api.Tags -import datadog.trace.common.writer.Payload -import datadog.trace.common.writer.TraceGenerator -import datadog.trace.core.DDSpanContext -import datadog.trace.test.util.DDSpecification -import org.junit.Assert -import org.msgpack.core.MessageFormat -import org.msgpack.core.MessagePack -import org.msgpack.core.MessageUnpacker - -import java.nio.ByteBuffer -import java.nio.channels.WritableByteChannel -import java.util.concurrent.atomic.AtomicInteger - -import static datadog.trace.api.config.GeneralConfig.EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED -import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.DD_MEASURED -import static datadog.trace.common.writer.TraceGenerator.generateRandomTraces -import static org.junit.jupiter.api.Assertions.assertEquals -import static org.junit.jupiter.api.Assertions.assertFalse -import static org.junit.jupiter.api.Assertions.assertNotNull -import static org.junit.jupiter.api.Assertions.assertTrue -import static org.msgpack.core.MessageFormat.FLOAT32 -import static org.msgpack.core.MessageFormat.FLOAT64 -import static org.msgpack.core.MessageFormat.INT16 -import static org.msgpack.core.MessageFormat.INT32 -import static org.msgpack.core.MessageFormat.INT64 -import static org.msgpack.core.MessageFormat.INT8 -import static org.msgpack.core.MessageFormat.NEGFIXINT -import static org.msgpack.core.MessageFormat.POSFIXINT -import static org.msgpack.core.MessageFormat.UINT16 -import static org.msgpack.core.MessageFormat.UINT32 -import static org.msgpack.core.MessageFormat.UINT64 -import static org.msgpack.core.MessageFormat.UINT8 - -class TraceMapperV05PayloadTest extends DDSpecification { - - def "body overflow causes a flush"() { - setup: - // disable process tags since they are only on the first span of the chunk otherwise the calculation woes - injectSysConfig(EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED, "false") - ProcessTags.reset() - // 4x 36 ASCII characters and 2 bytes of msgpack string prefix - int dictionarySpacePerTrace = 4 * (36 + 2) - // enough space for two traces with distinct string values, plus the header - int dictionarySize = dictionarySpacePerTrace * 2 + 5 - TraceMapperV0_5 traceMapper = new TraceMapperV0_5(dictionarySize) - List repeatedTrace = Collections.singletonList(new TraceGenerator.PojoSpan( - UUID.randomUUID().toString(), - UUID.randomUUID().toString(), - UUID.randomUUID().toString(), - DDTraceId.ZERO, - DDSpanId.ZERO, - DDSpanId.ZERO, - 10000, - 100, - 0, - Collections.emptyMap(), - Collections.emptyMap(), - UUID.randomUUID().toString(), - false, - PrioritySampling.UNSET, - 0, - null)) - int traceSize = calculateSize(repeatedTrace) - // 30KB body - int bufferSize = 30 << 10 - int tracesRequiredToOverflowBody = Math.ceil(((double)bufferSize) / traceSize) + 1 - List> traces = new ArrayList<>(tracesRequiredToOverflowBody) - for (int i = 0; i < tracesRequiredToOverflowBody; ++i) { - traces.add(repeatedTrace) - } - // the last one won't be flushed - List> flushedTraces = new ArrayList<>(traces) - flushedTraces.remove(traces.size() - 1) - // need space for the overflowing buffer, the dictionary, and two small array headers - PayloadVerifier verifier = new PayloadVerifier(flushedTraces, traceMapper, bufferSize + dictionarySize + 1 + 1 + 5) - MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(bufferSize, verifier)) - when: - for (List trace : traces) { - packer.format(trace, traceMapper) - } - then: - verifier.verifyTracesConsumed() - cleanup: - injectSysConfig(EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED, "true") - ProcessTags.reset() - } - - def "test dictionary compressed traces written correctly"() { - setup: - List> traces = generateRandomTraces(traceCount, lowCardinality) - TraceMapperV0_5 traceMapper = new TraceMapperV0_5(dictionarySize) - 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 - } - } - packer.flush() - - then: - if (tracesFitInBuffer) { - verifier.verifyTracesConsumed() - } - - where: - bufferSize | dictionarySize | traceCount | lowCardinality - 10 << 10 | 10 << 10 | 0 | true - 10 << 10 | 10 << 10 | 1 | true - 10 << 10 | 10 << 10 | 10 | true - 10 << 10 | 10 << 10 | 100 | true - 10 << 10 | 100 << 10 | 1 | true - 10 << 10 | 100 << 10 | 10 | true - 10 << 10 | 100 << 10 | 100 | true - 10 << 10 | 10 << 10 | 0 | false - 10 << 10 | 10 << 10 | 1 | false - 10 << 10 | 10 << 10 | 10 | false - 10 << 10 | 10 << 10 | 100 | false - 10 << 10 | 100 << 10 | 1 | false - 10 << 10 | 100 << 10 | 10 | false - 10 << 10 | 100 << 10 | 100 | false - 100 << 10 | 10 << 10 | 0 | true - 100 << 10 | 10 << 10 | 1 | true - 100 << 10 | 10 << 10 | 10 | true - 100 << 10 | 10 << 10 | 100 | true - 100 << 10 | 100 << 10 | 1 | true - 100 << 10 | 100 << 10 | 10 | true - 100 << 10 | 100 << 10 | 100 | true - 100 << 10 | 10 << 10 | 0 | false - 100 << 10 | 10 << 10 | 1 | false - 100 << 10 | 10 << 10 | 10 | false - 100 << 10 | 10 << 10 | 100 | false - 100 << 10 | 100 << 10 | 1 | false - 100 << 10 | 100 << 10 | 10 | false - 100 << 10 | 100 << 10 | 100 | false - 100 << 10 | 100 << 10 | 1000 | false - } - - void 'test process tags serialization'() { - setup: - assertNotNull(ProcessTags.tagsForSerialization) - def spans = (1..2).collect { - new TraceGenerator.PojoSpan( - 'service', - 'operation', - 'resource', - DDTraceId.ONE, - it, - -1L, - 123L, - 456L, - 0, - [:], - [:], - 'type', - false, - 0, - 0, - 'origin') - } - - def traces = [spans] - TraceMapperV0_5 traceMapper = new TraceMapperV0_5() - PayloadVerifier verifier = new PayloadVerifier(traces, traceMapper) - MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(20 << 10, verifier)) - - when: - packer.format(spans, traceMapper) - packer.flush() - - then: - verifier.verifyTracesConsumed() - } - - private static final class PayloadVerifier implements ByteBufferConsumer, WritableByteChannel { - - private final List> expectedTraces - private final TraceMapperV0_5 mapper - private ByteBuffer captured - - private int position = 0 - - private PayloadVerifier(List> traces, TraceMapperV0_5 mapper) { - this (traces, mapper, 200 << 10) - } - - private PayloadVerifier(List> traces, TraceMapperV0_5 mapper, int size) { - this.expectedTraces = traces - this.mapper = mapper - this.captured = ByteBuffer.allocate(size) - } - - void skipLargeTrace() { - ++position - } - - @Override - void accept(int messageCount, ByteBuffer buffer) { - def processTagsCount = 0 - try { - Payload payload = mapper.newPayload().withBody(messageCount, buffer) - payload.writeTo(this) - captured.flip() - MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(captured) - int header = unpacker.unpackArrayHeader() - assertEquals(2, header) - int dictionarySize = unpacker.unpackArrayHeader() - String[] dictionary = new String[dictionarySize] - for (int i = 0; i < dictionary.length; ++i) { - dictionary[i] = unpacker.unpackString() - } - int traceCount = unpacker.unpackArrayHeader() - for (int i = 0; i < traceCount; ++i) { - List expectedTrace = expectedTraces.get(position++) - int spanCount = unpacker.unpackArrayHeader() - assertEquals(expectedTrace.size(), spanCount) - for (int k = 0; k < spanCount; ++k) { - TraceGenerator.PojoSpan expectedSpan = expectedTrace.get(k) - int elementCount = unpacker.unpackArrayHeader() - assertEquals(12, elementCount) - String serviceName = dictionary[unpacker.unpackInt()] - assertEqualsWithNullAsEmpty(expectedSpan.getServiceName(), serviceName) - String operationName = dictionary[unpacker.unpackInt()] - assertEqualsWithNullAsEmpty(expectedSpan.getOperationName(), operationName) - String resourceName = dictionary[unpacker.unpackInt()] - assertEqualsWithNullAsEmpty(expectedSpan.getResourceName(), resourceName) - long traceId = unpacker.unpackValue().asNumberValue().toLong() - assertEquals(expectedSpan.getTraceId().toLong(), traceId) - long spanId = unpacker.unpackValue().asNumberValue().toLong() - assertEquals(expectedSpan.getSpanId(), spanId) - long parentId = unpacker.unpackValue().asNumberValue().toLong() - assertEquals(expectedSpan.getParentId(), parentId) - long startTime = unpacker.unpackLong() - assertEquals(expectedSpan.getStartTime(), startTime) - long duration = unpacker.unpackLong() - assertEquals(expectedSpan.getDurationNano(), duration) - int error = unpacker.unpackInt() - assertEquals(expectedSpan.getError(), error) - int metaSize = unpacker.unpackMapHeader() - HashMap meta = new HashMap<>() - for (int j = 0; j < metaSize; ++j) { - meta.put(dictionary[unpacker.unpackInt()], dictionary[unpacker.unpackInt()]) - } - for (Map.Entry entry : meta.entrySet()) { - if (Tags.HTTP_STATUS.equals(entry.getKey())) { - assertEquals(String.valueOf(expectedSpan.getHttpStatusCode()), entry.getValue()) - } else if(DDTags.ORIGIN_KEY.equals(entry.getKey())) { - assertEquals(expectedSpan.getOrigin(), entry.getValue()) - } else if (DDTags.PROCESS_TAGS.equals(entry.getKey())) { - processTagsCount++ - assertTrue(Config.get().isExperimentalPropagateProcessTagsEnabled()) - assertEquals(0, k) - assertEquals(ProcessTags.tagsForSerialization.toString(), entry.getValue()) - } else { - Object tag = expectedSpan.getTag(entry.getKey()) - if (null != tag) { - assertEquals(String.valueOf(tag), entry.getValue()) - } else { - assertEquals(expectedSpan.getBaggage().get(entry.getKey()), entry.getValue()) - } - } - } - int metricsSize = unpacker.unpackMapHeader() - HashMap metrics = new HashMap<>() - for (int j = 0; j < metricsSize; ++j) { - String key = dictionary[unpacker.unpackInt()] - Number n = null - MessageFormat format = unpacker.getNextFormat() - switch (format) { - case NEGFIXINT: - case POSFIXINT: - case INT8: - case UINT8: - case INT16: - case UINT16: - case INT32: - case UINT32: - n = unpacker.unpackInt() - break - case INT64: - case UINT64: - n = unpacker.unpackLong() - break - case FLOAT32: - n = unpacker.unpackFloat() - break - case FLOAT64: - n = unpacker.unpackDouble() - break - default: - Assert.fail("Unexpected type in metrics values: " + format + " for key " + key) - } - if (DD_MEASURED.toString() == key) { - assert ((n == 1) && expectedSpan.isMeasured()) || !expectedSpan.isMeasured() - } else if (DDSpanContext.PRIORITY_SAMPLING_KEY == key) { - //check that priority sampling is only on first and last span - if (k == 0 || k == spanCount -1) { - assertEquals(expectedSpan.samplingPriority(), n.intValue()) - } else { - assertFalse(expectedSpan.hasSamplingPriority()) - } - } else { - metrics.put(key, n) - } - } - for (Map.Entry metric : metrics.entrySet()) { - if (metric.getValue() instanceof Double || metric.getValue() instanceof Float) { - assertEquals(((Number)expectedSpan.getTag(metric.getKey())).doubleValue(), metric.getValue().doubleValue(), 0.001, metric.getKey()) - } else { - assertEquals(expectedSpan.getTag(metric.getKey()), metric.getValue(), metric.getKey()) - } - } - String type = dictionary[unpacker.unpackInt()] - assertEquals(expectedSpan.getType(), type) - } - } - } catch (IOException e) { - Assert.fail(e.getMessage()) - } finally { - assert processTagsCount == (Config.get().isExperimentalPropagateProcessTagsEnabled() ? 1 : 0) - 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 assertEqualsWithNullAsEmpty(CharSequence expected, CharSequence actual) { - if (null == expected) { - assertEquals("", actual) - } else { - assertEquals(expected.toString(), actual.toString()) - } - } - - static int calculateSize(List trace) { - AtomicInteger size = new AtomicInteger() - def packer = new MsgPackWriter(new FlushingBuffer(1024, new ByteBufferConsumer() { - @Override - void accept(int messageCount, ByteBuffer buffer) { - size.set(buffer.limit() - buffer.position()) - } - })) - packer.format(trace, new TraceMapperV0_5(1024)) - packer.flush() - return size.get() - } -} diff --git a/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/PayloadVerifiers.java b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/PayloadVerifiers.java new file mode 100644 index 00000000000..4f2604a8d6b --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/PayloadVerifiers.java @@ -0,0 +1,100 @@ +package datadog.trace.common.writer.ddagent; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.fail; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.WritableByteChannel; +import org.msgpack.core.MessageFormat; +import org.msgpack.core.MessageUnpacker; + +/** + * Shared decode helpers for the TraceMapper payload tests. The V0.4 and V0.5 payloads use different + * wire formats (V0.4 = inline strings, V0.5 = dictionary-compressed string indices), so the span + * decoding lives in each test's own {@code PayloadVerifier}; only the format-agnostic pieces live + * here. + */ +final class PayloadVerifiers { + private PayloadVerifiers() {} + + /** A serialized empty string round-trips as "", so a null expected value matches "". */ + static void assertEqualsWithNullAsEmpty(CharSequence expected, CharSequence actual) { + if (expected == null) { + assertEquals("", actual); + } else { + assertEquals(expected.toString(), actual.toString()); + } + } + + /** Unpacks a msgpack numeric value, matching the encoder's int/long/float/double formats. */ + static Number unpackNumber(MessageUnpacker unpacker, String key) throws IOException { + MessageFormat format = unpacker.getNextFormat(); + switch (format) { + case NEGFIXINT: + case POSFIXINT: + case INT8: + case UINT8: + case INT16: + case UINT16: + case INT32: + case UINT32: + return unpacker.unpackInt(); + case INT64: + case UINT64: + return unpacker.unpackLong(); + case FLOAT32: + return unpacker.unpackFloat(); + case FLOAT64: + return unpacker.unpackDouble(); + default: + fail("Unexpected type in metrics values: " + format + " for key " + key); + return null; // unreachable + } + } + + /** + * Growable capturing channel the payload verifiers write the mapper output into before decoding. + * Grows by the source's remaining bytes when needed. + */ + static final class CapturingChannel implements WritableByteChannel { + private ByteBuffer captured; + + CapturingChannel(int size) { + this.captured = ByteBuffer.allocate(size); + } + + @Override + public int write(ByteBuffer src) { + if (captured.remaining() < src.remaining()) { + ByteBuffer bigger = ByteBuffer.allocate(captured.capacity() + src.remaining()); + captured.flip(); + bigger.put(captured); + captured = bigger; + return write(src); + } + captured.put(src); + return src.position(); + } + + /** Flips the buffer and returns it ready for reading. */ + ByteBuffer flipForReading() { + captured.flip(); + return captured; + } + + /** Resets the buffer for the next payload. */ + void resetForWriting() { + captured.position(0); + captured.limit(captured.capacity()); + } + + @Override + public boolean isOpen() { + return true; + } + + @Override + public void close() {} + } +} diff --git a/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV04PayloadTest.java b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV04PayloadTest.java new file mode 100644 index 00000000000..318da568a57 --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV04PayloadTest.java @@ -0,0 +1,404 @@ +package datadog.trace.common.writer.ddagent; + +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.DD_MEASURED; +import static datadog.trace.common.writer.TraceGenerator.generateRandomTraces; +import static datadog.trace.common.writer.ddagent.PayloadVerifiers.assertEqualsWithNullAsEmpty; +import static datadog.trace.common.writer.ddagent.PayloadVerifiers.unpackNumber; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +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 static org.junit.jupiter.params.provider.Arguments.arguments; + +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.DD64bTraceId; +import datadog.trace.api.DDTags; +import datadog.trace.api.DDTraceId; +import datadog.trace.api.ProcessTags; +import datadog.trace.bootstrap.instrumentation.api.Tags; +import datadog.trace.common.writer.Payload; +import datadog.trace.common.writer.TraceGenerator.PojoSpan; +import datadog.trace.core.DDSpanContext; +import datadog.trace.junit.utils.config.WithConfigExtension; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Stream; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.msgpack.core.MessagePack; +import org.msgpack.core.MessageUnpacker; + +@ExtendWith(WithConfigExtension.class) +class TraceMapperV04PayloadTest { + + // 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()); + } + + @ParameterizedTest(name = "buffer={0} traces={1} lowCardinality={2}") + @MethodSource("tracesWrittenCorrectlyArguments") + void tracesWrittenCorrectly(int bufferSize, int traceCount, boolean lowCardinality) { + List> traces = generateRandomTraces(traceCount, lowCardinality); + TraceMapperV0_4 traceMapper = new TraceMapperV0_4(); + PayloadVerifier verifier = new PayloadVerifier(traces, traceMapper); + MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(bufferSize, verifier)); + boolean tracesFitInBuffer = true; + for (List trace : traces) { + if (!packer.format(trace, traceMapper)) { + verifier.skipLargeTrace(); + tracesFitInBuffer = false; + // in the real like the mapper is always reset each trace. + // here we need to force it when we fail since the buffer will be reset as well + traceMapper.reset(); + } + } + packer.flush(); + + if (tracesFitInBuffer) { + verifier.verifyTracesConsumed(); + } + } + + private static Stream tracesWrittenCorrectlyArguments() { + return Stream.of( + arguments(20 << 10, 0, true), + arguments(20 << 10, 1, true), + arguments(30 << 10, 1, true), + arguments(30 << 10, 2, true), + arguments(20 << 10, 0, false), + arguments(20 << 10, 1, false), + arguments(30 << 10, 1, false), + arguments(30 << 10, 2, false), + arguments(100 << 10, 0, true), + arguments(100 << 10, 1, true), + arguments(100 << 10, 10, true), + arguments(100 << 10, 100, true), + arguments(100 << 10, 1000, true), + arguments(100 << 10, 0, false), + arguments(100 << 10, 1, false), + arguments(100 << 10, 10, false), + arguments(100 << 10, 100, false), + arguments(100 << 10, 1000, false)); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("fullSixtyFourBitTraceAndSpanIdentifiersArguments") + void fullSixtyFourBitTraceAndSpanIdentifiers( + String scenario, DDTraceId traceId, long spanId, long parentId) { + PojoSpan span = + new PojoSpan( + "service", + "operation", + "resource", + traceId, + spanId, + parentId, + 123L, + 456L, + 0, + Collections.emptyMap(), + Collections.emptyMap(), + "type", + false, + 0, + 0, + "origin"); + List> traces = Collections.singletonList(Collections.singletonList(span)); + TraceMapperV0_4 traceMapper = new TraceMapperV0_4(); + PayloadVerifier verifier = new PayloadVerifier(traces, traceMapper); + MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(20 << 10, verifier)); + + packer.format(Collections.singletonList(span), traceMapper); + packer.flush(); + + verifier.verifyTracesConsumed(); + } + + private static Stream fullSixtyFourBitTraceAndSpanIdentifiersArguments() { + return Stream.of( + arguments("ONE", DD64bTraceId.ONE, 2L, 3L), + arguments("MAX", DD64bTraceId.MAX, 2L, 3L), + arguments("negative", DD64bTraceId.from(-10), -11L, -12L)); + } + + @Test + void metaStructSupport() { + PojoSpan span = + new PojoSpan( + "service", + "operation", + "resource", + DDTraceId.ONE, + 1L, + -1L, + 123L, + 456L, + 0, + Collections.emptyMap(), + Collections.emptyMap(), + "type", + false, + 0, + 0, + "origin"); + List> stack = new ArrayList<>(); + for (StackTraceElement element : Thread.currentThread().getStackTrace()) { + Map frame = new HashMap<>(); + frame.put("file", element.getFileName() != null ? element.getFileName() : ""); + frame.put("class_name", element.getClassName() != null ? element.getClassName() : ""); + frame.put("function", element.getMethodName() != null ? element.getMethodName() : ""); + stack.add(frame); + } + span.setMetaStruct("stack", stack); + List> traces = Collections.singletonList(Collections.singletonList(span)); + TraceMapperV0_4 traceMapper = new TraceMapperV0_4(); + PayloadVerifier verifier = + new PayloadVerifier( + traces, + traceMapper, + (expected, received) -> { + MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(received); + List expectedStack = (List) expected; + int size = unpacker.unpackArrayHeader(); + assertEquals(expectedStack.size(), size); + for (Object entry : expectedStack) { + @SuppressWarnings("unchecked") + Map stackEntry = (Map) entry; + int fields = unpacker.unpackMapHeader(); + for (int f = 0; f < fields; ++f) { + String field = unpacker.unpackString(); + assertEquals(stackEntry.get(field), unpacker.unpackString()); + } + } + }); + MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(20 << 10, verifier)); + + packer.format(Collections.singletonList(span), traceMapper); + packer.flush(); + + verifier.verifyTracesConsumed(); + } + + @Test + void processTagsSerialization() { + assertNotNull(ProcessTags.getTagsForSerialization()); + List spans = new ArrayList<>(); + for (long spanId = 1; spanId <= 2; ++spanId) { + spans.add( + new PojoSpan( + "service", + "operation", + "resource", + DDTraceId.ONE, + spanId, + -1L, + 123L, + 456L, + 0, + Collections.emptyMap(), + Collections.emptyMap(), + "type", + false, + 0, + 0, + "origin")); + } + + List> traces = Collections.singletonList(spans); + TraceMapperV0_4 traceMapper = new TraceMapperV0_4(); + PayloadVerifier verifier = new PayloadVerifier(traces, traceMapper); + MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(20 << 10, verifier)); + + packer.format(spans, traceMapper); + packer.flush(); + + verifier.verifyTracesConsumed(); + } + + private static final class PayloadVerifier implements ByteBufferConsumer { + + private final List> expectedTraces; + private final TraceMapperV0_4 mapper; + private final MetaStructVerifier metaStructVerifier; + private final PayloadVerifiers.CapturingChannel channel = + new PayloadVerifiers.CapturingChannel(200 << 10); + + private int position = 0; + + private PayloadVerifier(List> traces, TraceMapperV0_4 mapper) { + this(traces, mapper, null); + } + + private PayloadVerifier( + List> traces, + TraceMapperV0_4 mapper, + MetaStructVerifier metaStructVerifier) { + this.expectedTraces = traces; + this.mapper = mapper; + this.metaStructVerifier = metaStructVerifier; + } + + void skipLargeTrace() { + ++position; + } + + @Override + public void accept(int messageCount, ByteBuffer buffer) { + if (expectedTraces.isEmpty() && messageCount == 0) { + return; + } + int processTagsCount = 0; + try { + Payload payload = mapper.newPayload().withBody(messageCount, buffer); + payload.writeTo(channel); + MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(channel.flipForReading()); + int traceCount = unpacker.unpackArrayHeader(); + for (int i = 0; i < traceCount; ++i) { + List expectedTrace = expectedTraces.get(position++); + int spanCount = unpacker.unpackArrayHeader(); + assertEquals(expectedTrace.size(), spanCount); + for (int k = 0; k < spanCount; ++k) { + PojoSpan expectedSpan = expectedTrace.get(k); + int elementCount = unpacker.unpackMapHeader(); + boolean hasMetaStruct = !expectedSpan.getMetaStruct().isEmpty(); + assertEquals(hasMetaStruct ? 13 : 12, elementCount); + assertEquals("service", unpacker.unpackString()); + String serviceName = unpacker.unpackString(); + assertEqualsWithNullAsEmpty(expectedSpan.getServiceName(), serviceName); + assertEquals("name", unpacker.unpackString()); + String operationName = unpacker.unpackString(); + assertEqualsWithNullAsEmpty(expectedSpan.getOperationName(), operationName); + assertEquals("resource", unpacker.unpackString()); + String resourceName = unpacker.unpackString(); + assertEqualsWithNullAsEmpty(expectedSpan.getResourceName(), resourceName); + assertEquals("trace_id", unpacker.unpackString()); + long traceId = unpacker.unpackValue().asNumberValue().toLong(); + assertEquals(expectedSpan.getTraceId().toLong(), traceId); + assertEquals("span_id", unpacker.unpackString()); + long spanId = unpacker.unpackValue().asNumberValue().toLong(); + assertEquals(expectedSpan.getSpanId(), spanId); + assertEquals("parent_id", unpacker.unpackString()); + long parentId = unpacker.unpackValue().asNumberValue().toLong(); + assertEquals(expectedSpan.getParentId(), parentId); + assertEquals("start", unpacker.unpackString()); + long startTime = unpacker.unpackLong(); + assertEquals(expectedSpan.getStartTime(), startTime); + assertEquals("duration", unpacker.unpackString()); + long duration = unpacker.unpackLong(); + assertEquals(expectedSpan.getDurationNano(), duration); + assertEquals("type", unpacker.unpackString()); + String type = unpacker.unpackString(); + assertEquals(expectedSpan.getType(), type); + assertEquals("error", unpacker.unpackString()); + int error = unpacker.unpackInt(); + assertEquals(expectedSpan.getError(), error); + assertEquals("metrics", unpacker.unpackString()); + int metricsSize = unpacker.unpackMapHeader(); + HashMap metrics = new HashMap<>(); + for (int j = 0; j < metricsSize; ++j) { + String key = unpacker.unpackString(); + Number n = unpackNumber(unpacker, key); + if (DD_MEASURED.toString().equals(key)) { + assertTrue( + (n.intValue() == 1 && expectedSpan.isMeasured()) || !expectedSpan.isMeasured()); + } else if (DDSpanContext.PRIORITY_SAMPLING_KEY.equals(key)) { + // check that priority sampling is only on first and last span + if (k == 0 || k == spanCount - 1) { + assertEquals(expectedSpan.samplingPriority(), n.intValue()); + } else { + assertFalse(expectedSpan.hasSamplingPriority()); + } + } else { + metrics.put(key, n); + } + } + for (Map.Entry metric : metrics.entrySet()) { + if (metric.getValue() instanceof Double || metric.getValue() instanceof Float) { + assertEquals( + ((Number) expectedSpan.getTag(metric.getKey())).doubleValue(), + metric.getValue().doubleValue(), + 0.001); + } else { + // Integer-typed metrics round-trip through msgpack's minimal encoding, so a Long + // tag can come back as an Integer (and vice versa). Compare numerically. + assertEquals( + ((Number) expectedSpan.getTag(metric.getKey())).longValue(), + metric.getValue().longValue()); + } + } + assertEquals("meta", unpacker.unpackString()); + int metaSize = unpacker.unpackMapHeader(); + HashMap meta = new HashMap<>(); + for (int j = 0; j < metaSize; ++j) { + meta.put(unpacker.unpackString(), unpacker.unpackString()); + } + for (Map.Entry entry : meta.entrySet()) { + if (Tags.HTTP_STATUS.equals(entry.getKey())) { + assertEquals(String.valueOf(expectedSpan.getHttpStatusCode()), entry.getValue()); + } else if (DDTags.ORIGIN_KEY.equals(entry.getKey())) { + assertEquals(expectedSpan.getOrigin(), entry.getValue()); + } else if (DDTags.PROCESS_TAGS.equals(entry.getKey())) { + assertTrue(Config.get().isExperimentalPropagateProcessTagsEnabled()); + assertEquals(0, k); + assertEquals(ProcessTags.getTagsForSerialization().toString(), entry.getValue()); + processTagsCount++; + } else { + Object tag = expectedSpan.getTag(entry.getKey()); + if (tag != null) { + assertEquals(String.valueOf(tag), entry.getValue()); + } else { + assertEquals(expectedSpan.getBaggage().get(entry.getKey()), entry.getValue()); + } + } + } + if (hasMetaStruct) { + Map metaStruct = expectedSpan.getMetaStruct(); + assertEquals("meta_struct", unpacker.unpackString()); + int metaStructSize = unpacker.unpackMapHeader(); + for (int j = 0; j < metaStructSize; ++j) { + String field = unpacker.unpackString(); + if (metaStructVerifier != null) { + byte[] binary = new byte[unpacker.unpackBinaryHeader()]; + unpacker.readPayload(binary); + metaStructVerifier.verify(metaStruct.get(field), binary); + } + } + } + } + } + } catch (IOException e) { + fail(e.getMessage()); + } finally { + mapper.reset(); + channel.resetForWriting(); + assertEquals( + Config.get().isExperimentalPropagateProcessTagsEnabled() ? 1 : 0, processTagsCount); + } + } + + void verifyTracesConsumed() { + assertEquals(expectedTraces.size(), position); + } + } + + @FunctionalInterface + private interface MetaStructVerifier { + void verify(E expected, byte[] received) throws IOException; + } +} diff --git a/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV05PayloadTest.java b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV05PayloadTest.java new file mode 100644 index 00000000000..7a04fd8924e --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/common/writer/ddagent/TraceMapperV05PayloadTest.java @@ -0,0 +1,355 @@ +package datadog.trace.common.writer.ddagent; + +import static datadog.trace.api.config.GeneralConfig.EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED; +import static datadog.trace.bootstrap.instrumentation.api.InstrumentationTags.DD_MEASURED; +import static datadog.trace.common.writer.TraceGenerator.generateRandomTraces; +import static datadog.trace.common.writer.ddagent.PayloadVerifiers.assertEqualsWithNullAsEmpty; +import static datadog.trace.common.writer.ddagent.PayloadVerifiers.unpackNumber; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +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 static org.junit.jupiter.params.provider.Arguments.arguments; + +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.DDTags; +import datadog.trace.api.DDTraceId; +import datadog.trace.api.ProcessTags; +import datadog.trace.api.sampling.PrioritySampling; +import datadog.trace.bootstrap.instrumentation.api.Tags; +import datadog.trace.common.writer.Payload; +import datadog.trace.common.writer.TraceGenerator.PojoSpan; +import datadog.trace.core.DDSpanContext; +import datadog.trace.junit.utils.config.WithConfig; +import datadog.trace.junit.utils.config.WithConfigExtension; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Stream; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.msgpack.core.MessagePack; +import org.msgpack.core.MessageUnpacker; + +@ExtendWith(WithConfigExtension.class) +class TraceMapperV05PayloadTest { + + // 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()); + } + + // disable process tags since they are only on the first span of the chunk otherwise the + // calculation woes + @Test + @WithConfig(key = EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED, value = "false") + void bodyOverflowCausesAFlush() { + // 4x 36 ASCII characters and 2 bytes of msgpack string prefix + int dictionarySpacePerTrace = 4 * (36 + 2); + // enough space for two traces with distinct string values, plus the header + int dictionarySize = dictionarySpacePerTrace * 2 + 5; + TraceMapperV0_5 traceMapper = new TraceMapperV0_5(dictionarySize); + List repeatedTrace = + Collections.singletonList( + new PojoSpan( + UUID.randomUUID().toString(), + UUID.randomUUID().toString(), + UUID.randomUUID().toString(), + DDTraceId.ZERO, + DDSpanId.ZERO, + DDSpanId.ZERO, + 10000, + 100, + 0, + Collections.emptyMap(), + Collections.emptyMap(), + UUID.randomUUID().toString(), + false, + PrioritySampling.UNSET, + 0, + null)); + int traceSize = calculateSize(repeatedTrace); + // 30KB body + int bufferSize = 30 << 10; + int tracesRequiredToOverflowBody = (int) Math.ceil((double) bufferSize / traceSize) + 1; + List> traces = new ArrayList<>(tracesRequiredToOverflowBody); + for (int i = 0; i < tracesRequiredToOverflowBody; ++i) { + traces.add(repeatedTrace); + } + // the last one won't be flushed + List> flushedTraces = new ArrayList<>(traces); + flushedTraces.remove(traces.size() - 1); + // need space for the overflowing buffer, the dictionary, and two small array headers + PayloadVerifier verifier = + new PayloadVerifier(flushedTraces, traceMapper, bufferSize + dictionarySize + 1 + 1 + 5); + MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(bufferSize, verifier)); + for (List trace : traces) { + packer.format(trace, traceMapper); + } + verifier.verifyTracesConsumed(); + } + + @ParameterizedTest(name = "buffer={0} dict={1} traces={2} lowCardinality={3}") + @MethodSource("dictionaryCompressedTracesWrittenCorrectlyArguments") + void dictionaryCompressedTracesWrittenCorrectly( + int bufferSize, int dictionarySize, int traceCount, boolean lowCardinality) { + List> traces = generateRandomTraces(traceCount, lowCardinality); + TraceMapperV0_5 traceMapper = new TraceMapperV0_5(dictionarySize); + PayloadVerifier verifier = new PayloadVerifier(traces, traceMapper); + MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(bufferSize, verifier)); + boolean tracesFitInBuffer = true; + for (List trace : traces) { + if (!packer.format(trace, traceMapper)) { + verifier.skipLargeTrace(); + tracesFitInBuffer = false; + } + } + packer.flush(); + if (tracesFitInBuffer) { + verifier.verifyTracesConsumed(); + } + } + + private static Stream dictionaryCompressedTracesWrittenCorrectlyArguments() { + return Stream.of( + arguments(10 << 10, 10 << 10, 0, true), + arguments(10 << 10, 10 << 10, 1, true), + arguments(10 << 10, 10 << 10, 10, true), + arguments(10 << 10, 10 << 10, 100, true), + arguments(10 << 10, 100 << 10, 1, true), + arguments(10 << 10, 100 << 10, 10, true), + arguments(10 << 10, 100 << 10, 100, true), + arguments(10 << 10, 10 << 10, 0, false), + arguments(10 << 10, 10 << 10, 1, false), + arguments(10 << 10, 10 << 10, 10, false), + arguments(10 << 10, 10 << 10, 100, false), + arguments(10 << 10, 100 << 10, 1, false), + arguments(10 << 10, 100 << 10, 10, false), + arguments(10 << 10, 100 << 10, 100, false), + arguments(100 << 10, 10 << 10, 0, true), + arguments(100 << 10, 10 << 10, 1, true), + arguments(100 << 10, 10 << 10, 10, true), + arguments(100 << 10, 10 << 10, 100, true), + arguments(100 << 10, 100 << 10, 1, true), + arguments(100 << 10, 100 << 10, 10, true), + arguments(100 << 10, 100 << 10, 100, true), + arguments(100 << 10, 10 << 10, 0, false), + arguments(100 << 10, 10 << 10, 1, false), + arguments(100 << 10, 10 << 10, 10, false), + arguments(100 << 10, 10 << 10, 100, false), + arguments(100 << 10, 100 << 10, 1, false), + arguments(100 << 10, 100 << 10, 10, false), + arguments(100 << 10, 100 << 10, 1000, false)); + } + + @Test + void processTagsSerialization() { + assertNotNull(ProcessTags.getTagsForSerialization()); + List spans = new ArrayList<>(); + for (long spanId = 1; spanId <= 2; ++spanId) { + spans.add( + new PojoSpan( + "service", + "operation", + "resource", + DDTraceId.ONE, + spanId, + -1L, + 123L, + 456L, + 0, + Collections.emptyMap(), + Collections.emptyMap(), + "type", + false, + 0, + 0, + "origin")); + } + + List> traces = Collections.singletonList(spans); + TraceMapperV0_5 traceMapper = new TraceMapperV0_5(); + PayloadVerifier verifier = new PayloadVerifier(traces, traceMapper); + MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(20 << 10, verifier)); + + packer.format(spans, traceMapper); + packer.flush(); + + verifier.verifyTracesConsumed(); + } + + private static final class PayloadVerifier implements ByteBufferConsumer { + + private final List> expectedTraces; + private final TraceMapperV0_5 mapper; + private final PayloadVerifiers.CapturingChannel channel; + + private int position = 0; + + private PayloadVerifier(List> traces, TraceMapperV0_5 mapper) { + this(traces, mapper, 200 << 10); + } + + private PayloadVerifier(List> traces, TraceMapperV0_5 mapper, int size) { + this.expectedTraces = traces; + this.mapper = mapper; + this.channel = new PayloadVerifiers.CapturingChannel(size); + } + + void skipLargeTrace() { + ++position; + } + + @Override + public void accept(int messageCount, ByteBuffer buffer) { + int processTagsCount = 0; + try { + Payload payload = mapper.newPayload().withBody(messageCount, buffer); + payload.writeTo(channel); + MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(channel.flipForReading()); + int header = unpacker.unpackArrayHeader(); + assertEquals(2, header); + int dictionarySize = unpacker.unpackArrayHeader(); + String[] dictionary = new String[dictionarySize]; + for (int i = 0; i < dictionary.length; ++i) { + dictionary[i] = unpacker.unpackString(); + } + int traceCount = unpacker.unpackArrayHeader(); + for (int i = 0; i < traceCount; ++i) { + List expectedTrace = expectedTraces.get(position++); + int spanCount = unpacker.unpackArrayHeader(); + assertEquals(expectedTrace.size(), spanCount); + for (int k = 0; k < spanCount; ++k) { + PojoSpan expectedSpan = expectedTrace.get(k); + int elementCount = unpacker.unpackArrayHeader(); + assertEquals(12, elementCount); + String serviceName = dictionary[unpacker.unpackInt()]; + assertEqualsWithNullAsEmpty(expectedSpan.getServiceName(), serviceName); + String operationName = dictionary[unpacker.unpackInt()]; + assertEqualsWithNullAsEmpty(expectedSpan.getOperationName(), operationName); + String resourceName = dictionary[unpacker.unpackInt()]; + assertEqualsWithNullAsEmpty(expectedSpan.getResourceName(), resourceName); + long traceId = unpacker.unpackValue().asNumberValue().toLong(); + assertEquals(expectedSpan.getTraceId().toLong(), traceId); + long spanId = unpacker.unpackValue().asNumberValue().toLong(); + assertEquals(expectedSpan.getSpanId(), spanId); + long parentId = unpacker.unpackValue().asNumberValue().toLong(); + assertEquals(expectedSpan.getParentId(), parentId); + long startTime = unpacker.unpackLong(); + assertEquals(expectedSpan.getStartTime(), startTime); + long duration = unpacker.unpackLong(); + assertEquals(expectedSpan.getDurationNano(), duration); + int error = unpacker.unpackInt(); + assertEquals(expectedSpan.getError(), error); + int metaSize = unpacker.unpackMapHeader(); + HashMap meta = new HashMap<>(); + for (int j = 0; j < metaSize; ++j) { + meta.put(dictionary[unpacker.unpackInt()], dictionary[unpacker.unpackInt()]); + } + for (Map.Entry entry : meta.entrySet()) { + if (Tags.HTTP_STATUS.equals(entry.getKey())) { + assertEquals(String.valueOf(expectedSpan.getHttpStatusCode()), entry.getValue()); + } else if (DDTags.ORIGIN_KEY.equals(entry.getKey())) { + assertEquals(expectedSpan.getOrigin(), entry.getValue()); + } else if (DDTags.PROCESS_TAGS.equals(entry.getKey())) { + processTagsCount++; + assertTrue(Config.get().isExperimentalPropagateProcessTagsEnabled()); + assertEquals(0, k); + assertEquals(ProcessTags.getTagsForSerialization().toString(), entry.getValue()); + } else { + Object tag = expectedSpan.getTag(entry.getKey()); + if (tag != null) { + assertEquals(String.valueOf(tag), entry.getValue()); + } else { + assertEquals(expectedSpan.getBaggage().get(entry.getKey()), entry.getValue()); + } + } + } + int metricsSize = unpacker.unpackMapHeader(); + HashMap metrics = new HashMap<>(); + for (int j = 0; j < metricsSize; ++j) { + String key = dictionary[unpacker.unpackInt()]; + Number n = unpackNumber(unpacker, key); + if (DD_MEASURED.toString().equals(key)) { + assertTrue( + (n.intValue() == 1 && expectedSpan.isMeasured()) || !expectedSpan.isMeasured()); + } else if (DDSpanContext.PRIORITY_SAMPLING_KEY.equals(key)) { + // check that priority sampling is only on first and last span + if (k == 0 || k == spanCount - 1) { + assertEquals(expectedSpan.samplingPriority(), n.intValue()); + } else { + assertFalse(expectedSpan.hasSamplingPriority()); + } + } else { + metrics.put(key, n); + } + } + for (Map.Entry metric : metrics.entrySet()) { + if (metric.getValue() instanceof Double || metric.getValue() instanceof Float) { + assertEquals( + ((Number) expectedSpan.getTag(metric.getKey())).doubleValue(), + metric.getValue().doubleValue(), + 0.001, + metric.getKey()); + } else { + // Integer-typed metrics round-trip through msgpack's minimal encoding, so a Long + // tag can come back as an Integer (and vice versa). Compare numerically. + assertEquals( + ((Number) expectedSpan.getTag(metric.getKey())).longValue(), + metric.getValue().longValue(), + metric.getKey()); + } + } + String type = dictionary[unpacker.unpackInt()]; + assertEquals(expectedSpan.getType(), type); + } + } + } catch (IOException e) { + fail(e.getMessage()); + } finally { + assertEquals( + Config.get().isExperimentalPropagateProcessTagsEnabled() ? 1 : 0, processTagsCount); + mapper.reset(); + channel.resetForWriting(); + } + } + + void verifyTracesConsumed() { + assertEquals(expectedTraces.size(), position); + } + } + + private static int calculateSize(List trace) { + AtomicInteger size = new AtomicInteger(); + MsgPackWriter packer = + new MsgPackWriter( + new FlushingBuffer( + 1024, + new ByteBufferConsumer() { + @Override + public void accept(int messageCount, ByteBuffer buffer) { + size.set(buffer.limit() - buffer.position()); + } + })); + packer.format(trace, new TraceMapperV0_5(1024)); + packer.flush(); + return size.get(); + } +}