Merge pull request #312 from DataDog/vickenty/ftl

AGTMETRICS-489 Add telemetry to dogstatsd-http-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,36 @@ 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) {
+    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 +95,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 +104,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 +118,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 +136,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 +160,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() {
@@ -93,6 +102,7 @@ public void run() {
      */
     public void send(byte[] payload) throws InterruptedException {
         queue.add(payload);
+        telemetry.onEnqueue(payload.length);
     }
 
     void runOnce(Map.Entry<BoundedQueue.Key, byte[]> item) throws InterruptedException {
@@ -119,37 +129,49 @@ 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()) {
-                case 400:
-                    responseBadRequest++;
-                    onSuccess();
-                    break;
-                case 200:
-                    responseOk++;
-                    onSuccess();
-                    break;
-                default:
-                    responseOther++;
-                    onError();
-                    queue.requeue(item);
-            }
+            handleResponse(res.statusCode(), item);
         } catch (IOException ex) {
             logger.log(Level.WARNING, "error sending request: {0}", ex.toString());
-            responseOther++;
-            onError();
-            queue.requeue(item);
+            handleTransportError(item);
         }
 
         backoff();
     }
 
+    void handleResponse(int code, Map.Entry<BoundedQueue.Key, byte[]> item)
+            throws InterruptedException {
+        int len = item.getValue().length;
+        switch (code) {
+            case 400:
+                telemetry.onResponse(code, len, false);
+                telemetry.onDrop(1, len);
+                decreaseBackoff();
+                break;
+            case 200:
+                telemetry.onResponse(code, len, true);
+                decreaseBackoff();
+                break;
+            default:
+                telemetry.onResponse(code, len, false);
+                increaseBackoff();
+                queue.requeue(item);
+        }
+    }
+
+    void handleTransportError(Map.Entry<BoundedQueue.Key, byte[]> item)
+            throws InterruptedException {
+        telemetry.onTransportError(item.getValue().length);
+        increaseBackoff();
+        queue.requeue(item);
+    }
+
     int delay;
 
-    void onSuccess() {
+    void decreaseBackoff() {
         delay >>= 4;
     }
 
-    void onError() {
+    void increaseBackoff() {
         if (delay < 64) delay <<= 1;
         if (delay == 0) delay = 1;
     }

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";
+    }
+}