New API: add Datadog agent client
This implementation is missing a few minorish things, but should be able to
generally talk to Datadog agent 'traces' endpoint.
Note: this also drops support for 0.3 Datadog Agent.
MODIFY
dd-trace/dd-trace.gradle
dd-trace/dd-trace.gradle
@@ -18,4 +18,5 @@ dependencies {
testCompile deps.bytebuddy
testCompile group: 'org.objenesis', name: 'objenesis', version: '2.6' // Last version to support Java7
testCompile group: 'nl.jqno.equalsverifier', name: 'equalsverifier', version: '2.5.2' // Last version to support Java7
+ testCompile group: 'com.github.tomakehurst', name: 'wiremock', 'version': '2.20.0'
}

ADD
dd-trace/src/main/java/datadog/trace/tracer/writer/AgentClient.java
+package datadog.trace.tracer.writer;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonValue;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import datadog.trace.tracer.LogRateLimiter;
+import datadog.trace.tracer.Trace;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.msgpack.jackson.dataformat.MessagePackFactory;
+
+@Slf4j
+class AgentClient {
+
+ static final String TRACES_ENDPOINT = "/v0.4/traces";
+
+ static final String CONTENT_TYPE = "Content-Type";
+ static final String MSGPACK = "application/msgpack";
+ static final String DATADOG_META_LANG = "Datadog-Meta-Lang";
+ static final String DATADOG_META_LANG_VERSION = "Datadog-Meta-Lang-Version";
+ static final String DATADOG_META_LANG_INTERPRETER = "Datadog-Meta-Lang-Interpreter";
+ static final String DATADOG_META_TRACER_VERSION = "Datadog-Meta-Tracer-Version";
+ static final String X_DATADOG_TRACE_COUNT = "X-Datadog-Trace-Count";
+
+ private static final long MILLISECONDS_BETWEEN_ERROR_LOG = TimeUnit.MINUTES.toMillis(5);
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(new MessagePackFactory());
+ private static final LogRateLimiter LOG_RATE_LIMITER =
+ new LogRateLimiter(log, MILLISECONDS_BETWEEN_ERROR_LOG);
+
+ private final URL tracesUrl;
+
+ AgentClient(final String host, final int port) {
+ final String url = "http://" + host + ":" + port + TRACES_ENDPOINT;
+ try {
+ tracesUrl = new URL(url);
+ } catch (final MalformedURLException e) {
+ // This should essentially mean agent should bail out from installing, we cannot meaningfully
+ // recover from this.
+ throw new RuntimeException("Cannot parse agent url: " + url, e);
+ }
+ }
+
+ /**
+ * Send traces to the Datadog agent
+ *
+ * @param traces the traces to be sent
+ * @param traceCount total number of traces
+ */
+ public SampleRateByService sendTraces(final List<Trace> traces, final int traceCount) {
+ final TracesRequest request = new TracesRequest(traces);
+ try {
+ final HttpURLConnection connection = createHttpConnection();
+ connection.setRequestProperty(X_DATADOG_TRACE_COUNT, String.valueOf(traceCount));
+
+ try (final OutputStream out = connection.getOutputStream()) {
+ OBJECT_MAPPER.writeValue(out, request);
+ }
+
+ final int responseCode = connection.getResponseCode();
+ if (responseCode != 200) {
+ throw new IOException(
+ String.format(
+ "Error while sending %d of %d traces to the DD agent. Status: %d, ResponseMessage: %s",
+ traces.size(), traceCount, responseCode, connection.getResponseMessage()));
+ }
+
+ try (final InputStream in = connection.getInputStream()) {
+ final TracesResponse response = OBJECT_MAPPER.readValue(in, TracesResponse.class);
+ return response.getSampleRateByService();
+ }
+ } catch (final IOException e) {
+ LOG_RATE_LIMITER.warn(
+ "Error while sending {} of {} traces to the DD agent.", traces.size(), traceCount, e);
+ }
+ return null;
+ }
+
+ private HttpURLConnection createHttpConnection() throws IOException {
+ final HttpURLConnection connection = (HttpURLConnection) tracesUrl.openConnection();
+ connection.setDoOutput(true);
+ connection.setDoInput(true);
+
+ connection.setRequestMethod("PUT");
+ connection.setRequestProperty(CONTENT_TYPE, MSGPACK);
+ connection.setRequestProperty(DATADOG_META_LANG, "java");
+
+ // TODO: set these variables properly!!!
+ connection.setRequestProperty(DATADOG_META_LANG_VERSION, "TODO: DDTraceOTInfo.JAVA_VERSION");
+ connection.setRequestProperty(
+ DATADOG_META_LANG_INTERPRETER, "TODO: DDTraceOTInfo.JAVA_VM_NAME");
+ connection.setRequestProperty(DATADOG_META_TRACER_VERSION, "TODO: DDTraceOTInfo.VERSION");
+
+ return connection;
+ }
+
+ private static class TracesRequest {
+
+ private final List<Trace> traces;
+
+ TracesRequest(final List<Trace> traces) {
+ this.traces = Collections.unmodifiableList(traces);
+ }
+
+ @JsonValue
+ public List<Trace> getTraces() {
+ return traces;
+ }
+ }
+
+ @JsonIgnoreProperties(ignoreUnknown = true)
+ private static class TracesResponse {
+
+ private final SampleRateByService sampleRateByService;
+
+ @JsonCreator
+ TracesResponse(@JsonProperty("rate_by_service") final SampleRateByService sampleRateByService) {
+ this.sampleRateByService = sampleRateByService;
+ }
+
+ public SampleRateByService getSampleRateByService() {
+ return sampleRateByService;
+ }
+ }
+}

ADD
dd-trace/src/main/java/datadog/trace/tracer/writer/SampleRateByService.java
+package datadog.trace.tracer.writer;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import java.util.Collections;
+import java.util.Map;
+import lombok.EqualsAndHashCode;
+
+/**
+ * Holds sample rate for all known services. This is reported by Dadadog agent in response to
+ * writing traces.
+ */
+@EqualsAndHashCode
+class SampleRateByService {
+
+ private final Map<String, Double> rateByService;
+
+ @JsonCreator
+ SampleRateByService(final Map<String, Double> rateByService) {
+ this.rateByService = Collections.unmodifiableMap(rateByService);
+ }
+
+ public Double getRate(final String service) {
+ // TODO: improve logic in this class to handle default value better
+ return rateByService.get(service);
+ }
+}

ADD
dd-trace/src/test/groovy/datadog/trace/tracer/writer/AgentClientTest.groovy
+package datadog.trace.tracer.writer
+
+import com.fasterxml.jackson.core.type.TypeReference
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.github.tomakehurst.wiremock.http.Request
+import com.github.tomakehurst.wiremock.junit.WireMockRule
+import com.github.tomakehurst.wiremock.matching.MatchResult
+import datadog.trace.tracer.Clock
+import datadog.trace.tracer.JsonSpan
+import datadog.trace.tracer.SpanContextImpl
+import datadog.trace.tracer.Trace
+import datadog.trace.tracer.TraceImpl
+import datadog.trace.tracer.Tracer
+import datadog.trace.tracer.sampling.Sampler
+import org.junit.Rule
+import org.msgpack.jackson.dataformat.MessagePackFactory
+import spock.lang.Specification
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse
+import static com.github.tomakehurst.wiremock.client.WireMock.equalTo
+import static com.github.tomakehurst.wiremock.client.WireMock.put
+import static com.github.tomakehurst.wiremock.client.WireMock.putRequestedFor
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo
+import static com.github.tomakehurst.wiremock.client.WireMock.verify
+import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig
+
+class AgentClientTest extends Specification {
+
+ private static final int TRACE_COUNT = 10
+ private static final String SERVICE_NAME = "service.name"
+
+ private final ObjectMapper objectMapper = new ObjectMapper(new MessagePackFactory())
+
+ def writer = Mock(Writer)
+ def sampler = Mock(Sampler) {
+ sample(_) >> true
+ }
+ def tracer = Mock(Tracer) {
+ getDefaultServiceName() >> SERVICE_NAME
+ getInterceptors() >> []
+ getWriter() >> writer
+ getSampler() >> sampler
+ }
+
+ @Rule
+ public WireMockRule wireMockRule = new WireMockRule(wireMockConfig().dynamicPort().dynamicHttpsPort())
+
+ private AgentClient client
+
+ def setup() {
+ client = new AgentClient("localhost", wireMockRule.port())
+
+ def response = ["rate_by_service": ["test": 0.1, "another test": 0.2]]
+
+ stubFor(put(urlEqualTo(AgentClient.TRACES_ENDPOINT))
+ .willReturn(aResponse()
+ .withStatus(200)
+ .withHeader("Content-Type", "application/msgpack")
+ .withBody(objectMapper.writeValueAsBytes(response))))
+ }
+
+ def "test send traces"() {
+ setup:
+ def traces = [createTrace("123"), createTrace("312")]
+
+ when:
+ def response = client.sendTraces(traces, TRACE_COUNT)
+
+ then: "got expected response"
+ response.getRate("test") == 0.1d
+ response.getRate("another test") == 0.2d
+ response.getRate("doesn't exist") == null
+ and: "request got expected parameters"
+ verify(putRequestedFor(urlEqualTo(AgentClient.TRACES_ENDPOINT))
+ .withHeader(AgentClient.CONTENT_TYPE, equalTo(AgentClient.MSGPACK))
+ .withHeader(AgentClient.DATADOG_META_LANG, equalTo("java"))
+ // TODO: fill in these headers
+ // .withHeader(AgentClient.DATADOG_META_LANG_VERSION, equalTo("java"))
+ // .withHeader(AgentClient.DATADOG_META_LANG_INTERPRETER, equalTo("java"))
+ // .withHeader(AgentClient.DATADOG_META_TRACER_VERSION, equalTo("java"))
+ .withHeader(AgentClient.X_DATADOG_TRACE_COUNT, equalTo(Integer.toString(TRACE_COUNT)))
+ .andMatching({ Request request ->
+ // Note: it is hard to see what's wrong when this fails... is there a better way?
+ MatchResult.of(objectMapper.readValue(request.getBody(), new TypeReference<List<List<JsonSpan>>>() {}) == traces.collect {it.getSpans().collect {new JsonSpan(it)}})
+ }))
+ }
+
+ def "test send empty list"() {
+ when:
+ def response = client.sendTraces([], TRACE_COUNT)
+
+ then: "got expected response"
+ response.getRate("test") == 0.1d
+ response.getRate("another test") == 0.2d
+ response.getRate("doesn't exist") == null
+ and: "request got expected parameters"
+ verify(putRequestedFor(urlEqualTo(AgentClient.TRACES_ENDPOINT))
+ .withHeader(AgentClient.CONTENT_TYPE, equalTo(AgentClient.MSGPACK))
+ .withHeader(AgentClient.DATADOG_META_LANG, equalTo("java"))
+ // TODO: fill in these headers
+ // .withHeader(AgentClient.DATADOG_META_LANG_VERSION, equalTo("java"))
+ // .withHeader(AgentClient.DATADOG_META_LANG_INTERPRETER, equalTo("java"))
+ // .withHeader(AgentClient.DATADOG_META_TRACER_VERSION, equalTo("java"))
+ .withHeader(AgentClient.X_DATADOG_TRACE_COUNT, equalTo(Integer.toString(TRACE_COUNT)))
+ .andMatching({ Request request ->
+ MatchResult.of(objectMapper.readValue(request.getBody(), new TypeReference<List<List<JsonSpan>>>() {}) == [])
+ }))
+ }
+
+ def "test failure"() {
+ setup:
+ stubFor(put(urlEqualTo(AgentClient.TRACES_ENDPOINT))
+ .willReturn(aResponse()
+ .withStatus(500)))
+ def trace = createTrace("123")
+
+ when:
+ def response = client.sendTraces([trace], TRACE_COUNT)
+
+ then:
+ response == null
+ }
+
+
+ def "test invalid url"() {
+ when:
+ client = new AgentClient("localhost", -100)
+
+ then:
+ thrown RuntimeException
+ }
+
+ Trace createTrace(String traceId) {
+ def clock = new Clock(tracer)
+ def parentContext = new SpanContextImpl("123", "456", "789")
+ def trace = new TraceImpl(tracer, parentContext, clock.createCurrentTimestamp())
+ trace.getRootSpan().setResource("test resource")
+ trace.getRootSpan().setType("test type")
+ trace.getRootSpan().setName("test name")
+ trace.getRootSpan().setMeta("number.key", 123)
+ trace.getRootSpan().setMeta("string.key", "meta string")
+ trace.getRootSpan().setMeta("boolean.key", true)
+
+ def childSpan = trace.createSpan(trace.getRootSpan().getContext())
+ childSpan.setResource("child span test resource")
+ childSpan.setType("child span test type")
+ childSpan.setName("child span test name")
+ childSpan.setMeta("child.span.number.key", 234)
+ childSpan.setMeta("child.span.string.key", "new meta string")
+ childSpan.setMeta("child.span.boolean.key", true)
+ childSpan.finish()
+
+ trace.getRootSpan().finish()
+ return trace
+ }
+
+}

ADD
dd-trace/src/test/groovy/datadog/trace/tracer/writer/SampleRateByServiceTest.groovy
+package datadog.trace.tracer.writer
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import nl.jqno.equalsverifier.EqualsVerifier
+import nl.jqno.equalsverifier.Warning
+import spock.lang.Specification
+
+class SampleRateByServiceTest extends Specification {
+
+ private static final Map<String, Double> TEST_MAP = ["test": 0.1d, "another test": 0.2d]
+
+ ObjectMapper objectMapper = new ObjectMapper()
+
+ def "test constructor and getter"() {
+ when:
+ def sampleRate = new SampleRateByService(TEST_MAP)
+
+ then:
+ sampleRate.getRate("test") == 0.1d
+ sampleRate.getRate("another test") == 0.2d
+ sampleRate.getRate("doesn't exist") == null
+ }
+
+ def "test JSON parsing"() {
+ when:
+ def sampleRate = objectMapper.readValue("{\"test\": 0.8, \"another test\": 0.9}", SampleRateByService)
+
+ then:
+ sampleRate.getRate("test") == 0.8d
+ sampleRate.getRate("another test") == 0.9d
+ }
+
+ def "test equals"() {
+ when:
+ EqualsVerifier.forClass(SampleRateByService).suppress(Warning.STRICT_INHERITANCE).verify()
+
+ then:
+ noExceptionThrown()
+ }
+}

New API: add Datadog agent client

This implementation is missing a few minorish things, but should be able to
generally talk to Datadog agent 'traces' endpoint.

Note: this also drops support for 0.3 Datadog Agent.