Implement service endpoint support
MODIFY
dd-trace/src/main/java/com/datadoghq/trace/DDTracer.java
dd-trace/src/main/java/com/datadoghq/trace/DDTracer.java
@@ -43,7 +43,7 @@ public class DDTracer extends ThreadLocalActiveSpanSource implements io.opentrac
private final Map<String, List<AbstractDecorator>> spanContextDecorators = new HashMap<>();
private final CodecRegistry registry;
- private final List<Service> services = new ArrayList<>();
+ private final Map<String, Service> services = new HashMap<>();
/** Default constructor, trace/spans are logged, no trace/span dropped */
public DDTracer() {
@@ -151,8 +151,8 @@ public class DDTracer extends ThreadLocalActiveSpanSource implements io.opentrac
* @param service additional service information
*/
public void addServiceInfo(final Service service) {
- services.add(service);
- // Update the write
+ services.put(service.getName(), service);
+ // Update the writer
try {
// We don't bother to send multiple times the list of services at this time
writer.writeServices(services);
@@ -167,7 +167,7 @@ public class DDTracer extends ThreadLocalActiveSpanSource implements io.opentrac
* @return the list of additional service information
*/
@JsonIgnore
- public List<Service> getServiceInfo() {
+ public Map<String, Service> getServiceInfo() {
return services;
}

MODIFY
dd-trace/src/main/java/com/datadoghq/trace/Service.java
dd-trace/src/main/java/com/datadoghq/trace/Service.java
package com.datadoghq.trace;
+import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonValue;
public class Service {
@@ -14,7 +16,7 @@ public class Service {
this.appType = appType;
}
- @JsonProperty("service")
+ @JsonIgnore
public String getName() {
return name;
}
@@ -29,6 +31,20 @@ public class Service {
return appType;
}
+ @Override
+ public String toString() {
+ return "Service{"
+ + "name='"
+ + name
+ + '\''
+ + ", appName='"
+ + appName
+ + '\''
+ + ", appType="
+ + appType
+ + '}';
+ }
+
public enum AppType {
WEB("web"),
DB("db"),
@@ -40,6 +56,7 @@ public class Service {
this.type = type;
}
+ @JsonValue
public String toString() {
return type;
}

MODIFY
dd-trace/src/main/java/com/datadoghq/trace/writer/DDAgentWriter.java
dd-trace/src/main/java/com/datadoghq/trace/writer/DDAgentWriter.java
@@ -5,6 +5,9 @@ import com.datadoghq.trace.Service;
import com.google.auto.service.AutoService;
import java.util.List;
import java.util.concurrent.Callable;
+import java.util.Map;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -44,6 +47,12 @@ public class DDAgentWriter implements Writer {
/** Effective thread pool, where real logic is done */
private final ExecutorService executor = Executors.newSingleThreadExecutor();
+ /** In memory collection of services waiting for departure */
+ private final BlockingQueue<Map<String, Service>> services =
+ new ArrayBlockingQueue<>(MAX_QUEUE_SIZE);
+
+ /** Async worker that posts the spans to the DD agent */
+ private final ExecutorService executor = Executors.newFixedThreadPool(2);
/** The DD agent api */
private final DDApi api;
@@ -81,14 +90,26 @@ public class DDAgentWriter implements Writer {
queueFullReported = false;
}
+ /* (non-Javadoc)
+ * @see com.datadoghq.trace.Writer#writeServices(java.util.List)
+ */
@Override
- public void writeServices(final List<Service> services) {}
+ public void writeServices(final Map<String, Service> services) {
+
+ if (!this.services.offer(services)) {
+ log.warn(
+ "Cannot add a service list to the async queue, queue is full. Queue max size: {}",
+ MAX_QUEUE_SIZE);
+ }
+ }
/* (non-Javadoc)
* @see com.datadoghq.trace.writer.Writer#start()
*/
@Override
public void start() {
+ executor.submit(new SpansSendingTask());
+ executor.submit(new ServicesSendingTask());
scheduledExecutor.scheduleAtFixedRate(
new TracesSendingTask(), 0, FLUSH_TIME_SECONDS, TimeUnit.SECONDS);
}
@@ -157,4 +178,35 @@ public class DDAgentWriter implements Writer {
}
}
}
+
+ /** Infinite tasks blocking until some spans come in the blocking queue. */
+ protected class ServicesSendingTask implements Runnable {
+
+ @Override
+ public void run() {
+ while (true) {
+ try {
+
+ //WAIT until a new service comes
+ final Map<String, Service> payload = DDAgentWriter.this.services.take();
+
+ //SEND the payload to the agent
+ log.debug("Async writer about to write {} services", payload.size());
+ if (api.sendServices(payload)) {
+ log.debug("Async writer just sent {} services", payload.size());
+ } else {
+ log.warn("Failed for Async writer to send {} services", payload.size());
+ }
+
+ } catch (final InterruptedException e) {
+ log.info("Async writer (services) interrupted.");
+
+ //The thread was interrupted, we break the LOOP
+ break;
+ } catch (final Throwable e) {
+ log.error("Unexpected error! Some services may have been dropped.", e);
+ }
+ }
+ }
+ }
}

MODIFY
dd-trace/src/main/java/com/datadoghq/trace/writer/DDApi.java
dd-trace/src/main/java/com/datadoghq/trace/writer/DDApi.java
@@ -2,6 +2,7 @@ package com.datadoghq.trace.writer;
import com.datadoghq.trace.DDBaseSpan;
import com.datadoghq.trace.DDTraceInfo;
+import com.datadoghq.trace.Service;
import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -10,6 +11,7 @@ import java.io.OutputStreamWriter;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.List;
+import java.util.Map;
import lombok.extern.slf4j.Slf4j;
/** The API pointing to a DD agent */
@@ -17,14 +19,17 @@ import lombok.extern.slf4j.Slf4j;
public class DDApi {
private static final String TRACES_ENDPOINT = "/v0.3/traces";
+ private static final String SERVICES_ENDPOINT = "/v0.3/services";
private final String tracesEndpoint;
+ private final String servicesEndpoint;
private final ObjectMapper objectMapper = new ObjectMapper();
private final JsonFactory jsonFactory = objectMapper.getFactory();
public DDApi(final String host, final int port) {
this.tracesEndpoint = "http://" + host + ":" + port + TRACES_ENDPOINT;
+ this.servicesEndpoint = "http://" + host + ":" + port + SERVICES_ENDPOINT;
}
/**
@@ -34,7 +39,7 @@ public class DDApi {
* @return the staus code returned
*/
public boolean sendTraces(final List<List<DDBaseSpan<?>>> traces) {
- final int status = callPUT(traces);
+ final int status = callPUT(tracesEndpoint, traces);
if (status == 200) {
log.debug("Succesfully sent {} traces to the DD agent.", traces.size());
return true;
@@ -44,6 +49,26 @@ public class DDApi {
}
}
+ /**
+ * Send service extra information to the services endpoint
+ *
+ * @param services the services to be sent
+ */
+ public boolean sendServices(final Map<String, Service> services) {
+ if (services == null) {
+ return true;
+ }
+ final int status = callPUT(servicesEndpoint, services);
+ if (status == 200) {
+ log.debug("Succesfully sent {} services to the DD agent.", services.size());
+ return true;
+ } else {
+ log.warn(
+ "Error while sending {} services to the DD agent. Status: {}", services.size(), status);
+ return false;
+ }
+ }
+
/**
* PUT to an endpoint the provided JSON content
*
@@ -51,10 +76,10 @@ public class DDApi {
* @param content
* @return the status code
*/
- private int callPUT(final Object content) {
+ private int callPUT(final String endpoint, final Object content) {
HttpURLConnection httpCon = null;
try {
- httpCon = getHttpURLConnection();
+ httpCon = getHttpURLConnection(endpoint);
} catch (final Exception e) {
log.warn("Error thrown before PUT call to the DD agent.", e);
return -1;
@@ -82,9 +107,9 @@ public class DDApi {
}
}
- private HttpURLConnection getHttpURLConnection() throws IOException {
+ private HttpURLConnection getHttpURLConnection(final String endpoint) throws IOException {
final HttpURLConnection httpCon;
- final URL url = new URL(tracesEndpoint);
+ final URL url = new URL(endpoint);
httpCon = (HttpURLConnection) url.openConnection();
httpCon.setDoOutput(true);
httpCon.setRequestMethod("PUT");

MODIFY
dd-trace/src/main/java/com/datadoghq/trace/writer/ListWriter.java
dd-trace/src/main/java/com/datadoghq/trace/writer/ListWriter.java
@@ -3,8 +3,8 @@ package com.datadoghq.trace.writer;
import com.datadoghq.trace.DDBaseSpan;
import com.datadoghq.trace.Service;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
-import sun.reflect.generics.reflectiveObjects.NotImplementedException;
/** List writer used by tests mostly */
public class ListWriter extends CopyOnWriteArrayList<List<DDBaseSpan<?>>> implements Writer {
@@ -23,8 +23,8 @@ public class ListWriter extends CopyOnWriteArrayList<List<DDBaseSpan<?>>> implem
}
@Override
- public void writeServices(final List<Service> services) {
- throw new NotImplementedException();
+ public void writeServices(final Map<String, Service> services) {
+ throw new UnsupportedOperationException();
}
@Override

MODIFY
dd-trace/src/main/java/com/datadoghq/trace/writer/LoggingWriter.java
dd-trace/src/main/java/com/datadoghq/trace/writer/LoggingWriter.java
@@ -4,6 +4,7 @@ import com.datadoghq.trace.DDBaseSpan;
import com.datadoghq.trace.Service;
import com.google.auto.service.AutoService;
import java.util.List;
+import java.util.Map;
import lombok.extern.slf4j.Slf4j;
@Slf4j
@@ -16,8 +17,8 @@ public class LoggingWriter implements Writer {
}
@Override
- public void writeServices(final List<Service> services) {
- log.info("additional service information: {}", services);
+ public void writeServices(final Map<String, Service> services) {
+ log.info("additional service information: {}", services.values());
}
@Override

MODIFY
dd-trace/src/main/java/com/datadoghq/trace/writer/Writer.java
dd-trace/src/main/java/com/datadoghq/trace/writer/Writer.java
@@ -3,6 +3,7 @@ package com.datadoghq.trace.writer;
import com.datadoghq.trace.DDBaseSpan;
import com.datadoghq.trace.Service;
import java.util.List;
+import java.util.Map;
/** A writer is responsible to send collected spans to some place */
public interface Writer {
@@ -19,7 +20,7 @@ public interface Writer {
*
* @param services a list of extra information about services
*/
- void writeServices(List<Service> services);
+ void writeServices(Map<String, Service> services);
/** Start the writer */
void start();

MODIFY
dd-trace/src/test/groovy/com/datadoghq/trace/ServiceTest.groovy
dd-trace/src/test/groovy/com/datadoghq/trace/ServiceTest.groovy
@@ -41,7 +41,7 @@ class ServiceTest extends Specification {
then:
tracer.getServiceInfo().size() == 1
- tracer.getServiceInfo().get(0) == service
+ tracer.getServiceInfo().get("service-name") == service
}
@@ -56,7 +56,7 @@ class ServiceTest extends Specification {
tracer.addServiceInfo(new Service("service-name", "app-name", Service.AppType.CUSTOM))
then:
- Mockito.verify(writer, Mockito.times(1)).writeServices(Mockito.any(List.class))
+ Mockito.verify(writer, Mockito.times(1)).writeServices(Mockito.any(Map.class))
}

MODIFY
dd-trace/src/test/groovy/com/datadoghq/trace/writer/DDApiTest.groovy
dd-trace/src/test/groovy/com/datadoghq/trace/writer/DDApiTest.groovy
package com.datadoghq.trace.writer
+import com.datadoghq.trace.Service
import com.datadoghq.trace.SpanFactory
import com.fasterxml.jackson.core.type.TypeReference
import com.fasterxml.jackson.databind.ObjectMapper
@@ -76,7 +77,7 @@ class DDApiTest extends Specification {
requestHeaders.get().get("Datadog-Meta-Lang") == "java"
requestHeaders.get().get("Datadog-Meta-Lang-Version") == System.getProperty("java.version", "unknown")
requestHeaders.get().get("Datadog-Meta-Tracer-Version") == "unknown"
- convert(requestBody.get()) == expectedRequestBody
+ convertList(requestBody.get()) == expectedRequestBody
cleanup:
agent.close()
@@ -113,7 +114,88 @@ class DDApiTest extends Specification {
])]
}
- static List<TreeMap<String, Object>> convert(String json) {
+ // Services endpoint
+ def "sending an empty map of services returns no errors"() {
+ setup:
+ def agent = ratpack {
+ handlers {
+ put("v0.3/services") {
+ response.status(200).send()
+ }
+ }
+ }
+ def client = new DDApi("localhost", agent.address.port)
+
+ expect:
+ client.sendServices()
+
+ cleanup:
+ agent.close()
+ }
+
+ def "non-200 response results in false returned for services endpoint"() {
+ setup:
+ def agent = ratpack {
+ handlers {
+ put("v0.3/services") {
+ response.status(404).send()
+ }
+ }
+ }
+ def client = new DDApi("localhost", agent.address.port)
+
+ expect:
+ !client.sendServices([:])
+
+ cleanup:
+ agent.close()
+ }
+
+ def "services content is sent as JSON"() {
+ setup:
+ def requestContentType = new AtomicReference<MediaType>()
+ def requestHeaders = new AtomicReference<Headers>()
+ def requestBody = new AtomicReference<String>()
+ def agent = ratpack {
+ handlers {
+ put("v0.3/services") {
+ requestContentType.set(request.contentType)
+ requestHeaders.set(request.headers)
+ request.body.then {
+ requestBody.set(it.text)
+ response.send()
+ }
+ }
+ }
+ }
+ def client = new DDApi("localhost", agent.address.port)
+
+ expect:
+ client.sendServices(services)
+ requestContentType.get().type == APPLICATION_JSON
+ requestHeaders.get().get("Datadog-Meta-Lang") == "java"
+ requestHeaders.get().get("Datadog-Meta-Lang-Version") == System.getProperty("java.version", "unknown")
+ requestHeaders.get().get("Datadog-Meta-Tracer-Version") == "unknown"
+ convertMap(requestBody.get()) == expectedRequestBody
+
+ cleanup:
+ agent.close()
+
+ // Populate thread info dynamically as it is different when run via gradle vs idea.
+ where:
+ services | expectedRequestBody
+ [:] | [:]
+ ["service-name": new Service("service-name", "app-name", Service.AppType.CUSTOM)] | ["service-name": new TreeMap<>([
+ "app" : "app-name",
+ "app_type": "custom"])
+ ]
+ }
+
+ static List<TreeMap<String, Object>> convertList(String json) {
return mapper.readValue(json, new TypeReference<List<TreeMap<String, Object>>>() {})
}
+
+ static TreeMap<String, Object> convertMap(String json) {
+ return mapper.readValue(json, new TypeReference<TreeMap<String, Object>>() {})
+ }
}

Implement service endpoint support