Encode telemetry as dogstatsd-http metrics.

dogstatsd-http/forwarder/pom.xml

@@ -15,6 +15,11 @@
     <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>

dogstatsd-http/forwarder/src/main/java/com/datadoghq/dogstatsd/http/forwarder/Telemetry.java

@@ -7,8 +7,11 @@
 
 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;
 
@@ -46,10 +49,60 @@ public static final class Snapshot {
         /** Totals keyed by HTTP code. */
         public Map<String, CodeCounters> byCode = new HashMap<>();
 
+        /** Default metric name prefix used when none is supplied to {@link Snapshot#encode}. */
+        static final String DEFAULT_PREFIX = "datadog.dogstatsd_http.client";
+
         Snapshot(long intervalStartMillis) {
             this.intervalStartMillis = intervalStartMillis;
         }
 
+        /**
+         * Encodes this snapshot into {@code pb} using the default metric.
+         *
+         * @param pb Builder to append metrics to.
+         */
+        public void encodeTo(PayloadBuilder pb) {
+            encodeTo(DEFAULT_PREFIX, pb);
+        }
+
+        /**
+         * Encodes this snapshot into {@code pb}.
+         *
+         * @param pb Builder to append metrics to.
+         * @param prefix Metric name prefix.
+         */
+        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();
+
+            for (Map.Entry<String, CodeCounters> e : byCode.entrySet()) {
+                List<String> tags = Collections.singletonList("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();
+            }
+        }
+
         /** Per-code totals within a snapshot's window. */
         public static final class CodeCounters {
             public long payloads;

dogstatsd-http/forwarder/src/test/java/com/datadoghq/dogstatsd/http/forwarder/TelemetryTest.java

@@ -12,7 +12,10 @@
 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 org.junit.Test;
@@ -170,4 +173,58 @@ public void lastSuccessAge() {
         t.onResponse(200, 1, true);
         assertEquals(0L, t.snapshot(null).lastSuccessAgeNanos);
     }
+
+    @Test
+    public void encodeTo() {
+        Telemetry t = new Telemetry();
+        t.onEnqueue(10);
+        t.onResponse(200, 5, true);
+        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();
+
+        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));
+        }
+
+        // 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;
+    }
 }