First version of the writer working with the Agent API
MODIFY
src/main/java/com/datadoghq/trace/Writer.java
src/main/java/com/datadoghq/trace/Writer.java
package com.datadoghq.trace;
+import java.util.List;
+
import io.opentracing.Span;
public interface Writer {
- public void write(Span span);
+ /**
+ * Write a trace represented by the entire list of all the finished spans
+ *
+ * @param trace
+ */
+ public void write(List<Span> trace);
public void close();
}

MODIFY
src/main/java/com/datadoghq/trace/writer/impl/DDAgentWriter.java
src/main/java/com/datadoghq/trace/writer/impl/DDAgentWriter.java
@@ -4,6 +4,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Semaphore;
import com.datadoghq.trace.Writer;
@@ -14,31 +15,39 @@ public class DDAgentWriter implements Writer {
protected static final String DEFAULT_HOSTNAME = "localhost";
protected static final int DEFAULT_PORT = 8126;
- protected static final int DEFAULT_MAX_TRACES = 1000;
+ protected static final int DEFAULT_MAX_SPANS = 1000;
protected static final int DEFAULT_BATCH_SIZE = 10;
protected static final int DEFAULT_MAX_SERVICES = 1000;
protected static final long DEFAULT_TIMEOUT = 5000;
- protected final BlockingQueue<Span> commandQueue;
+ private final Semaphore tokens;
+ protected final BlockingQueue<List<Span>> traces;
protected final Thread asyncWriterThread;
+ protected final DDApi api;
+
public DDAgentWriter() {
super();
- commandQueue = new ArrayBlockingQueue<Span>(DEFAULT_MAX_TRACES);
+ tokens = new Semaphore(DEFAULT_MAX_SPANS);
+ traces = new ArrayBlockingQueue<List<Span>>(DEFAULT_MAX_SPANS);
+
+ api = new DDApi(DEFAULT_HOSTNAME, DEFAULT_PORT);
asyncWriterThread = new Thread(new SpansSendingTask(), "dd.DDAgentWriter-SpansSendingTask");
asyncWriterThread.setDaemon(true);
asyncWriterThread.start();
}
- public void write(Span span) {
- try{
+ public void write(List<Span> trace) {
//Try to add a new span in the queue
- commandQueue.add(span);
- }catch(IllegalStateException e){
+ boolean proceed = tokens.tryAcquire(trace.size());
+
+ if(proceed){
+ traces.add(trace);
+ }else{
//It was not possible to add the span the queue is full!
//FIXME proper logging
- System.out.println("Cannot add the following span as the async queue is full: "+span);
+ System.out.println("Cannot add the following trace as the async queue is full: "+trace);
}
}
@@ -47,19 +56,32 @@ public class DDAgentWriter implements Writer {
}
protected class SpansSendingTask implements Runnable {
+
+ protected final List<List<Span>> payload = new ArrayList<List<Span>>();
+
public void run() {
while (true) {
try {
- //Wait until a new span comes
- Span span = commandQueue.take();
+ //WAIT until a new span comes
+ payload.add(traces.take());
//Drain all spans up to a certain batch suze
- List<Span> spans = new ArrayList<Span>();
- spans.add(span);
- commandQueue.drainTo(spans, DEFAULT_BATCH_SIZE);
+ traces.drainTo(payload, DEFAULT_BATCH_SIZE);
+
+ //SEND the payload to the agent
+ api.sendTraces(payload);
+
+ //Compute the number of spans sent
+ int spansCount = 0;
+ for(List<Span> trace:payload){
+ spansCount+=trace.size();
+ }
+
+ //Force garbage collect of the payload
+ payload.clear();
- //Then write to the agent
- System.out.println(spans);
+ //Release the tokens
+ tokens.release(spansCount);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
// FIXME proper logging

MODIFY
src/main/java/com/datadoghq/trace/writer/impl/DDApi.java
src/main/java/com/datadoghq/trace/writer/impl/DDApi.java
@@ -46,7 +46,7 @@ public class DDApi {
return true;
}else{
- //FIXME log status here
+ //FIXME log issue here
return false;
}

First version of the writer working with the Agent API