Skip to content

Allow telemetry to be sent via regular dogstatsd #323

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vickenty merged 1 commit into from
Aug 6, 2026
Merged

Allow telemetry to be sent via regular dogstatsd #323

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 0 additions & 5 deletions dogstatsd-http/forwarder/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,6 @@
<description>HTTP forwarder for DogStatsD metrics.</description>

<dependencies>
<dependency>
<groupId>com.datadoghq</groupId>
<artifactId>dogstatsd-http-core</artifactId>
<version>1.0.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
Expand Down
80 changes: 45 additions & 35 deletions dogstatsd-http/forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Telemetry.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,11 +7,8 @@

package com.datadoghq.dogstatsd.http.forwarder;

import com.datadoghq.dogstatsd.http.serializer.PayloadBuilder;
import java.time.Clock;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.LongSupplier;

Expand Down Expand Up @@ -56,50 +53,63 @@ public static final class Snapshot {
this.intervalStartMillis = intervalStartMillis;
}

/** Destination for the metrics of an encoded snapshot. */
public interface Encoder {
/**
* Emits a gauge.
*
* @param name Full metric name.
* @param value Metric value.
*/
void gauge(String name, double value);

/**
* Emits a count.
*
* @param name Full metric name.
* @param value Metric value.
* @param tags Tags to attach, or {@code null} for none.
*/
void count(String name, double value, String[] tags);
}

/**
* Encodes this snapshot into {@code pb} using the default metric.
* Encodes this snapshot into {@code enc} using the default metric name prefix.
*
* @param pb Builder to append metrics to.
* @param enc Encoder to emit metrics to.
*/
public void encodeTo(PayloadBuilder pb) {
encodeTo(DEFAULT_PREFIX, pb);
public void encodeTo(Encoder enc) {
encodeTo(DEFAULT_PREFIX, enc);
}

/**
* Encodes this snapshot into {@code pb}.
* Encodes this snapshot into {@code enc}.
*
* <p>Metrics carry no timestamp — the encoder decides how to timestamp them.
*
* @param pb Builder to append metrics to.
* @param prefix Metric name prefix.
* @param enc Encoder to emit metrics to.
*/
public void encodeTo(String prefix, PayloadBuilder pb) {
long ts = intervalStartMillis / 1000;

pb.count(prefix + ".enqueued_payloads").addPoint(ts, enqueuedPayloads).close();
pb.count(prefix + ".enqueued_bytes").addPoint(ts, enqueuedBytes).close();
pb.count(prefix + ".delivered_payloads").addPoint(ts, deliveredPayloads).close();
pb.count(prefix + ".delivered_bytes").addPoint(ts, deliveredBytes).close();
pb.count(prefix + ".dropped_payloads").addPoint(ts, droppedPayloads).close();
pb.count(prefix + ".dropped_bytes").addPoint(ts, droppedBytes).close();

pb.gauge(prefix + ".queue_payloads").addPoint(ts, queuePayloads).close();
pb.gauge(prefix + ".queue_bytes").addPoint(ts, queueBytes).close();
pb.gauge(prefix + ".queue_max_bytes").addPoint(ts, queueMaxBytes).close();

pb.gauge(prefix + ".oldest_enqueued_age_seconds")
.addPoint(ts, oldestEnqueuedAgeNanos / 1e9)
.close();
pb.gauge(prefix + ".last_success_age_seconds")
.addPoint(ts, lastSuccessAgeNanos / 1e9)
.close();
public void encodeTo(String prefix, Encoder enc) {
enc.count(prefix + ".enqueued_payloads", (double) enqueuedPayloads, null);
enc.count(prefix + ".enqueued_bytes", (double) enqueuedBytes, null);
enc.count(prefix + ".delivered_payloads", (double) deliveredPayloads, null);
enc.count(prefix + ".delivered_bytes", (double) deliveredBytes, null);
enc.count(prefix + ".dropped_payloads", (double) droppedPayloads, null);
enc.count(prefix + ".dropped_bytes", (double) droppedBytes, null);

enc.gauge(prefix + ".queue_payloads", (double) queuePayloads);
enc.gauge(prefix + ".queue_bytes", (double) queueBytes);
enc.gauge(prefix + ".queue_max_bytes", (double) queueMaxBytes);

enc.gauge(prefix + ".oldest_enqueued_age_seconds", oldestEnqueuedAgeNanos / 1e9);
enc.gauge(prefix + ".last_success_age_seconds", lastSuccessAgeNanos / 1e9);

for (Map.Entry<String, CodeCounters> e : byCode.entrySet()) {
List<String> tags = Collections.singletonList("code:" + e.getKey());
String[] tags = {"code:" + e.getKey()};
CodeCounters c = e.getValue();
pb.count(prefix + ".response_payloads")
.setTags(tags)
.addPoint(ts, c.payloads)
.close();
pb.count(prefix + ".response_bytes").setTags(tags).addPoint(ts, c.bytes).close();
enc.count(prefix + ".response_payloads", (double) c.payloads, tags);
enc.count(prefix + ".response_bytes", (double) c.bytes, tags);
}
}

Expand Down
87 changes: 44 additions & 43 deletions ...sd-http/forwarder/src/test/java/com/datadoghq/dogstatsd/http/forwarder/TelemetryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,12 +12,12 @@
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;

import com.datadoghq.dogstatsd.http.serializer.PayloadBuilder;
import java.io.ByteArrayOutputStream;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.time.Clock;
import java.time.Instant;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import org.junit.Test;

public class TelemetryTest {
Expand Down Expand Up @@ -174,6 +174,28 @@ public void lastSuccessAge() {
assertEquals(0L, t.snapshot(null).lastSuccessAgeNanos);
}

/** Encoder that records every emitted metric, keyed by name and tags. */
private static final class Recorder implements Telemetry.Snapshot.Encoder {
final Map<String, Double> gauges = new HashMap<>();
final Map<String, Double> counts = new HashMap<>();

@Override
public void gauge(String name, double value) {
gauges.put(name, value);
}

@Override
public void count(String name, double value, String[] tags) {
counts.put(tags == null ? name : name + Arrays.toString(tags), value);
}
}

private static void assertMetric(String name, double expected, Map<String, Double> metrics) {
Double actual = metrics.get(name);
assertNotNull("missing " + name, actual);
assertEquals(name, expected, actual, 0);
}

@Test
public void encodeTo() {
Telemetry t = new Telemetry();
Expand All @@ -182,49 +204,28 @@ public void encodeTo() {
t.onResponse(503, 7, false);
t.onDrop(1, 25);

ByteArrayOutputStream out = new ByteArrayOutputStream();
PayloadBuilder pb = new PayloadBuilder(out::writeBytes);
t.snapshot(null).encodeTo(pb);
pb.close();
byte[] p = out.toByteArray();
Recorder rec = new Recorder();
t.snapshot(null).encodeTo(rec);

String prefix = "datadog.dogstatsd_http.client";
for (String suffix :
new String[] {
".enqueued_payloads",
".enqueued_bytes",
".delivered_payloads",
".delivered_bytes",
".dropped_payloads",
".dropped_bytes",
".queue_payloads",
".queue_bytes",
".queue_max_bytes",
".oldest_enqueued_age_seconds",
".last_success_age_seconds",
".response_payloads",
".response_bytes",
}) {
assertTrue("missing " + suffix, contains(p, prefix + suffix));
}
assertMetric(prefix + ".enqueued_payloads", 1, rec.counts);
assertMetric(prefix + ".enqueued_bytes", 10, rec.counts);
assertMetric(prefix + ".delivered_payloads", 1, rec.counts);
assertMetric(prefix + ".delivered_bytes", 5, rec.counts);
assertMetric(prefix + ".dropped_payloads", 1, rec.counts);
assertMetric(prefix + ".dropped_bytes", 25, rec.counts);

assertMetric(prefix + ".queue_payloads", 0, rec.gauges);
assertMetric(prefix + ".queue_bytes", 0, rec.gauges);
assertMetric(prefix + ".queue_max_bytes", 0, rec.gauges);
// Ages come off the real clock here; only their presence is deterministic.
assertNotNull(rec.gauges.get(prefix + ".oldest_enqueued_age_seconds"));
assertNotNull(rec.gauges.get(prefix + ".last_success_age_seconds"));

// Per-code totals are tagged with the HTTP code.
assertTrue(contains(p, "code:200"));
assertTrue(contains(p, "code:503"));
}

/** True if {@code needle} appears verbatim (as UTF-8) anywhere in {@code haystack}. */
private static boolean contains(byte[] haystack, String needle) {
byte[] n = needle.getBytes(StandardCharsets.UTF_8);
outer:
for (int i = 0; i + n.length <= haystack.length; i++) {
for (int j = 0; j < n.length; j++) {
if (haystack[i + j] != n[j]) {
continue outer;
}
}
return true;
}
return false;
assertMetric(prefix + ".response_payloads[code:200]", 1, rec.counts);
assertMetric(prefix + ".response_bytes[code:200]", 5, rec.counts);
assertMetric(prefix + ".response_payloads[code:503]", 1, rec.counts);
assertMetric(prefix + ".response_bytes[code:503]", 7, rec.counts);
}
}
Loading