Add telemetry to the forwarder.

Author: vickenty

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

@@ -12,25 +12,29 @@
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.LongSupplier;
 
 class BoundedQueue {
-    // Key represents a tuple of integers (tries, clock).
+    // Key represents a tuple of integers (tries, clock). enqueuedAtNanos records when the
+    // payload first entered the queue (in System.nanoTime() units) and is preserved across
+    // requeues so that "age of oldest item" remains accurate after retries.
     static class Key implements Comparable<Key> {
         final long tries;
         final long clock;
+        final long enqueuedAtNanos;
 
         Key(long clock) {
-            this.tries = 0;
-            this.clock = clock;
+            this(0, clock, System.nanoTime());
         }
 
-        private Key(long tries, long clock) {
+        private Key(long tries, long clock, long enqueuedAtNanos) {
             this.tries = tries;
             this.clock = clock;
+            this.enqueuedAtNanos = enqueuedAtNanos;
         }
 
         Key next() {
-            return new Key(tries + 1, clock);
+            return new Key(tries + 1, clock, enqueuedAtNanos);
         }
 
         @Override
@@ -52,17 +56,40 @@ public int compareTo(Key o) {
 
     final TreeMap<Key, byte[]> items = new TreeMap<>();
 
-    long droppedItems;
-    long droppedBytes;
+    final Telemetry telemetry;
+    final LongSupplier nanos;
 
     Lock lock = new ReentrantLock();
     Condition notEmpty = lock.newCondition();
     Condition notFull = lock.newCondition();
 
     BoundedQueue(long maxBytes, long maxTries, WhenFull whenFull) {
+        this(maxBytes, maxTries, whenFull, null);
+    }
+
+    BoundedQueue(long maxBytes, long maxTries, WhenFull whenFull, Telemetry telemetry) {
+        this(maxBytes, maxTries, whenFull, telemetry, System::nanoTime);
+    }
+
+    BoundedQueue(
+            long maxBytes,
+            long maxTries,
+            WhenFull whenFull,
+            Telemetry telemetry,
+            LongSupplier nanos) {
         this.maxBytes = maxBytes;
         this.maxTries = maxTries;
         this.whenFull = whenFull;
+        this.telemetry = telemetry;
+        this.nanos = nanos;
+    }
+
+    long droppedPayloads;
+    long droppedBytes;
+
+    private void recordDrop(long bytes) {
+        droppedPayloads++;
+        droppedBytes += bytes;
     }
 
     void add(byte[] item) throws InterruptedException {
@@ -72,8 +99,7 @@ void add(byte[] item) throws InterruptedException {
     void requeue(Map.Entry<Key, byte[]> item) throws InterruptedException {
         Key nextKey = item.getKey().next();
         if (nextKey.tries > maxTries) {
-            droppedItems++;
-            droppedBytes += item.getValue().length;
+            telemetry.onDrop(1, item.getValue().length);
             return;
         }
         put(nextKey, item.getValue(), WhenFull.DROP);
@@ -82,7 +108,7 @@ void requeue(Map.Entry<Key, byte[]> item) throws InterruptedException {
     // Must be called when lock is held.
     private Key newKey() {
         clock++;
-        return new Key(clock);
+        return new Key(0, clock, nanos.getAsLong());
     }
 
     private void put(Key key, byte[] item, WhenFull whenFull) throws InterruptedException {
@@ -96,7 +122,13 @@ private void put(Key key, byte[] item, WhenFull whenFull) throws InterruptedExce
             bytes += item.length;
             notEmpty.signal();
         } finally {
+            long droppedPayloads = this.droppedPayloads;
+            long droppedBytes = this.droppedBytes;
+            this.droppedPayloads = 0;
+            this.droppedBytes = 0;
             lock.unlock();
+            // Avoid potential lock ordering issues.
+            telemetry.onDrop(droppedPayloads, droppedBytes);
         }
     }
 
@@ -108,9 +140,8 @@ private void ensureSpace(int length, WhenFull whenFull) throws InterruptedExcept
             switch (whenFull) {
                 case DROP:
                     Map.Entry<Key, byte[]> last = items.pollLastEntry();
-                    droppedItems++;
-                    droppedBytes += last.getValue().length;
                     bytes -= last.getValue().length;
+                    recordDrop(last.getValue().length);
                     break;
                 case BLOCK:
                     notFull.await();
@@ -133,4 +164,23 @@ Map.Entry<Key, byte[]> next() throws InterruptedException {
             lock.unlock();
         }
     }
+
+    void snapshot(long now, Telemetry.Snapshot s) {
+        lock.lock();
+        try {
+            long oldestAge = 0L;
+            for (Key k : items.keySet()) {
+                long age = now - k.enqueuedAtNanos;
+                if (age > oldestAge) {
+                    oldestAge = age;
+                }
+            }
+            s.queuePayloads = items.size();
+            s.queueBytes = bytes;
+            s.queueMaxBytes = maxBytes;
+            s.oldestEnqueuedAgeNanos = oldestAge;
+        } finally {
+            lock.unlock();
+        }
+    }
 }

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

@@ -39,7 +39,7 @@ public class Forwarder extends Thread {
     String localData;
     String externalData;
 
-    int responseOk, responseBadRequest, responseOther;
+    final Telemetry telemetry;
 
     /**
      * Creates a new forwarder targeting the given URL.
@@ -60,7 +60,8 @@ public Forwarder(
             Duration connectTimeout,
             Duration requestTimeout) {
         this.url = url;
-        this.queue = new BoundedQueue(maxRequestsBytes, maxTries, whenFull);
+        this.telemetry = new Telemetry();
+        this.queue = new BoundedQueue(maxRequestsBytes, maxTries, whenFull, this.telemetry);
         this.requestTimeout = requestTimeout;
         this.client =
                 HttpClient.newBuilder()
@@ -69,6 +70,14 @@ public Forwarder(
                         .build();
     }
 
+    /**
+     * Captures a snapshot of the forwarder's telemetry counters and queue state, clearing delta
+     * counters so subsequent snapshots report activity since this call.
+     */
+    public Telemetry.Snapshot snapshot() {
+        return telemetry.snapshot(queue);
+    }
+
     /** Runs the forwarding loop, delivering queued payloads until the thread is interrupted. */
     @Override
     public void run() {
@@ -92,6 +101,7 @@ public void run() {
      *     ({@link WhenFull#BLOCK} mode only)
      */
     public void send(byte[] payload) throws InterruptedException {
+        telemetry.onEnqueue(payload.length);
         queue.add(payload);
     }
 
@@ -119,23 +129,24 @@ void runOnce(Map.Entry<BoundedQueue.Key, byte[]> item) throws InterruptedExcepti
             logger.log(
                     Level.INFO, "response {0}: {1}", new Object[] {res.statusCode(), res.body()});
 
-            switch (res.statusCode()) {
+            int code = res.statusCode();
+            switch (code) {
                 case 400:
-                    responseBadRequest++;
+                    telemetry.onResponse(code, payload.length, false);
                     decreaseBackoff();
                     break;
                 case 200:
-                    responseOk++;
+                    telemetry.onResponse(code, payload.length, true);
                     decreaseBackoff();
                     break;
                 default:
-                    responseOther++;
+                    telemetry.onResponse(code, payload.length, false);
                     increaseBackoff();
                     queue.requeue(item);
             }
         } catch (IOException ex) {
             logger.log(Level.WARNING, "error sending request: {0}", ex.toString());
-            responseOther++;
+            telemetry.onTransportError(payload.length);
             increaseBackoff();
             queue.requeue(item);
         }

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

@@ -0,0 +1,47 @@
+/* Unless explicitly stated otherwise all files in this repository are
+ * licensed under the Apache 2.0 License.
+ *
+ * This product includes software developed at Datadog
+ *  (https://www.datadoghq.com/) Copyright 2026 Datadog, Inc.
+ */
+
+package com.datadoghq.dogstatsd.http.forwarder;
+
+/** Canonical telemetry labels for HTTP response codes. */
+final class HttpCode {
+
+    private static final String[] CATEGORIES = {"0xx", "1xx", "2xx", "3xx", "4xx", "5xx"};
+
+    private static final String[] NAMES = new String[506];
+
+    static {
+        for (int i = 0; i < NAMES.length; i++) {
+            NAMES[i] = CATEGORIES[i / 100];
+        }
+        NAMES[0] = "0";
+        NAMES[200] = "200";
+        NAMES[400] = "400";
+        NAMES[403] = "403";
+        NAMES[404] = "404";
+        NAMES[429] = "429";
+        NAMES[500] = "500";
+        NAMES[501] = "501";
+        NAMES[502] = "502";
+        NAMES[503] = "503";
+        NAMES[504] = "504";
+        NAMES[505] = "505";
+    }
+
+    private HttpCode() {}
+
+    static String name(int code) {
+        if (code >= 0 && code < NAMES.length) {
+            return NAMES[code];
+        }
+        int category = code / 100;
+        if (category >= 0 && category < CATEGORIES.length) {
+            return CATEGORIES[category];
+        }
+        return "xxx";
+    }
+}

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

@@ -0,0 +1,121 @@
+/* Unless explicitly stated otherwise all files in this repository are
+ * licensed under the Apache 2.0 License.
+ *
+ * This product includes software developed at Datadog
+ *  (https://www.datadoghq.com/) Copyright 2026 Datadog, Inc.
+ */
+
+package com.datadoghq.dogstatsd.http.forwarder;
+
+import java.time.Clock;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.LongSupplier;
+
+/** Thread-safe counter store for {@link Forwarder} telemetry. */
+public class Telemetry {
+
+    /** HTTP status code used to record transport-level (no-response) errors. */
+    public static final int TRANSPORT_ERROR_CODE = 0;
+
+    /** Point-in-time view of cumulative counters and queue state. */
+    public static final class Snapshot {
+        /**
+         * Wall-clock time (Unix epoch milliseconds) at the start of the interval covered by this
+         * snapshot — i.e., the moment of the previous snapshot, or telemetry construction if this
+         * is the first snapshot.
+         */
+        public long intervalStartMillis;
+
+        public long enqueuedPayloads;
+        public long deliveredPayloads;
+        public long enqueuedBytes;
+        public long deliveredBytes;
+        public long queuePayloads;
+        public long queueBytes;
+        public long queueMaxBytes;
+        public long droppedPayloads;
+        public long droppedBytes;
+
+        /** Nanos elapsed since the oldest queued item was enqueued; {@code 0} if queue is empty. */
+        public long oldestEnqueuedAgeNanos;
+
+        /** Nanos elapsed since the last successful submission; {@code 0} if none yet. */
+        public long lastSuccessAgeNanos;
+
+        /** Totals keyed by HTTP code. */
+        public Map<String, CodeCounters> byCode = new HashMap<>();
+
+        Snapshot(long intervalStartMillis) {
+            this.intervalStartMillis = intervalStartMillis;
+        }
+
+        /** Per-code totals within a snapshot's window. */
+        public static final class CodeCounters {
+            public long payloads;
+            public long bytes;
+        }
+    }
+
+    private final Clock clock;
+    private final LongSupplier nanos;
+
+    private Snapshot current;
+
+    private long lastSuccessNanos;
+    private boolean everDelivered;
+
+    public Telemetry() {
+        this(Clock.systemUTC(), System::nanoTime);
+    }
+
+    Telemetry(Clock clock, LongSupplier nanos) {
+        this.clock = clock;
+        this.nanos = nanos;
+        this.current = new Snapshot(clock.millis());
+    }
+
+    synchronized void onEnqueue(int len) {
+        current.enqueuedPayloads++;
+        current.enqueuedBytes += len;
+    }
+
+    synchronized void onResponse(int code, int len, boolean delivered) {
+        Snapshot.CodeCounters c =
+                current.byCode.computeIfAbsent(
+                        HttpCode.name(code), k -> new Snapshot.CodeCounters());
+        c.payloads++;
+        c.bytes += len;
+        if (delivered) {
+            current.deliveredPayloads++;
+            current.deliveredBytes += len;
+            lastSuccessNanos = nanos.getAsLong();
+            everDelivered = true;
+        }
+    }
+
+    synchronized void onTransportError(int len) {
+        onResponse(TRANSPORT_ERROR_CODE, len, false);
+    }
+
+    /** Records a dropped payload. */
+    synchronized void onDrop(long payloads, long bytes) {
+        current.droppedPayloads += payloads;
+        current.droppedBytes += bytes;
+    }
+
+    /**
+     * Captures a snapshot using the supplied queue stats, then swaps in a fresh accumulator so
+     * subsequent snapshots report deltas since this call.
+     */
+    public synchronized Snapshot snapshot(BoundedQueue q) {
+        long now = nanos.getAsLong();
+        Snapshot s = current;
+        current = new Snapshot(clock.millis());
+        s.lastSuccessAgeNanos = everDelivered ? now - lastSuccessNanos : 0L;
+        if (q != null) {
+            q.snapshot(now, s);
+        }
+        return s;
+    }
+}