diff --git a/dd-trace-core/src/traceAgentTest/groovy/AbstractTraceAgentTest.groovy b/dd-trace-core/src/traceAgentTest/groovy/AbstractTraceAgentTest.groovy deleted file mode 100644 index f2fe38cc18d..00000000000 --- a/dd-trace-core/src/traceAgentTest/groovy/AbstractTraceAgentTest.groovy +++ /dev/null @@ -1,60 +0,0 @@ -import datadog.trace.api.ConfigDefaults -import datadog.trace.api.config.TracerConfig -import datadog.trace.test.util.DDSpecification -import org.testcontainers.containers.GenericContainer -import org.testcontainers.containers.startupcheck.MinimumDurationRunningStartupCheckStrategy -import spock.lang.Shared - -import java.time.Duration - -abstract class AbstractTraceAgentTest extends DDSpecification { - @Shared - def agentContainer - - def setupSpec() { - /* - CI will provide us with agent container running along side our build. - When building locally, however, we need to take matters into our own hands - and we use 'testcontainers' for this. - */ - if ("true" != System.getenv("CI")) { - agentContainer = new GenericContainer("datadog/agent:7.40.1") - .withEnv(["DD_APM_ENABLED": "true", - "DD_BIND_HOST" : "0.0.0.0", - "DD_API_KEY" : "invalid_key_but_this_is_fine", - "DD_HOSTNAME" : "doesnotexist", - "DD_LOGS_STDOUT": "yes"]) - .withExposedPorts(datadog.trace.api.ConfigDefaults.DEFAULT_TRACE_AGENT_PORT) - .withStartupTimeout(Duration.ofSeconds(120)) - // Apparently we need to sleep for a bit so agent's response `{"service:,env:":1}` in rate_by_service. - // This is clearly a race-condition and maybe we should avoid verifying complete response - .withStartupCheckStrategy(new MinimumDurationRunningStartupCheckStrategy(Duration.ofSeconds(10))) - agentContainer.start() - } - } - - def setup() { - injectSysConfig(TracerConfig.AGENT_HOST, getAgentContainerHost()) - injectSysConfig(TracerConfig.TRACE_AGENT_PORT, getAgentContainerPort()) - } - - String getAgentContainerHost() { - if (agentContainer) { - return (String) agentContainer.getHost() - } - - return System.getenv("CI_AGENT_HOST") - } - - String getAgentContainerPort() { - if (agentContainer) { - return (String) agentContainer.getMappedPort(ConfigDefaults.DEFAULT_TRACE_AGENT_PORT) - } - - return ConfigDefaults.DEFAULT_TRACE_AGENT_PORT - } - - def cleanupSpec() { - agentContainer?.stop() - } -} diff --git a/dd-trace-core/src/traceAgentTest/groovy/DDApiIntegrationTest.groovy b/dd-trace-core/src/traceAgentTest/groovy/DDApiIntegrationTest.groovy deleted file mode 100644 index b2d538a26b4..00000000000 --- a/dd-trace-core/src/traceAgentTest/groovy/DDApiIntegrationTest.groovy +++ /dev/null @@ -1,206 +0,0 @@ -import static datadog.trace.api.ProtocolVersion.V0_4 -import static datadog.trace.api.ProtocolVersion.V0_5 -import static datadog.trace.api.ProtocolVersion.V1_0 - -import datadog.communication.ddagent.DDAgentFeaturesDiscovery -import datadog.communication.http.OkHttpUtils -import datadog.communication.serialization.ByteBufferConsumer -import datadog.communication.serialization.FlushingBuffer -import datadog.communication.serialization.msgpack.MsgPackWriter -import datadog.metrics.api.statsd.StatsDClient -import datadog.metrics.impl.MonitoringImpl -import datadog.trace.api.Config -import datadog.trace.api.ProtocolVersion -import datadog.trace.common.writer.ListWriter -import datadog.trace.common.writer.Payload -import datadog.trace.common.writer.RemoteApi -import datadog.trace.common.writer.RemoteResponseListener -import datadog.trace.common.writer.ddagent.DDAgentApi -import datadog.trace.common.writer.ddagent.TraceMapper -import datadog.trace.common.writer.ddagent.TraceMapperV0_4 -import datadog.trace.common.writer.ddagent.TraceMapperV0_5 -import datadog.trace.common.writer.ddagent.TraceMapperV1 -import datadog.trace.core.CoreTracer -import datadog.trace.core.DDSpan -import java.nio.ByteBuffer -import java.util.concurrent.TimeUnit -import java.util.concurrent.atomic.AtomicReference -import okhttp3.HttpUrl -import okhttp3.OkHttpClient -import spock.lang.Shared - -class DDApiIntegrationTest extends AbstractTraceAgentTest { - def tracer - DDSpan span - - // Looks like okHttp needs to resolve this, even for connection over socket - static final SOMEHOST = "datadoghq.com" - static final SOMEPORT = 123 - - @Shared - Process process - @Shared - File socketPath - - def discovery - def udsDiscovery - def api - def unixDomainSocketApi - TraceMapper mapper - String traceEndpoint - - def endpoint = new AtomicReference(null) - def agentResponse = new AtomicReference>>(null) - - RemoteResponseListener responseListener = { String receivedEndpoint, Map> responseJson -> - endpoint.set(receivedEndpoint) - agentResponse.set(responseJson) - } - - def setupSpec() { - File tmpDir = File.createTempDir() - tmpDir.deleteOnExit() - socketPath = new File(tmpDir, "socket") - println "!!!socat UNIX-LISEN:${socketPath},reuseaddr,fork TCP-CONNECT:${agentContainerHost}:${agentContainerPort}" - process = Runtime.getRuntime().exec("socat UNIX-LISTEN:${socketPath},reuseaddr,fork TCP-CONNECT:${agentContainerHost}:${agentContainerPort}") - } - - def setup() { - tracer = CoreTracer.builder().writer(new ListWriter()).build() - span = tracer.buildSpan("datadog", "fakeOperation").start() - Thread.sleep(1) - span.finish() - } - - def cleanup() { - tracer?.close() - } - - def cleanupSpec() { - process?.destroy() - } - - def beforeTest(ProtocolVersion protocol) { - MonitoringImpl monitoring = new MonitoringImpl(StatsDClient.NO_OP, 1, TimeUnit.SECONDS) - HttpUrl agentUrl = HttpUrl.get(Config.get().getAgentUrl()) - OkHttpClient httpClient = OkHttpUtils.buildHttpClient(agentUrl, 5000) - discovery = new DDAgentFeaturesDiscovery(httpClient, monitoring, agentUrl, protocol, true, false) - api = new DDAgentApi(httpClient, agentUrl, discovery, monitoring, false) - api.addResponseListener(responseListener) - HttpUrl udsAgentUrl = HttpUrl.get(String.format("http://%s:%d", SOMEHOST, SOMEPORT)) - OkHttpClient udsClient = OkHttpUtils.buildHttpClient(true, socketPath.toString(), null, 5000) - udsDiscovery = new DDAgentFeaturesDiscovery(udsClient, monitoring, agentUrl, protocol, true, false) - unixDomainSocketApi = new DDAgentApi(udsClient, udsAgentUrl, udsDiscovery, monitoring, false) - unixDomainSocketApi.addResponseListener(responseListener) - mapper = [ - (V1_0): new TraceMapperV1(), - (V0_5): new TraceMapperV0_5(), - ].get(protocol, new TraceMapperV0_4()) - traceEndpoint = protocol.endpoint() - } - - def "Sending empty traces succeeds (test #test)"() { - setup: - beforeTest(protocol) - expect: - RemoteApi.Response response = api.sendSerializedTraces(prepareRequest(traces, mapper)) - assert !response.response().isEmpty() - assert !response.exception().present - assert response.status().present - assert 200 == response.status().asInt - assert response.success() - assert discovery.getTraceEndpoint() == traceEndpoint - assert endpoint.get() == "${Config.get().getAgentUrl()}/${traceEndpoint}" - assert agentResponse.get()["rate_by_service"] instanceof Map - - where: - // spotless:off - traces | test | protocol - [] | 1 | V0_5 - (1..16).collect { [] } | 4 | V0_5 - [] | 5 | V0_4 - (1..16).collect { [] } | 8 | V0_4 - // spotless:on - } - - def "Sending traces succeeds"() { - setup: - beforeTest(protocol) - expect: - RemoteApi.Response response = api.sendSerializedTraces(prepareRequest([[span]], mapper)) - assert !response.response().isEmpty() - assert !response.exception().present - assert response.status().present - assert 200 == response.status().asInt - assert response.success() - assert discovery.getTraceEndpoint() == traceEndpoint - assert endpoint.get() == "${Config.get().getAgentUrl()}/${traceEndpoint}" - assert agentResponse.get()["rate_by_service"] instanceof Map - - where: - protocol << [V0_5, V0_4] - } - - def "Sending empty traces to unix domain socket succeeds (test #test)"() { - setup: - beforeTest(protocol) - expect: - RemoteApi.Response response = unixDomainSocketApi.sendSerializedTraces(prepareRequest(traces, mapper)) - assert !response.response().isEmpty() - assert !response.exception().present - assert response.status().present - assert 200 == response.status().asInt - assert response.success() - assert udsDiscovery.getTraceEndpoint() == traceEndpoint - assert endpoint.get() == "http://${SOMEHOST}:${SOMEPORT}/${traceEndpoint}" - assert agentResponse.get()["rate_by_service"] instanceof Map - - where: - // spotless:off - traces | test | protocol - [] | 1 | V0_5 - [] | 3 | V0_4 - // spotless:on - } - - def "Sending traces to unix domain socket succeeds (protocol #protocol)"() { - setup: - beforeTest(protocol) - expect: - RemoteApi.Response response = unixDomainSocketApi.sendSerializedTraces(prepareRequest([[span]], mapper)) - assert !response.response().isEmpty() - assert !response.exception().present - assert response.status().present - assert 200 == response.status().asInt - assert response.success() - assert udsDiscovery.getTraceEndpoint() == traceEndpoint - assert endpoint.get() == "http://${SOMEHOST}:${SOMEPORT}/${traceEndpoint}" - assert agentResponse.get()["rate_by_service"] instanceof Map - - where: - protocol << [V0_5, V0_4] - } - - - static class Traces implements ByteBufferConsumer { - int traceCount - ByteBuffer buffer - - @Override - void accept(int messageCount, ByteBuffer buffer) { - this.buffer = buffer - this.traceCount = messageCount - } - } - - Payload prepareRequest(List> traces, TraceMapper traceMapper) { - Traces traceCapture = new Traces() - def packer = new MsgPackWriter(new FlushingBuffer(1 << 10, traceCapture)) - for (trace in traces) { - packer.format(trace, traceMapper) - } - packer.flush() - return traceMapper.newPayload() - .withBody(traceCapture.traceCount, traceCapture.buffer) - } -} diff --git a/dd-trace-core/src/traceAgentTest/groovy/DataStreamsIntegrationTest.groovy b/dd-trace-core/src/traceAgentTest/groovy/DataStreamsIntegrationTest.groovy deleted file mode 100644 index 15f71c0e45c..00000000000 --- a/dd-trace-core/src/traceAgentTest/groovy/DataStreamsIntegrationTest.groovy +++ /dev/null @@ -1,74 +0,0 @@ -import datadog.communication.ddagent.DDAgentFeaturesDiscovery -import datadog.communication.ddagent.SharedCommunicationObjects -import datadog.communication.http.OkHttpUtils -import datadog.trace.api.Config -import datadog.trace.api.TraceConfig -import datadog.trace.api.datastreams.DataStreamsTags -import datadog.trace.api.time.ControllableTimeSource -import datadog.trace.api.datastreams.StatsPoint -import datadog.trace.common.metrics.EventListener -import datadog.trace.common.metrics.OkHttpSink -import datadog.trace.core.datastreams.DefaultDataStreamsMonitoring -import okhttp3.HttpUrl -import spock.lang.Ignore -import spock.util.concurrent.PollingConditions - -import java.util.concurrent.CopyOnWriteArrayList - -import static datadog.trace.common.metrics.EventListener.EventType.OK - -@Ignore("The agent in CI doesn't have a valid API key. Unlike metrics and traces, data streams fails in this case") -class DataStreamsIntegrationTest extends AbstractTraceAgentTest { - - def "Sending stats bucket to agent should notify with OK event"() { - given: - def conditions = new PollingConditions(timeout: 1) - - def sharedCommunicationObjects = new SharedCommunicationObjects() - sharedCommunicationObjects.createRemaining(Config.get()) - - OkHttpSink sink = new OkHttpSink( - OkHttpUtils.buildHttpClient(HttpUrl.parse(Config.get().getAgentUrl()), 5000L), - Config.get().getAgentUrl(), - DDAgentFeaturesDiscovery.V01_DATASTREAMS_ENDPOINT, - false, - true, - [:]) - - def listener = new BlockingListener() - sink.register(listener) - - def timeSource = new ControllableTimeSource() - - def traceConfig = Mock(TraceConfig) { - isDataStreamsEnabled() >> true - } - - when: - def dataStreams = new DefaultDataStreamsMonitoring(sink, sharedCommunicationObjects.featuresDiscovery(Config.get()), timeSource, { traceConfig }, Config.get()) - dataStreams.start() - def tg = DataStreamsTags.create("testType", null, "testTopic", "testGroup", null) - dataStreams.add(new StatsPoint(tg, 1, 2, 5, timeSource.currentTimeNanos, 0, 0, 0, null)) - timeSource.advance(Config.get().getDataStreamsBucketDurationNanoseconds()) - dataStreams.report() - - then: - sharedCommunicationObjects.featuresDiscovery.supportsDataStreams() - conditions.eventually { - assert listener.events.size() == 1 - } - listener.events[0] == OK - - cleanup: - dataStreams.close() - } - - static class BlockingListener implements EventListener { - List events = new CopyOnWriteArrayList<>() - - @Override - void onEvent(EventType eventType, String message) { - events.add(eventType) - } - } -} diff --git a/dd-trace-core/src/traceAgentTest/groovy/TraceGenerator.groovy b/dd-trace-core/src/traceAgentTest/groovy/TraceGenerator.groovy deleted file mode 100644 index a0c81cbd0cb..00000000000 --- a/dd-trace-core/src/traceAgentTest/groovy/TraceGenerator.groovy +++ /dev/null @@ -1,435 +0,0 @@ -import static datadog.trace.api.ProcessTags.tagsForSerialization -import static datadog.trace.api.TagMap.fromMap -import static datadog.trace.api.sampling.PrioritySampling.UNSET -import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND -import static java.lang.Thread.currentThread -import static java.util.Collections.emptyList - -import datadog.trace.api.DDSpanId -import datadog.trace.api.DDTags -import datadog.trace.api.DDTraceId -import datadog.trace.api.IdGenerationStrategy -import datadog.trace.api.TagMap -import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString -import datadog.trace.core.CoreSpan -import datadog.trace.core.Metadata -import datadog.trace.core.MetadataConsumer -import datadog.trace.core.SpanKindFilter -import java.util.concurrent.ThreadLocalRandom -import java.util.concurrent.TimeUnit - -class TraceGenerator { - - static List> generateRandomTraces(int howMany, boolean lowCardinality) { - List> traces = new ArrayList<>(howMany) - for (int i = 0; i < howMany; ++i) { - int traceSize = ThreadLocalRandom.current().nextInt(2, 20) - traces.add(generateRandomTrace(traceSize, lowCardinality)) - } - return traces - } - - private static List generateRandomTrace(int size, boolean lowCardinality) { - List trace = new ArrayList<>(size) - long traceId = ThreadLocalRandom.current().nextLong(1, Long.MAX_VALUE) - for (int i = 0; i < size; ++i) { - trace.add(randomSpan(traceId, lowCardinality)) - } - return trace - } - - private static final IdGenerationStrategy ID_GENERATION_STRATEGY = IdGenerationStrategy.fromName("RANDOM") - - private static CoreSpan randomSpan(long traceId, boolean lowCardinality) { - ThreadLocalRandom random = ThreadLocalRandom.current() - Map baggage = new HashMap<>() - if (random.nextBoolean()) { - baggage.put("baggage-key", lowCardinality ? "x" : randomString(100)) - if (random.nextBoolean()) { - baggage.put("tag.1", "bar") - baggage.put("tag.2", "qux") - } - } - Map tags = new HashMap<>() - int tagCount = random.nextInt(0, 20) - for (int i = 0; i < tagCount; ++i) { - tags.put("tag." + i, random.nextBoolean() ? "foo" : randomString(2000)) - tags.put("tag.1." + i, lowCardinality ? "y" : UUID.randomUUID()) - switch (random.nextInt(8)) { - case 0: - tags.put("tag.3." + i , BigDecimal.valueOf(random.nextDouble())) - break - case 1: - tags.put("tag.3." + i , BigInteger.valueOf(random.nextLong())) - break - default: - break - } - } - int metricCount = random.nextInt(0, 20) - for (int i = 0; i < metricCount; ++i) { - String name = "metric." + i - Number metric = null - switch (random.nextInt(4)) { - case 0: - metric = random.nextInt() - break - case 1: - metric = random.nextLong() - break - case 2: - metric = random.nextFloat() - break - case 3: - metric = random.nextDouble() - break - } - tags.put(name, metric) - } - return new PojoSpan( - "service-" + random.nextInt(lowCardinality ? 1 : 10), - "operation-" + random.nextInt(lowCardinality ? 1 : 100), - UTF8BytesString.create("resource-" + random.nextInt(lowCardinality ? 1 : 100)), - DDTraceId.from(traceId), - ID_GENERATION_STRATEGY.generateSpanId(), - DDSpanId.ZERO, - TimeUnit.MILLISECONDS.toNanos(System.currentTimeMillis()), - random.nextLong(500, 10_000_000), - random.nextInt(2), - baggage, - tags, - "type-" + random.nextInt(lowCardinality ? 1 : 100), - random.nextBoolean()) - } - - private static String randomString(int maxLength) { - char[] chars = new char[ThreadLocalRandom.current().nextInt(maxLength)] - for (int i = 0; i < chars.length; ++i) { - char next = (char) ThreadLocalRandom.current().nextInt((int) Character.MAX_VALUE) - if (Character.isSurrogate(next)) { - if (i < chars.length - 1) { - chars[i++] = '\uD801' - chars[i] = '\uDC01' - } else { - chars[i] = 'a' - } - } else { - chars[i] = next - } - } - return new String(chars) - } - - static class PojoSpan implements CoreSpan { - - private final CharSequence serviceName - private final CharSequence operationName - private final CharSequence resourceName - private final DDTraceId traceId - private final long spanId - private final long parentId - private final long start - private final long duration - private final int error - private final String type - private final boolean measured - private final Metadata metadata - - PojoSpan( - String serviceName, - String operationName, - CharSequence resourceName, - DDTraceId traceId, - long spanId, - long parentId, - long start, - long duration, - int error, - Map baggage, - Map tags, - String type, - boolean measured) { - this.serviceName = UTF8BytesString.create(serviceName) - this.operationName = UTF8BytesString.create(operationName) - this.resourceName = UTF8BytesString.create(resourceName) - this.traceId = traceId - this.spanId = spanId - this.parentId = parentId - this.start = start - this.duration = duration - this.error = error - this.type = type - this.measured = measured - this.metadata = new Metadata(currentThread().getId(), - UTF8BytesString.create(currentThread().getName()), fromMap(tags), baggage, UNSET, measured, topLevel, null, null, 0, - tagsForSerialization, emptyList()) - } - - @Override - PojoSpan getLocalRootSpan() { - return this - } - - @Override - String getServiceName() { - return serviceName - } - - @Override - CharSequence getServiceNameSource() { - return null - } - - @Override - CharSequence getOperationName() { - return operationName - } - - @Override - CharSequence getResourceName() { - return resourceName - } - - @Override - DDTraceId getTraceId() { - return traceId - } - - @Override - long getSpanId() { - return spanId - } - - @Override - long getParentId() { - return parentId - } - - @Override - long getStartTime() { - return start - } - - @Override - long getDurationNano() { - return duration - } - - @Override - int getError() { - return error - } - - @Override - short getHttpStatusCode() { - return 0 - } - - @Override - CharSequence getOrigin(){ - return null - } - - @Override - PojoSpan setMeasured(boolean measured) { - return this - } - - @Override - PojoSpan setErrorMessage(String errorMessage) { - return this - } - - @Override - PojoSpan addThrowable(Throwable error) { - return this - } - - @Override - PojoSpan setTag(String tag, String value) { - return this - } - - @Override - PojoSpan setTag(String tag, boolean value) { - return this - } - - @Override - PojoSpan setTag(String tag, int value) { - return this - } - - @Override - PojoSpan setTag(String tag, long value) { - return this - } - - @Override - PojoSpan setTag(String tag, double value) { - return this - } - - @Override - PojoSpan setTag(String tag, Number value) { - return this - } - - @Override - PojoSpan setTag(String tag, CharSequence value) { - return this - } - - @Override - PojoSpan setTag(String tag, Object value) { - return this - } - - @Override - PojoSpan removeTag(String tag) { - return this - } - - @Override - boolean isMeasured() { - return measured - } - - @Override - boolean isTopLevel() { - return false - } - - @Override - boolean isForceKeep() { - return false - } - - @Override - boolean isKind(SpanKindFilter filter) { - Object kind = unsafeGetTag(SPAN_KIND) - return filter.matches(kind == null ? null : kind.toString()) - } - - Map getBaggage() { - return metadata.getBaggage() - } - - TagMap getTags() { - return metadata.getTags() - } - - @Override - String getType() { - return type - } - - @Override - void processServiceTags() {} - - @Override - void processTagsAndBaggage(MetadataConsumer consumer) { - consumer.accept(metadata) - } - - @Override - PojoSpan setSamplingPriority(int samplingPriority, int samplingMechanism) { - return this - } - - @Override - PojoSpan setSamplingPriority(int samplingPriority, CharSequence rate, double sampleRate, int samplingMechanism) { - return this - } - - @Override - PojoSpan setSpanSamplingPriority(double rate, int limit) { - return this - } - - @Override - PojoSpan setMetric(CharSequence name, int value) { - return this - } - - @Override - PojoSpan setMetric(CharSequence name, long value) { - return this - } - - @Override - PojoSpan setMetric(CharSequence name, float value) { - return this - } - - @Override - PojoSpan setMetric(CharSequence name, double value) { - return this - } - - @Override - PojoSpan setFlag(CharSequence name, boolean value) { - return this - } - - @Override - int samplingPriority() { - return UNSET - } - - @Override - U getTag(CharSequence name, U defaultValue) { - U value = getTag(name) - return null == value ? defaultValue : value - } - - @Override - U getTag(CharSequence name) { - // replicate logic here because DDSpanContext has to pretend some of its - // fields are elements of a map for backward compatibility reasons - String tag = String.valueOf(name) - Object value = null - switch (tag) { - case DDTags.THREAD_ID: - value = metadata.getThreadId() - break - case DDTags.THREAD_NAME: - value = metadata.getThreadName() - break - default: - value = tags.get(tag) - } - return value as U - } - - @Override - U unsafeGetTag(CharSequence name, U defaultValue) { - return getTag(name, defaultValue) - } - - @Override - U unsafeGetTag(CharSequence name) { - return getTag(name) - } - - @Override - boolean hasSamplingPriority() { - return false - } - - @Override - Map getMetaStruct() { - return [:] - } - - @Override - PojoSpan setMetaStruct(String field, Object value) { - return this - } - - @Override - int getLongRunningVersion() { - return 0 - } - } -} diff --git a/dd-trace-core/src/traceAgentTest/groovy/TraceMapperRealAgentTest.groovy b/dd-trace-core/src/traceAgentTest/groovy/TraceMapperRealAgentTest.groovy deleted file mode 100644 index 2fec61aad74..00000000000 --- a/dd-trace-core/src/traceAgentTest/groovy/TraceMapperRealAgentTest.groovy +++ /dev/null @@ -1,102 +0,0 @@ -import static TraceGenerator.generateRandomTraces -import static datadog.trace.api.ProtocolVersion.V0_4 -import static datadog.trace.api.ProtocolVersion.V0_5 -import static datadog.trace.api.ProtocolVersion.V1_0 - -import datadog.communication.ddagent.DDAgentFeaturesDiscovery -import datadog.communication.http.OkHttpUtils -import datadog.metrics.api.statsd.StatsDClient -import datadog.metrics.impl.MonitoringImpl -import datadog.trace.api.Config -import datadog.trace.common.writer.PayloadDispatcherImpl -import datadog.trace.common.writer.ddagent.DDAgentApi -import datadog.trace.common.writer.ddagent.DDAgentMapperDiscovery -import datadog.trace.core.CoreSpan -import datadog.trace.core.monitor.HealthMetrics -import java.util.concurrent.TimeUnit -import okhttp3.HttpUrl -import okhttp3.OkHttpClient - -class TraceMapperRealAgentTest extends AbstractTraceAgentTest { - HttpUrl agentUrl - OkHttpClient client - MonitoringImpl monitoring - - def setup() { - agentUrl = HttpUrl.parse(Config.get().getAgentUrl()) - client = OkHttpUtils.buildHttpClient(agentUrl, 30_000) - monitoring = new MonitoringImpl(StatsDClient.NO_OP, 1, TimeUnit.SECONDS) - } - - def "send random traces"() { - setup: - HealthMetrics healthMetrics = Mock(HealthMetrics) - DDAgentFeaturesDiscovery discovery = new DDAgentFeaturesDiscovery(client, monitoring, agentUrl, protocol, true, false) - DDAgentApi api = new DDAgentApi(client, agentUrl, discovery, monitoring, false) - PayloadDispatcherImpl dispatcher = new PayloadDispatcherImpl(new DDAgentMapperDiscovery(discovery), api, healthMetrics, monitoring) - List> traces = generateRandomTraces(traceCount, lowCardinality) - when: - for (List trace : traces) { - dispatcher.addTrace(trace) - } - dispatcher.flush() - then: - 0 * healthMetrics.onFailedSerialize(_, _) - 0 * healthMetrics.onFailedSend(_, _, _) - _ * healthMetrics.onSend(_, _, _) - _ * healthMetrics.onSerialize(_) - _ * healthMetrics.onFailedPublish(_) - 0 * _ - - where: - bufferSize | traceCount | lowCardinality | protocol - 10 << 10 | 0 | true | V1_0 - 10 << 10 | 1 | true | V1_0 - 30 << 10 | 1 | true | V1_0 - 30 << 10 | 2 | true | V1_0 - 10 << 10 | 0 | false | V1_0 - 10 << 10 | 1 | false | V1_0 - 30 << 10 | 1 | false | V1_0 - 30 << 10 | 2 | false | V1_0 - 100 << 10 | 0 | true | V1_0 - 100 << 10 | 1 | true | V1_0 - 100 << 10 | 10 | true | V1_0 - 100 << 10 | 100 | true | V1_0 - 100 << 10 | 0 | false | V1_0 - 100 << 10 | 1 | false | V1_0 - 100 << 10 | 10 | false | V1_0 - 100 << 10 | 100 | false | V1_0 - 10 << 10 | 0 | true | V0_5 - 10 << 10 | 1 | true | V0_5 - 30 << 10 | 1 | true | V0_5 - 30 << 10 | 2 | true | V0_5 - 10 << 10 | 0 | false | V0_5 - 10 << 10 | 1 | false | V0_5 - 30 << 10 | 1 | false | V0_5 - 30 << 10 | 2 | false | V0_5 - 100 << 10 | 0 | true | V0_5 - 100 << 10 | 1 | true | V0_5 - 100 << 10 | 10 | true | V0_5 - 100 << 10 | 100 | true | V0_5 - 100 << 10 | 0 | false | V0_5 - 100 << 10 | 1 | false | V0_5 - 100 << 10 | 10 | false | V0_5 - 100 << 10 | 100 | false | V0_5 - 10 << 10 | 0 | true | V0_4 - 10 << 10 | 1 | true | V0_4 - 30 << 10 | 1 | true | V0_4 - 30 << 10 | 2 | true | V0_4 - 10 << 10 | 0 | false | V0_4 - 10 << 10 | 1 | false | V0_4 - 30 << 10 | 1 | false | V0_4 - 30 << 10 | 2 | false | V0_4 - 100 << 10 | 0 | true | V0_4 - 100 << 10 | 1 | true | V0_4 - 100 << 10 | 10 | true | V0_4 - 100 << 10 | 100 | true | V0_4 - 100 << 10 | 0 | false | V0_4 - 100 << 10 | 1 | false | V0_4 - 100 << 10 | 10 | false | V0_4 - 100 << 10 | 100 | false | V0_4 - } -} diff --git a/dd-trace-core/src/traceAgentTest/java/AbstractTraceAgentTest.java b/dd-trace-core/src/traceAgentTest/java/AbstractTraceAgentTest.java new file mode 100644 index 00000000000..3c30bbe8cfa --- /dev/null +++ b/dd-trace-core/src/traceAgentTest/java/AbstractTraceAgentTest.java @@ -0,0 +1,75 @@ +import static datadog.trace.api.ConfigDefaults.DEFAULT_TRACE_AGENT_PORT; +import static datadog.trace.api.config.TracerConfig.AGENT_HOST; +import static datadog.trace.api.config.TracerConfig.TRACE_AGENT_PORT; +import static datadog.trace.test.junit.utils.config.WithConfigExtension.injectSysConfig; + +import datadog.trace.test.util.DDJavaSpecification; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.startupcheck.MinimumDurationRunningStartupCheckStrategy; + +abstract class AbstractTraceAgentTest extends DDJavaSpecification { + + private static GenericContainer agentContainer; + + @BeforeAll + static void setupSpec() { + // CI will provide us with agent container running along side our build. + // When building locally, however, we need to take matters into our own hands + // and we use 'testcontainers' for this. + if (!"true".equals(System.getenv("CI"))) { + Map env = new HashMap<>(); + env.put("DD_APM_ENABLED", "true"); + env.put("DD_BIND_HOST", "0.0.0.0"); + env.put("DD_API_KEY", "invalid_key_but_this_is_fine"); + env.put("DD_HOSTNAME", "doesnotexist"); + env.put("DD_LOGS_STDOUT", "yes"); + agentContainer = + new GenericContainer<>("datadog/agent:7.40.1") + .withEnv(env) + .withExposedPorts(DEFAULT_TRACE_AGENT_PORT) + .withStartupTimeout(Duration.ofSeconds(120)) + // Apparently we need to sleep for a bit so agent's response + // `{"service:,env:":1}` in rate_by_service. + // This is clearly a race-condition and maybe we should avoid verifying complete + // response + .withStartupCheckStrategy( + new MinimumDurationRunningStartupCheckStrategy(Duration.ofSeconds(10))); + agentContainer.start(); + } + } + + @BeforeEach + void setup() { + injectSysConfig(AGENT_HOST, getAgentContainerHost()); + injectSysConfig(TRACE_AGENT_PORT, getAgentContainerPort()); + } + + static String getAgentContainerHost() { + if (agentContainer != null) { + return agentContainer.getHost(); + } + + return System.getenv("CI_AGENT_HOST"); + } + + static String getAgentContainerPort() { + if (agentContainer != null) { + return String.valueOf(agentContainer.getMappedPort(DEFAULT_TRACE_AGENT_PORT)); + } + + return String.valueOf(DEFAULT_TRACE_AGENT_PORT); + } + + @AfterAll + static void cleanupSpec() { + if (agentContainer != null) { + agentContainer.stop(); + } + } +} diff --git a/dd-trace-core/src/traceAgentTest/java/DDApiIntegrationTest.java b/dd-trace-core/src/traceAgentTest/java/DDApiIntegrationTest.java new file mode 100644 index 00000000000..3dc16208364 --- /dev/null +++ b/dd-trace-core/src/traceAgentTest/java/DDApiIntegrationTest.java @@ -0,0 +1,259 @@ +import static datadog.trace.api.ProtocolVersion.V0_5; +import static datadog.trace.api.ProtocolVersion.V1_0; +import static java.util.stream.Collectors.toList; +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.assertTrue; + +import datadog.communication.ddagent.DDAgentFeaturesDiscovery; +import datadog.communication.http.OkHttpUtils; +import datadog.communication.serialization.ByteBufferConsumer; +import datadog.communication.serialization.FlushingBuffer; +import datadog.communication.serialization.msgpack.MsgPackWriter; +import datadog.metrics.api.statsd.StatsDClient; +import datadog.metrics.impl.MonitoringImpl; +import datadog.trace.api.Config; +import datadog.trace.api.ProtocolVersion; +import datadog.trace.common.writer.ListWriter; +import datadog.trace.common.writer.Payload; +import datadog.trace.common.writer.RemoteApi; +import datadog.trace.common.writer.RemoteResponseListener; +import datadog.trace.common.writer.ddagent.DDAgentApi; +import datadog.trace.common.writer.ddagent.TraceMapper; +import datadog.trace.common.writer.ddagent.TraceMapperV0_4; +import datadog.trace.common.writer.ddagent.TraceMapperV0_5; +import datadog.trace.common.writer.ddagent.TraceMapperV1; +import datadog.trace.core.CoreTracer; +import datadog.trace.core.DDSpan; +import java.io.File; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Stream; +import okhttp3.HttpUrl; +import okhttp3.OkHttpClient; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.tabletest.junit.TableTest; + +class DDApiIntegrationTest extends AbstractTraceAgentTest { + + // Looks like okHttp needs to resolve this, even for connection over socket + static final String SOMEHOST = "datadoghq.com"; + static final int SOMEPORT = 123; + + static Process process; + static File socketPath; + + CoreTracer tracer; + DDSpan span; + + DDAgentFeaturesDiscovery discovery; + DDAgentFeaturesDiscovery udsDiscovery; + DDAgentApi api; + DDAgentApi unixDomainSocketApi; + TraceMapper mapper; + String traceEndpoint; + + AtomicReference endpoint = new AtomicReference<>(null); + AtomicReference>> agentResponse = new AtomicReference<>(null); + + RemoteResponseListener responseListener = + (receivedEndpoint, responseJson) -> { + endpoint.set(receivedEndpoint); + agentResponse.set(responseJson); + }; + + @BeforeAll + static void startSocatProxy() throws IOException { + File tmpDir = Files.createTempDirectory("dd-api-integration-test").toFile(); + tmpDir.deleteOnExit(); + socketPath = new File(tmpDir, "socket"); + System.out.println( + "!!!socat UNIX-LISTEN:" + + socketPath + + ",reuseaddr,fork TCP-CONNECT:" + + getAgentContainerHost() + + ":" + + getAgentContainerPort()); + process = + Runtime.getRuntime() + .exec( + "socat UNIX-LISTEN:" + + socketPath + + ",reuseaddr,fork TCP-CONNECT:" + + getAgentContainerHost() + + ":" + + getAgentContainerPort()); + } + + @BeforeEach + void initTracer() throws InterruptedException { + tracer = CoreTracer.builder().writer(new ListWriter()).build(); + span = (DDSpan) tracer.buildSpan("datadog", "fakeOperation").start(); + Thread.sleep(1); + span.finish(); + } + + @AfterEach + void cleanup() { + if (tracer != null) { + tracer.close(); + } + } + + @AfterAll + static void stopSocatProxy() { + if (process != null) { + process.destroy(); + } + } + + void beforeTest(ProtocolVersion protocol) { + MonitoringImpl monitoring = new MonitoringImpl(StatsDClient.NO_OP, 1, TimeUnit.SECONDS); + HttpUrl agentUrl = HttpUrl.get(Config.get().getAgentUrl()); + OkHttpClient httpClient = OkHttpUtils.buildHttpClient(agentUrl, 5000); + discovery = + new DDAgentFeaturesDiscovery(httpClient, monitoring, agentUrl, protocol, true, false); + api = new DDAgentApi(httpClient, agentUrl, discovery, monitoring, false); + api.addResponseListener(responseListener); + HttpUrl udsAgentUrl = HttpUrl.get(String.format("http://%s:%d", SOMEHOST, SOMEPORT)); + OkHttpClient udsClient = OkHttpUtils.buildHttpClient(true, socketPath.toString(), null, 5000); + udsDiscovery = + new DDAgentFeaturesDiscovery(udsClient, monitoring, agentUrl, protocol, true, false); + unixDomainSocketApi = new DDAgentApi(udsClient, udsAgentUrl, udsDiscovery, monitoring, false); + unixDomainSocketApi.addResponseListener(responseListener); + if (protocol == V1_0) { + mapper = new TraceMapperV1(); + } else if (protocol == V0_5) { + mapper = new TraceMapperV0_5(); + } else { + mapper = new TraceMapperV0_4(); + } + traceEndpoint = protocol.endpoint(); + } + + @TableTest({ + "scenario | traceCount | protocol", + "empty traces | 0 | V0_5 ", + "16 empty traces | 16 | V0_5 ", + "empty traces | 0 | V0_4 ", + "16 empty traces | 16 | V0_4 " + }) + void sendingEmptyTracesSucceeds(int traceCount, ProtocolVersion protocol) throws IOException { + beforeTest(protocol); + + RemoteApi.Response response = + api.sendSerializedTraces(prepareRequest(emptyTraces(traceCount), mapper)); + assertFalse(response.response().isEmpty()); + assertFalse(response.exception().isPresent()); + assertTrue(response.status().isPresent()); + assertEquals(200, response.status().getAsInt()); + assertTrue(response.success()); + assertEquals(traceEndpoint, discovery.getTraceEndpoint()); + assertEquals(Config.get().getAgentUrl() + "/" + traceEndpoint, endpoint.get()); + assertInstanceOf(Map.class, agentResponse.get().get("rate_by_service")); + } + + // spotless:off + @TableTest({ + "scenario | protocol", + "V0_5 | V0_5 ", + "V0_4 | V0_4 " + }) + // spotless:on + void sendingTracesSucceeds(ProtocolVersion protocol) throws IOException { + beforeTest(protocol); + + RemoteApi.Response response = + api.sendSerializedTraces( + prepareRequest(Collections.singletonList(Collections.singletonList(span)), mapper)); + assertFalse(response.response().isEmpty()); + assertFalse(response.exception().isPresent()); + assertTrue(response.status().isPresent()); + assertEquals(200, response.status().getAsInt()); + assertTrue(response.success()); + assertEquals(traceEndpoint, discovery.getTraceEndpoint()); + assertEquals(Config.get().getAgentUrl() + "/" + traceEndpoint, endpoint.get()); + assertInstanceOf(Map.class, agentResponse.get().get("rate_by_service")); + } + + // spotless:off + @TableTest({ + "scenario | protocol", + "empty traces | V0_5 ", + "empty traces | V0_4 " + }) + // spotless:on + void sendingEmptyTracesToUnixDomainSocketSucceeds(ProtocolVersion protocol) throws IOException { + beforeTest(protocol); + + RemoteApi.Response response = + unixDomainSocketApi.sendSerializedTraces( + prepareRequest(Collections.>emptyList(), mapper)); + assertFalse(response.response().isEmpty()); + assertFalse(response.exception().isPresent()); + assertTrue(response.status().isPresent()); + assertEquals(200, response.status().getAsInt()); + assertTrue(response.success()); + assertEquals(traceEndpoint, udsDiscovery.getTraceEndpoint()); + assertEquals("http://" + SOMEHOST + ":" + SOMEPORT + "/" + traceEndpoint, endpoint.get()); + assertInstanceOf(Map.class, agentResponse.get().get("rate_by_service")); + } + + // spotless:off + @TableTest({ + "scenario | protocol", + "V0_5 | V0_5 ", + "V0_4 | V0_4 " + }) + // spotless:on + void sendingTracesToUnixDomainSocketSucceeds(ProtocolVersion protocol) throws IOException { + beforeTest(protocol); + + RemoteApi.Response response = + unixDomainSocketApi.sendSerializedTraces( + prepareRequest(Collections.singletonList(Collections.singletonList(span)), mapper)); + assertFalse(response.response().isEmpty()); + assertFalse(response.exception().isPresent()); + assertTrue(response.status().isPresent()); + assertEquals(200, response.status().getAsInt()); + assertTrue(response.success()); + assertEquals(traceEndpoint, udsDiscovery.getTraceEndpoint()); + assertEquals("http://" + SOMEHOST + ":" + SOMEPORT + "/" + traceEndpoint, endpoint.get()); + assertInstanceOf(Map.class, agentResponse.get().get("rate_by_service")); + } + + private static List> emptyTraces(int count) { + return Stream.generate(Collections::emptyList).limit(count).collect(toList()); + } + + Payload prepareRequest(List> traces, TraceMapper traceMapper) throws IOException { + Traces traceCapture = new Traces(); + MsgPackWriter packer = new MsgPackWriter(new FlushingBuffer(1 << 10, traceCapture)); + for (List trace : traces) { + packer.format(trace, traceMapper); + } + packer.flush(); + return traceMapper.newPayload().withBody(traceCapture.traceCount, traceCapture.buffer); + } + + static class Traces implements ByteBufferConsumer { + int traceCount; + ByteBuffer buffer; + + @Override + public void accept(int messageCount, ByteBuffer buffer) { + this.buffer = buffer; + this.traceCount = messageCount; + } + } +} diff --git a/dd-trace-core/src/traceAgentTest/java/DataStreamsIntegrationTest.java b/dd-trace-core/src/traceAgentTest/java/DataStreamsIntegrationTest.java new file mode 100644 index 00000000000..f5f3628debd --- /dev/null +++ b/dd-trace-core/src/traceAgentTest/java/DataStreamsIntegrationTest.java @@ -0,0 +1,103 @@ +import static datadog.trace.common.metrics.EventListener.EventType.OK; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import datadog.communication.ddagent.DDAgentFeaturesDiscovery; +import datadog.communication.ddagent.SharedCommunicationObjects; +import datadog.communication.http.OkHttpUtils; +import datadog.trace.api.Config; +import datadog.trace.api.TraceConfig; +import datadog.trace.api.datastreams.DataStreamsTags; +import datadog.trace.api.datastreams.StatsPoint; +import datadog.trace.api.time.ControllableTimeSource; +import datadog.trace.common.metrics.EventListener; +import datadog.trace.common.metrics.OkHttpSink; +import datadog.trace.core.datastreams.DefaultDataStreamsMonitoring; +import java.lang.reflect.Method; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import okhttp3.HttpUrl; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +@Disabled( + "The agent in CI doesn't have a valid API key. Unlike metrics and traces, data streams fails in this case") +class DataStreamsIntegrationTest extends AbstractTraceAgentTest { + + @Test + void sendingStatsBucketToAgentShouldNotifyWithOkEvent() throws ReflectiveOperationException { + SharedCommunicationObjects sharedCommunicationObjects = new SharedCommunicationObjects(); + sharedCommunicationObjects.createRemaining(Config.get()); + + OkHttpSink sink = + new OkHttpSink( + OkHttpUtils.buildHttpClient(HttpUrl.parse(Config.get().getAgentUrl()), 5000L), + Config.get().getAgentUrl(), + DDAgentFeaturesDiscovery.V01_DATASTREAMS_ENDPOINT, + false, + true, + Collections.emptyMap()); + + BlockingListener listener = new BlockingListener(); + sink.register(listener); + + ControllableTimeSource timeSource = new ControllableTimeSource(); + + TraceConfig traceConfig = mock(TraceConfig.class); + when(traceConfig.isDataStreamsEnabled()).thenReturn(true); + + try (DefaultDataStreamsMonitoring dataStreams = + new DefaultDataStreamsMonitoring( + sink, + sharedCommunicationObjects.featuresDiscovery(Config.get()), + timeSource, + () -> traceConfig, + Config.get())) { + + dataStreams.start(); + DataStreamsTags tags = + DataStreamsTags.create("testType", null, "testTopic", "testGroup", null); + dataStreams.add( + new StatsPoint(tags, 1, 2, 5, timeSource.getCurrentTimeNanos(), 0, 0, 0, null)); + timeSource.advance(Config.get().getDataStreamsBucketDurationNanoseconds()); + invokeReport(dataStreams); + + assertTrue(sharedCommunicationObjects.featuresDiscovery(Config.get()).supportsDataStreams()); + // conditions.eventually { assert listener.events.size() == 1 } + waitForEvents(listener, 1); + assertEquals(OK, listener.events.get(0)); + } + } + + private static void waitForEvents(BlockingListener listener, int expectedCount) { + long deadline = System.currentTimeMillis() + 1000; + while (System.currentTimeMillis() < deadline && listener.events.size() < expectedCount) { + try { + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } + assertEquals(expectedCount, listener.events.size()); + } + + private static void invokeReport(DefaultDataStreamsMonitoring dataStreams) + throws ReflectiveOperationException { + Method report = DefaultDataStreamsMonitoring.class.getDeclaredMethod("report"); + report.setAccessible(true); + report.invoke(dataStreams); + } + + static class BlockingListener implements EventListener { + List events = new CopyOnWriteArrayList<>(); + + @Override + public void onEvent(EventType eventType, String message) { + events.add(eventType); + } + } +} diff --git a/dd-trace-core/src/traceAgentTest/java/TraceGenerator.java b/dd-trace-core/src/traceAgentTest/java/TraceGenerator.java new file mode 100644 index 00000000000..b349e937e68 --- /dev/null +++ b/dd-trace-core/src/traceAgentTest/java/TraceGenerator.java @@ -0,0 +1,458 @@ +import static datadog.trace.api.ProcessTags.getTagsForSerialization; +import static datadog.trace.api.TagMap.fromMap; +import static datadog.trace.api.sampling.PrioritySampling.UNSET; +import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND; +import static java.lang.Thread.currentThread; +import static java.util.Collections.emptyList; +import static java.util.Collections.emptyMap; + +import datadog.trace.api.DDSpanId; +import datadog.trace.api.DDTags; +import datadog.trace.api.DDTraceId; +import datadog.trace.api.IdGenerationStrategy; +import datadog.trace.api.TagMap; +import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString; +import datadog.trace.core.CoreSpan; +import datadog.trace.core.Metadata; +import datadog.trace.core.MetadataConsumer; +import datadog.trace.core.SpanKindFilter; +import java.math.BigDecimal; +import java.math.BigInteger; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; + +class TraceGenerator { + + static List> generateRandomTraces(int howMany, boolean lowCardinality) { + List> traces = new ArrayList<>(howMany); + for (int i = 0; i < howMany; ++i) { + int traceSize = ThreadLocalRandom.current().nextInt(2, 20); + traces.add(generateRandomTrace(traceSize, lowCardinality)); + } + return traces; + } + + private static List generateRandomTrace(int size, boolean lowCardinality) { + List trace = new ArrayList<>(size); + long traceId = ThreadLocalRandom.current().nextLong(1, Long.MAX_VALUE); + for (int i = 0; i < size; ++i) { + trace.add(randomSpan(traceId, lowCardinality)); + } + return trace; + } + + private static final IdGenerationStrategy ID_GENERATION_STRATEGY = + IdGenerationStrategy.fromName("RANDOM"); + + private static CoreSpan randomSpan(long traceId, boolean lowCardinality) { + ThreadLocalRandom random = ThreadLocalRandom.current(); + Map baggage = new HashMap<>(); + if (random.nextBoolean()) { + baggage.put("baggage-key", lowCardinality ? "x" : randomString(100)); + if (random.nextBoolean()) { + baggage.put("tag.1", "bar"); + baggage.put("tag.2", "qux"); + } + } + Map tags = new HashMap<>(); + int tagCount = random.nextInt(0, 20); + for (int i = 0; i < tagCount; ++i) { + tags.put("tag." + i, random.nextBoolean() ? "foo" : randomString(2000)); + tags.put("tag.1." + i, lowCardinality ? "y" : UUID.randomUUID()); + switch (random.nextInt(8)) { + case 0: + tags.put("tag.3." + i, BigDecimal.valueOf(random.nextDouble())); + break; + case 1: + tags.put("tag.3." + i, BigInteger.valueOf(random.nextLong())); + break; + default: + break; + } + } + int metricCount = random.nextInt(0, 20); + for (int i = 0; i < metricCount; ++i) { + String name = "metric." + i; + Number metric = null; + switch (random.nextInt(4)) { + case 0: + metric = random.nextInt(); + break; + case 1: + metric = random.nextLong(); + break; + case 2: + metric = random.nextFloat(); + break; + case 3: + metric = random.nextDouble(); + break; + } + tags.put(name, metric); + } + return new PojoSpan( + "service-" + random.nextInt(lowCardinality ? 1 : 10), + "operation-" + random.nextInt(lowCardinality ? 1 : 100), + UTF8BytesString.create("resource-" + random.nextInt(lowCardinality ? 1 : 100)), + DDTraceId.from(traceId), + ID_GENERATION_STRATEGY.generateSpanId(), + DDSpanId.ZERO, + TimeUnit.MILLISECONDS.toNanos(System.currentTimeMillis()), + random.nextLong(500, 10_000_000), + random.nextInt(2), + baggage, + tags, + "type-" + random.nextInt(lowCardinality ? 1 : 100), + random.nextBoolean()); + } + + private static String randomString(int maxLength) { + char[] chars = new char[ThreadLocalRandom.current().nextInt(maxLength)]; + for (int i = 0; i < chars.length; ++i) { + char next = (char) ThreadLocalRandom.current().nextInt((int) Character.MAX_VALUE); + if (Character.isSurrogate(next)) { + if (i < chars.length - 1) { + chars[i++] = '\uD801'; + chars[i] = '\uDC01'; + } else { + chars[i] = 'a'; + } + } else { + chars[i] = next; + } + } + return new String(chars); + } + + static class PojoSpan implements CoreSpan { + + private final CharSequence serviceName; + private final CharSequence operationName; + private final CharSequence resourceName; + private final DDTraceId traceId; + private final long spanId; + private final long parentId; + private final long start; + private final long duration; + private final int error; + private final String type; + private final boolean measured; + private final Metadata metadata; + + PojoSpan( + String serviceName, + String operationName, + CharSequence resourceName, + DDTraceId traceId, + long spanId, + long parentId, + long start, + long duration, + int error, + Map baggage, + Map tags, + String type, + boolean measured) { + this.serviceName = UTF8BytesString.create(serviceName); + this.operationName = UTF8BytesString.create(operationName); + this.resourceName = UTF8BytesString.create(resourceName); + this.traceId = traceId; + this.spanId = spanId; + this.parentId = parentId; + this.start = start; + this.duration = duration; + this.error = error; + this.type = type; + this.measured = measured; + this.metadata = + new Metadata( + currentThread().getId(), + UTF8BytesString.create(currentThread().getName()), + fromMap(tags), + baggage, + UNSET, + measured, + isTopLevel(), + null, + null, + 0, + getTagsForSerialization(), + emptyList()); + } + + @Override + public PojoSpan getLocalRootSpan() { + return this; + } + + @Override + public String getServiceName() { + return serviceName.toString(); + } + + @Override + public CharSequence getServiceNameSource() { + return null; + } + + @Override + public CharSequence getOperationName() { + return operationName; + } + + @Override + public CharSequence getResourceName() { + return resourceName; + } + + @Override + public DDTraceId getTraceId() { + return traceId; + } + + @Override + public long getSpanId() { + return spanId; + } + + @Override + public long getParentId() { + return parentId; + } + + @Override + public long getStartTime() { + return start; + } + + @Override + public long getDurationNano() { + return duration; + } + + @Override + public int getError() { + return error; + } + + @Override + public short getHttpStatusCode() { + return 0; + } + + @Override + public CharSequence getOrigin() { + return null; + } + + @Override + public PojoSpan setMeasured(boolean measured) { + return this; + } + + @Override + public PojoSpan setErrorMessage(String errorMessage) { + return this; + } + + @Override + public PojoSpan addThrowable(Throwable error) { + return this; + } + + @Override + public PojoSpan setTag(String tag, String value) { + return this; + } + + @Override + public PojoSpan setTag(String tag, boolean value) { + return this; + } + + @Override + public PojoSpan setTag(String tag, int value) { + return this; + } + + @Override + public PojoSpan setTag(String tag, long value) { + return this; + } + + @Override + public PojoSpan setTag(String tag, double value) { + return this; + } + + @Override + public PojoSpan setTag(String tag, Number value) { + return this; + } + + @Override + public PojoSpan setTag(String tag, CharSequence value) { + return this; + } + + @Override + public PojoSpan setTag(String tag, Object value) { + return this; + } + + @Override + public PojoSpan removeTag(String tag) { + return this; + } + + @Override + public boolean isMeasured() { + return measured; + } + + @Override + public boolean isTopLevel() { + return false; + } + + @Override + public boolean isForceKeep() { + return false; + } + + @Override + public boolean isKind(SpanKindFilter filter) { + Object kind = unsafeGetTag(SPAN_KIND); + return filter.matches(kind == null ? null : kind.toString()); + } + + Map getBaggage() { + return metadata.getBaggage(); + } + + TagMap getTags() { + return metadata.getTags(); + } + + @Override + public String getType() { + return type; + } + + @Override + public void processServiceTags() {} + + @Override + public void processTagsAndBaggage(MetadataConsumer consumer) { + consumer.accept(metadata); + } + + @Override + public PojoSpan setSamplingPriority(int samplingPriority, int samplingMechanism) { + return this; + } + + @Override + public PojoSpan setSamplingPriority( + int samplingPriority, CharSequence rate, double sampleRate, int samplingMechanism) { + return this; + } + + @Override + public PojoSpan setSpanSamplingPriority(double rate, int limit) { + return this; + } + + @Override + public PojoSpan setMetric(CharSequence name, int value) { + return this; + } + + @Override + public PojoSpan setMetric(CharSequence name, long value) { + return this; + } + + @Override + public PojoSpan setMetric(CharSequence name, float value) { + return this; + } + + @Override + public PojoSpan setMetric(CharSequence name, double value) { + return this; + } + + @Override + public PojoSpan setFlag(CharSequence name, boolean value) { + return this; + } + + @Override + public int samplingPriority() { + return UNSET; + } + + @Override + @SuppressWarnings("unchecked") + public U getTag(CharSequence name, U defaultValue) { + U value = getTag(name); + return null == value ? defaultValue : value; + } + + @Override + @SuppressWarnings("unchecked") + public U getTag(CharSequence name) { + // replicate logic here because DDSpanContext has to pretend some of its + // fields are elements of a map for backward compatibility reasons + String tag = String.valueOf(name); + Object value; + switch (tag) { + case DDTags.THREAD_ID: + value = metadata.getThreadId(); + break; + case DDTags.THREAD_NAME: + value = metadata.getThreadName(); + break; + default: + value = getTags().get(tag); + } + return (U) value; + } + + @Override + public U unsafeGetTag(CharSequence name, U defaultValue) { + return getTag(name, defaultValue); + } + + @Override + public U unsafeGetTag(CharSequence name) { + return getTag(name); + } + + @Override + public boolean hasSamplingPriority() { + return false; + } + + @Override + public Map getMetaStruct() { + return emptyMap(); + } + + @Override + public PojoSpan setMetaStruct(String field, Object value) { + return this; + } + + @Override + public int getLongRunningVersion() { + return 0; + } + } +} diff --git a/dd-trace-core/src/traceAgentTest/java/TraceMapperRealAgentTest.java b/dd-trace-core/src/traceAgentTest/java/TraceMapperRealAgentTest.java new file mode 100644 index 00000000000..3da085b96d6 --- /dev/null +++ b/dd-trace-core/src/traceAgentTest/java/TraceMapperRealAgentTest.java @@ -0,0 +1,113 @@ +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.atLeast; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; + +import datadog.communication.ddagent.DDAgentFeaturesDiscovery; +import datadog.communication.http.OkHttpUtils; +import datadog.metrics.api.statsd.StatsDClient; +import datadog.metrics.impl.MonitoringImpl; +import datadog.trace.api.Config; +import datadog.trace.api.ProtocolVersion; +import datadog.trace.common.writer.PayloadDispatcherImpl; +import datadog.trace.common.writer.ddagent.DDAgentApi; +import datadog.trace.common.writer.ddagent.DDAgentMapperDiscovery; +import datadog.trace.core.CoreSpan; +import datadog.trace.core.monitor.HealthMetrics; +import java.util.List; +import java.util.concurrent.TimeUnit; +import okhttp3.HttpUrl; +import okhttp3.OkHttpClient; +import org.junit.jupiter.api.BeforeEach; +import org.tabletest.junit.TableTest; + +class TraceMapperRealAgentTest extends AbstractTraceAgentTest { + + HttpUrl agentUrl; + OkHttpClient client; + MonitoringImpl monitoring; + + @BeforeEach + void setUpClient() { + agentUrl = HttpUrl.parse(Config.get().getAgentUrl()); + client = OkHttpUtils.buildHttpClient(agentUrl, 30_000); + monitoring = new MonitoringImpl(StatsDClient.NO_OP, 1, TimeUnit.SECONDS); + } + + @TableTest({ + "scenario | traceCount | lowCardinality | protocol", + "1 | 0 | true | V1_0 ", + "2 | 1 | true | V1_0 ", + "3 | 1 | true | V1_0 ", + "4 | 2 | true | V1_0 ", + "5 | 0 | false | V1_0 ", + "6 | 1 | false | V1_0 ", + "7 | 1 | false | V1_0 ", + "8 | 2 | false | V1_0 ", + "9 | 0 | true | V1_0 ", + "10 | 1 | true | V1_0 ", + "11 | 10 | true | V1_0 ", + "12 | 100 | true | V1_0 ", + "13 | 0 | false | V1_0 ", + "14 | 1 | false | V1_0 ", + "15 | 10 | false | V1_0 ", + "16 | 100 | false | V1_0 ", + "17 | 0 | true | V0_5 ", + "18 | 1 | true | V0_5 ", + "19 | 1 | true | V0_5 ", + "20 | 2 | true | V0_5 ", + "21 | 0 | false | V0_5 ", + "22 | 1 | false | V0_5 ", + "23 | 1 | false | V0_5 ", + "24 | 2 | false | V0_5 ", + "25 | 0 | true | V0_5 ", + "26 | 1 | true | V0_5 ", + "27 | 10 | true | V0_5 ", + "28 | 100 | true | V0_5 ", + "29 | 0 | false | V0_5 ", + "30 | 1 | false | V0_5 ", + "31 | 10 | false | V0_5 ", + "32 | 100 | false | V0_5 ", + "33 | 0 | true | V0_4 ", + "34 | 1 | true | V0_4 ", + "35 | 1 | true | V0_4 ", + "36 | 2 | true | V0_4 ", + "37 | 0 | false | V0_4 ", + "38 | 1 | false | V0_4 ", + "39 | 1 | false | V0_4 ", + "40 | 2 | false | V0_4 ", + "41 | 0 | true | V0_4 ", + "42 | 1 | true | V0_4 ", + "43 | 10 | true | V0_4 ", + "44 | 100 | true | V0_4 ", + "45 | 0 | false | V0_4 ", + "46 | 1 | false | V0_4 ", + "47 | 10 | false | V0_4 ", + "48 | 100 | false | V0_4 " + }) + void sendRandomTraces(int traceCount, boolean lowCardinality, ProtocolVersion protocol) { + HealthMetrics healthMetrics = mock(HealthMetrics.class); + DDAgentFeaturesDiscovery discovery = + new DDAgentFeaturesDiscovery(client, monitoring, agentUrl, protocol, true, false); + DDAgentApi api = new DDAgentApi(client, agentUrl, discovery, monitoring, false); + PayloadDispatcherImpl dispatcher = + new PayloadDispatcherImpl( + new DDAgentMapperDiscovery(discovery), api, healthMetrics, monitoring); + List> traces = TraceGenerator.generateRandomTraces(traceCount, lowCardinality); + + for (List trace : traces) { + dispatcher.addTrace((List>) (List) trace); + } + dispatcher.flush(); + + verify(healthMetrics, never()).onFailedSerialize(any(), any()); + verify(healthMetrics, never()).onFailedSend(anyInt(), anyInt(), any()); + verify(healthMetrics, atLeast(0)).onSend(anyInt(), anyInt(), any()); + verify(healthMetrics, atLeast(0)).onSerialize(anyInt()); + verify(healthMetrics, atLeast(0)).onFailedPublish(anyInt(), anyInt()); + verifyNoMoreInteractions(healthMetrics); + } +}