Adding StatsDMonitor
Adding StatsDMonitor to DDAgentWriter
Modified Writer.createAgentWriter to read the health Monitor configs and construct StatsDMonitor when health metrics are enabled
Updated DDTracerTest to check for use of appropriate Monitor class depending on the Config
MODIFY
dd-trace-ot/dd-trace-ot.gradle
dd-trace-ot/dd-trace-ot.gradle
@@ -33,6 +33,8 @@ dependencies {
compile deps.opentracing
compile group: 'io.opentracing.contrib', name: 'opentracing-tracerresolver', version: '0.1.0'
+ compile group: 'com.datadoghq', name: 'java-dogstatsd-client', version: '2.1.1'
+
compile deps.jackson
compile deps.slf4j
compile group: 'com.squareup.okhttp3', name: 'okhttp', version: '3.11.0' // Last version to support Java7

MODIFY
dd-trace-ot/src/main/java/datadog/trace/common/writer/DDAgentWriter.java
dd-trace-ot/src/main/java/datadog/trace/common/writer/DDAgentWriter.java
@@ -13,7 +13,10 @@ import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
+import com.timgroup.statsd.NonBlockingStatsDClient;
+import com.timgroup.statsd.StatsDClient;
import datadog.opentracing.DDSpan;
+import datadog.opentracing.DDTraceOTInfo;
import datadog.trace.common.util.DaemonThreadFactory;
import java.util.ArrayList;
import java.util.List;
@@ -83,7 +86,8 @@ public class DDAgentWriter implements Writer {
this(api, monitor, DISRUPTOR_BUFFER_SIZE, FLUSH_PAYLOAD_DELAY);
}
- public DDAgentWriter(final DDApi api) {
+ /** Old signature (pre-Monitor) used in tests */
+ private DDAgentWriter(final DDApi api) {
this(api, new NoopMonitor());
}
@@ -122,6 +126,19 @@ public class DDAgentWriter implements Writer {
apiPhaser.register(); // Register on behalf of the scheduled executor thread.
}
+ // Exposing some statistics for consumption by monitors
+ public final long getDisruptorCapacity() {
+ return disruptor.getRingBuffer().getBufferSize();
+ }
+
+ public final long getDisruptorUtilizedCapacity() {
+ return getDisruptorCapacity() - getDisruptorRemainingCapacity();
+ }
+
+ public final long getDisruptorRemainingCapacity() {
+ return disruptor.getRingBuffer().remainingCapacity();
+ }
+
@Override
public void write(final List<DDSpan> trace) {
// We can't add events after shutdown otherwise it will never complete shutting down.
@@ -209,7 +226,18 @@ public class DDAgentWriter implements Writer {
@Override
public String toString() {
- return "DDAgentWriter { api=" + api + " }";
+ // DQH - I don't particularly like the instanceof check,
+ // but I decided it was preferable to adding an isNoop method onto
+ // Monitor or checking the result of Monitor#toString() to determine
+ // if something is *probably* the NoopMonitor.
+
+ String str = "DDAgentWriter { api=" + api;
+ if (!(monitor instanceof NoopMonitor)) {
+ str += ", monitor=" + monitor;
+ }
+ str += " }";
+
+ return str;
}
private void scheduleFlush() {
@@ -428,4 +456,125 @@ public class DDAgentWriter implements Writer {
return "NoOp";
}
}
+
+ public static final class StatsDMonitor implements Monitor {
+ public static final String PREFIX = "datadog.tracer";
+
+ public static final String LANG_TAG = "lang";
+ public static final String LANG_VERSION_TAG = "lang_version";
+ public static final String LANG_INTERPRETER_TAG = "lang_interpreter";
+ public static final String LANG_INTERPRETER_VENDOR_TAG = "lang_interpreter_vendor";
+ public static final String TRACER_VERSION_TAG = "tracer_version";
+
+ private final String host;
+ private final int port;
+ private final StatsDClient statsd;
+
+ // DQH - Made a conscious choice to not take a Config object here.
+ // Letting the creating of the Monitor take the Config,
+ // so it can decide which Monitor variant to create.
+
+ public StatsDMonitor(final String host, final int port) {
+ this.host = host;
+ this.port = port;
+
+ statsd = new NonBlockingStatsDClient(PREFIX, host, port, getDefaultTags());
+ }
+
+ protected static final String[] getDefaultTags() {
+ return new String[] {
+ tag(LANG_TAG, "java"),
+ tag(LANG_VERSION_TAG, DDTraceOTInfo.JAVA_VERSION),
+ tag(LANG_INTERPRETER_TAG, DDTraceOTInfo.JAVA_VM_NAME),
+ tag(LANG_INTERPRETER_VENDOR_TAG, DDTraceOTInfo.JAVA_VM_VENDOR),
+ tag(TRACER_VERSION_TAG, DDTraceOTInfo.VERSION)
+ };
+ }
+
+ private static final String tag(final String tagPrefix, final String tagValue) {
+ return tagPrefix + ":" + tagValue;
+ }
+
+ @Override
+ public void onStart(final DDAgentWriter agentWriter) {
+ statsd.recordGaugeValue("queue.max_length", agentWriter.getDisruptorCapacity());
+ }
+
+ @Override
+ public void onShutdown(final DDAgentWriter agentWriter, final boolean flushSuccess) {}
+
+ @Override
+ public void onPublish(final DDAgentWriter agentWriter, final List<DDSpan> trace) {
+ statsd.incrementCounter("queue.accepted");
+ statsd.count("queue.accepted_lengths", trace.size());
+ }
+
+ @Override
+ public void onFailedPublish(final DDAgentWriter agentWriter, final List<DDSpan> trace) {
+ statsd.incrementCounter("queue.dropped");
+ }
+
+ @Override
+ public void onScheduleFlush(final DDAgentWriter agentWriter, final boolean previousIncomplete) {
+ // not recorded
+ }
+
+ @Override
+ public void onSerialize(
+ final DDAgentWriter agentWriter, final List<DDSpan> trace, final byte[] serializedTrace) {
+ // DQH - Because of Java tracer's 2 phase acceptance and serialization scheme, this doesn't
+ // map precisely
+ statsd.count("queue.accepted_size", serializedTrace.length);
+ }
+
+ @Override
+ public void onFailedSerialize(
+ final DDAgentWriter agentWriter, final List<DDSpan> trace, final Throwable optionalCause) {
+ // TODO - DQH - make a new stat for serialization failure -- or maybe count this towards
+ // api.errors???
+ }
+
+ @Override
+ public void onSend(
+ final DDAgentWriter agentWriter,
+ final int representativeCount,
+ final int sizeInBytes,
+ final DDApi.Response response) {
+ onSendAttempt(agentWriter, representativeCount, sizeInBytes, response);
+ }
+
+ @Override
+ public void onFailedSend(
+ final DDAgentWriter agentWriter,
+ final int representativeCount,
+ final int sizeInBytes,
+ final DDApi.Response response) {
+ onSendAttempt(agentWriter, representativeCount, sizeInBytes, response);
+ }
+
+ private void onSendAttempt(
+ final DDAgentWriter agentWriter,
+ final int representativeCount,
+ final int sizeInBytes,
+ final DDApi.Response response) {
+ statsd.incrementCounter("api.requests");
+ statsd.recordGaugeValue("queue.length", representativeCount);
+ // TODO: missing queue.spans (# of spans being sent)
+ statsd.recordGaugeValue("queue.size", sizeInBytes);
+
+ if (response.exception() != null) {
+ // covers communication errors -- both not receiving a response or
+ // receiving malformed response (even when otherwise successful)
+ statsd.incrementCounter("api.errors");
+ }
+
+ if (response.status() != null) {
+ statsd.incrementCounter("api.responses", "status: " + response.status());
+ }
+ }
+
+ public String toString() {
+ return "StatsD { host=" + host + ":" + port + " }";
+ }
+ }
}

MODIFY
dd-trace-ot/src/main/java/datadog/trace/common/writer/Writer.java
dd-trace-ot/src/main/java/datadog/trace/common/writer/Writer.java
@@ -62,9 +62,33 @@ public interface Writer extends Closeable {
}
private static Writer createAgentWriter(final Config config) {
- return new DDAgentWriter(
- new DDApi(
- config.getAgentHost(), config.getAgentPort(), config.getAgentUnixDomainSocket()));
+ return new DDAgentWriter(createApi(config), createMonitor(config));
+ }
+
+ private static final DDApi createApi(final Config config) {
+ return new DDApi(
+ config.getAgentHost(), config.getAgentPort(), config.getAgentUnixDomainSocket());
+ }
+
+ private static final DDAgentWriter.Monitor createMonitor(final Config config) {
+ if (!config.isHealthMetricsEnabled()) {
+ return new DDAgentWriter.NoopMonitor();
+ } else {
+ String host = config.getHealthMetricsStatsdHost();
+ if (host == null) {
+ host = config.getJmxFetchStatsdHost();
+ }
+ if (host == null) {
+ host = config.getAgentHost();
+ }
+
+ Integer port = config.getHealthMetricsStatsdPort();
+ if (port == null) {
+ port = config.getJmxFetchStatsdPort();
+ }
+
+ return new DDAgentWriter.StatsDMonitor(host, port);
+ }
}
private Builder() {}

MODIFY
dd-trace-ot/src/test/groovy/datadog/trace/DDTracerTest.groovy
dd-trace-ot/src/test/groovy/datadog/trace/DDTracerTest.groovy
@@ -15,6 +15,7 @@ import org.junit.contrib.java.lang.system.RestoreSystemProperties
import static datadog.trace.api.Config.DEFAULT_SERVICE_NAME
import static datadog.trace.api.Config.HEADER_TAGS
+import static datadog.trace.api.Config.HEALTH_METRICS_ENABLED
import static datadog.trace.api.Config.PREFIX
import static datadog.trace.api.Config.PRIORITY_SAMPLING
import static datadog.trace.api.Config.SERVICE_MAPPING
@@ -46,6 +47,7 @@ class DDTracerTest extends DDSpecification {
tracer.serviceName == "unnamed-java-app"
tracer.sampler instanceof RateByServiceSampler
tracer.writer.toString() == "DDAgentWriter { api=DDApi { tracesUrl=http://localhost:8126/v0.3/traces } }"
+ tracer.writer.monitor instanceof DDAgentWriter.NoopMonitor
tracer.spanContextDecorators.size() == 15
@@ -53,6 +55,18 @@ class DDTracerTest extends DDSpecification {
tracer.extractor instanceof HttpCodec.CompoundExtractor
}
+ def "verify enabling health monitor"() {
+ setup:
+ System.setProperty(PREFIX + HEALTH_METRICS_ENABLED, "true")
+
+ when:
+ def tracer = new DDTracer(new Config())
+
+ then:
+ tracer.writer.toString() == "DDAgentWriter { api=DDApi { tracesUrl=http://localhost:8126/v0.3/traces }, monitor=StatsD { host=localhost:8125 } }"
+ tracer.writer.monitor instanceof DDAgentWriter.StatsDMonitor
+ }
+
def "verify overriding sampler"() {
setup:

Adding StatsDMonitor

Adding StatsDMonitor to DDAgentWriter

Modified Writer.createAgentWriter to read the health Monitor configs and construct StatsDMonitor when health metrics are enabled

Updated DDTracerTest to check for use of appropriate Monitor class depending on the Config