Coalesce repeated events so no program can flood trace (0.7.0)

README.md

@@ -138,8 +138,46 @@ generates, reads or writes an ID.
 | **Returns immediately** | The HTTP call runs on one daemon thread the client owns. A Spigot plugin can report from the server thread and no tick waits on the network. |
 | **Never throws** | A server that is down, slow, or rejecting the key is a dropped report, not an exception in your program. Drops are logged at `FINE` if you gave a logger, otherwise not at all. |
 | **Bounded** | At most 256 reports wait to be sent; past that, new ones are dropped without a log line, even at `FINE`. A trace server that is unreachable for a week costs a few kilobytes, not your heap. |
+| **Throttled** | Repeats of one event within a minute are counted, not sent: the first goes out at once, the rest as one event when the minute is up. See [below](#repeated-events-are-coalesced). |
 | **`close()` drains** | Reports already queued get up to the client timeout (5 s total) to be sent before the thread stops, so a CLI that reports and exits at once does not lose its event. Still bounded: an unreachable server delays exit by at most the timeout. |
 
+## Repeated events are coalesced
+
+Since 0.7.0, a program cannot flood the trace server, however often it
+reports. Before, one server whose `/power` command was being run every
+100 ms — another plugin dispatching it — sent hundreds of thousands of
+`command` events in a few days, one per run.
+
+- An event is its name, value and tags (once `version`, `install` and any
+  server-wide tags are added; tag order does not matter). The first report of
+  an event is sent at once and opens a **60-second window**.
+- Identical reports inside the window are only counted. When the window
+  closes, a count above zero is sent as **one more event**: the same name
+  and tags, plus the tag `coalesced` with the count — and the count as its
+  `value` when the event had no value. An event that has a value keeps it,
+  since it stands for that many more of the same value.
+- A window that sent a count opens the next one, so an event repeated
+  without pause costs one report a minute. A window that counted nothing
+  just ends.
+- `close()` sends every count still open, inside the same 5-second bound.
+- `startup` is never coalesced or held back.
+- At most 256 distinct events are counted at once. A new event while 256 are
+  being counted is dropped without a log line, like a report that finds the
+  queue full, so memory stays bounded even for a program that reports a new
+  tag every call.
+
+So the number of times something happened is the sum, over stored events,
+of `value` for events tagged `coalesced` and 1 for the rest — for a valueless
+event like `command`, simply `SUM(COALESCE(value, 1))`. Counting rows still
+works for "was it used, by how many servers"; it now undercounts "how many
+times", by design.
+
+A minute is long enough that a command dispatched every tick is one event a
+minute instead of 1,200, and short enough that a person trying a command
+sees it on the trace server at once (the first report is never delayed) and
+the count soon after. The server buckets by day, so nothing it shows is
+finer than that anyway.
+
 ## Opting out
 
 Reporting is **opt-out**, and the person running the program always has the
@@ -213,7 +251,7 @@ plugins already vendor bStats' `Metrics.java`.
 <dependency>
     <groupId>com.github.Stephenson-Software</groupId>
     <artifactId>trace-client-java</artifactId>
-    <version>0.6.1</version>
+    <version>0.7.0</version>
 </dependency>
 ```
 

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>software.stephenson</groupId>
     <artifactId>trace-client</artifactId>
-    <version>0.6.1</version>
+    <version>0.7.0</version>
     <packaging>jar</packaging>
 
     <name>trace-client</name>

src/main/java/software/stephenson/trace/TraceClient.java

@@ -1,5 +1,5 @@
 /*
- * trace-client 0.6.1 -- https://github.com/Stephenson-Software/trace-client-java
+ * trace-client 0.7.0 -- https://github.com/Stephenson-Software/trace-client-java
  *
  * One call to report that a program was used. Copy this file into a project as
  * is, or depend on the artifact; either way there is nothing else to add.
@@ -20,11 +20,14 @@
 import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.TreeMap;
 import java.util.UUID;
-import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -52,6 +55,13 @@
  *       without a log line, even at {@link Level#FINE}. A trace server
  *       that is unreachable for a week costs a few kilobytes, not the
  *       host's heap.</li>
+ *   <li><b>It is throttled.</b> Repeats of one event -- same name, value and
+ *       tags -- within {@value #COALESCE_WINDOW_MS} ms are counted, not sent:
+ *       the first goes out at once, the rest as one event when the window
+ *       closes, carrying how many there were (see {@link #report(String,
+ *       Double, Map)}). A program, or another plugin dispatching its
+ *       command in a loop, cannot flood the trace server. {@code startup}
+ *       is never held back.</li>
  * </ul>
  *
  * <p>Reporting is opt-out, and the person running the program always has the
@@ -125,11 +135,35 @@
 public final class TraceClient {
 
     /** This client's version, as sent in the User-Agent. */
-    public static final String VERSION = "0.6.1";
+    public static final String VERSION = "0.7.0";
 
     /** How many reports may wait to be sent before new ones are dropped. */
     public static final int QUEUE_CAPACITY = 256;
 
+    /**
+     * How long repeats of one event are counted rather than sent. A minute:
+     * long enough that a command dispatched every tick, or every 100 ms by
+     * another plugin, is one event a minute instead of hundreds; short
+     * enough that a person using a command a few times sees it on the trace
+     * server within a minute, and that close() seldom has much to flush.
+     * The server buckets by day, so nothing it shows is finer than this.
+     */
+    public static final long COALESCE_WINDOW_MS = 60_000;
+
+    /**
+     * How many distinct events may be counted at once. An event that is not
+     * already being counted while this many are is dropped, without a log
+     * line -- like a report that finds the queue full -- so a program
+     * reporting a new name or tag every call costs bounded memory too.
+     */
+    public static final int MAX_COALESCED_EVENTS = 256;
+
+    /** The tag a coalesced event carries how many repeats it stands for in. */
+    public static final String COALESCED_TAG = "coalesced";
+
+    /** The event that is never coalesced: a start is never held back. */
+    static final String STARTUP = "startup";
+
     private static final int CONNECT_TIMEOUT_MS = 5_000;
     private static final int READ_TIMEOUT_MS = 5_000;
 
@@ -206,7 +240,17 @@ public final class TraceClient {
     private final String disabledReason; // null when enabled
     private final Map<String, String> serverWideTags; // never null; read once, at build()
     private final String installId; // null when disabled or when there is none
-    private final ThreadPoolExecutor executor; // null when disabled
+    private final long coalesceWindowNanos;
+    private final ScheduledThreadPoolExecutor executor; // null when disabled
+
+    // The rest is guarded by lock: report() runs on the host's threads,
+    // flushes run on the sender thread.
+    private final Object lock = new Object();
+    // Events being counted, by coalesceKey(), oldest first.
+    private final Map<String, Window> windows = new LinkedHashMap<>();
+    private int queued; // sends waiting for the sender thread, not yet started
+    private ScheduledFuture<?> flushTimer; // the next flushDue(), or null
+    private boolean closed;
 
     private TraceClient(Builder builder) {
         this.endpoint = builder.baseUrl.replaceAll("/+$", "") + "/api/metrics";
@@ -222,17 +266,21 @@ private TraceClient(Builder builder) {
         // After the opt-outs, never before: a disabled client neither makes
         // up an ID nor writes one to disk.
         this.installId = disabledReason == null ? resolveInstallId(builder, serverWide) : null;
+        this.coalesceWindowNanos = TimeUnit.MILLISECONDS.toNanos(builder.coalesceWindowMs);
         if (disabledReason == null) {
-            this.executor = new ThreadPoolExecutor(
-                    1, 1, 30, TimeUnit.SECONDS,
-                    new ArrayBlockingQueue<Runnable>(QUEUE_CAPACITY),
+            // Scheduled, so the one thread that sends also closes windows:
+            // no second thread. Its queue is unbounded; enqueue() is the
+            // bound, and at most one flush is ever scheduled beside the sends.
+            this.executor = new ScheduledThreadPoolExecutor(1,
                     runnable -> {
                         Thread thread = new Thread(runnable, "trace-client/" + application);
                         thread.setDaemon(true);
                         return thread;
                     },
                     new ThreadPoolExecutor.DiscardPolicy());
+            this.executor.setKeepAliveTime(30, TimeUnit.SECONDS);
             this.executor.allowCoreThreadTimeOut(true);
+            this.executor.setRemoveOnCancelPolicy(true);
         } else {
             this.executor = null;
         }
@@ -710,19 +758,169 @@ public void report(String name) {
      * Reports that {@code name} happened, with an optional numeric value and
      * optional string tags. Returns immediately; see the class comment. A
      * {@code null} or blank name sends nothing.
+     *
+     * <p>Repeats are coalesced. The first report of an event -- its name,
+     * value and tags, once {@code version}, {@code install} and the
+     * server-wide tags are added -- is sent at once and opens a window of
+     * {@value #COALESCE_WINDOW_MS} ms. Identical reports inside the window are
+     * only counted. When it closes, a count above zero is sent as one more
+     * event: the same name and tags plus the tag {@value #COALESCED_TAG} with
+     * the count, and the count as its {@code value} when the event had none
+     * (an event with a value keeps it: it stands for that many more of the
+     * same value). A window that sent a count opens the next one, so an
+     * event repeated without pause costs one report a minute; one that
+     * counted nothing ends. {@link #close()} sends every count still open.
+     *
+     * <p>So the number of times an event happened is the sum, over the
+     * events the server stored, of {@code value} where the
+     * {@value #COALESCED_TAG} tag is present and 1 where it is not -- for
+     * a valueless event, simply {@code COALESCE(value, 1)}.
+     *
+     * <p>{@code startup} is never coalesced, nor held back: every one is
+     * sent at once.
      */
     public void report(String name, Double value, Map<String, String> tags) {
         if (executor == null || name == null || name.trim().isEmpty()) {
             return;
         }
-        final String body = json(application, name, value,
-                withServerWideTags(withInstall(withVersion(tags, version), installId), serverWideTags));
-        executor.execute(() -> send(body));
+        // json() leaves out a value that is NaN or infinite, so they are the
+        // same event as no value at all.
+        Double finite = value == null || value.isNaN() || value.isInfinite() ? null : value;
+        Map<String, String> merged =
+                withServerWideTags(withInstall(withVersion(tags, version), installId), serverWideTags);
+        String body = json(application, name, finite, merged);
+        synchronized (lock) {
+            if (closed) {
+                return;
+            }
+            if (!STARTUP.equals(name)) {
+                String key = json(application, name, finite, new TreeMap<>(merged)); // tag order is not identity
+                Window window = windows.get(key);
+                if (window != null) {
+                    window.count++;
+                    return;
+                }
+                if (windows.size() >= MAX_COALESCED_EVENTS) {
+                    return; // bounded, like a full queue: dropped without a log line
+                }
+                windows.put(key, new Window(name, finite, merged, System.nanoTime() + coalesceWindowNanos));
+                scheduleFlush();
+            }
+            enqueue(body);
+        }
+    }
+
+    /** One event being counted: what to send if it repeats, and until when. */
+    private static final class Window {
+        final String name;
+        final Double value; // null, or finite
+        final Map<String, String> tags;
+        long deadline; // System.nanoTime() at which the window closes
+        long count; // repeats since the window opened
+
+        Window(String name, Double value, Map<String, String> tags, long deadline) {
+            this.name = name;
+            this.value = value;
+            this.tags = tags;
+            this.deadline = deadline;
+        }
+    }
+
+    /**
+     * Queues {@code body} for the sender thread, unless
+     * {@value #QUEUE_CAPACITY} sends are already waiting. Caller holds
+     * {@code lock}.
+     */
+    private void enqueue(final String body) {
+        if (queued >= QUEUE_CAPACITY) {
+            return;
+        }
+        queued++;
+        executor.execute(() -> {
+            synchronized (lock) {
+                queued--;
+            }
+            send(body);
+        });
+    }
+
+    /** Schedules flushDue() for the earliest window, if none is scheduled. Caller holds {@code lock}. */
+    private void scheduleFlush() {
+        if (flushTimer != null || windows.isEmpty()) {
+            return;
+        }
+        // Windows are opened and reopened in deadline order, but a scan is
+        // cheap and does not depend on it.
+        long now = System.nanoTime();
+        long wait = Long.MAX_VALUE;
+        for (Window window : windows.values()) {
+            wait = Math.min(wait, window.deadline - now);
+        }
+        flushTimer = executor.schedule(this::flushDue, Math.max(0, wait), TimeUnit.NANOSECONDS);
+    }
+
+    /** Runs on the sender thread when the earliest window closes. */
+    private void flushDue() {
+        synchronized (lock) {
+            flushTimer = null;
+            if (closed) {
+                return;
+            }
+            flush(false);
+            scheduleFlush();
+        }
+    }
+
+    /**
+     * Sends a count for every window that has closed -- every window at all
+     * when {@code everything} -- and ends or reopens it. Caller holds
+     * {@code lock}.
+     */
+    private void flush(boolean everything) {
+        long now = System.nanoTime();
+        for (Iterator<Window> it = windows.values().iterator(); it.hasNext(); ) {
+            Window window = it.next();
+            if (!everything && now - window.deadline < 0) {
+                continue;
+            }
+            if (window.count > 0) {
+                enqueue(json(application, window.name,
+                        window.value != null ? window.value : Double.valueOf(window.count),
+                        withCount(window.tags, window.count)));
+            }
+            if (everything || window.count == 0) {
+                it.remove();
+            } else {
+                window.count = 0;
+                window.deadline = now + coalesceWindowNanos;
+            }
+        }
     }
 
     /**
-     * Stops the sending thread, giving reports already queued up to
-     * {@value #READ_TIMEOUT_MS} ms in total to be sent first. A program that
+     * The tags plus {@value #COALESCED_TAG} with the count, unless they
+     * already carry one or are at {@link #MAX_TAGS}. A copy.
+     */
+    static Map<String, String> withCount(Map<String, String> tags, long count) {
+        Map<String, String> merged = new LinkedHashMap<>(tags);
+        if (!merged.containsKey(COALESCED_TAG) && merged.size() < MAX_TAGS) {
+            merged.put(COALESCED_TAG, String.valueOf(count));
+        }
+        return merged;
+    }
+
+    /** How many distinct events are being counted. For tests. */
+    int coalescing() {
+        synchronized (lock) {
+            return windows.size();
+        }
+    }
+
+    /**
+     * Sends every count still being coalesced (see {@link #report(String,
+     * Double, Map)}), then stops the sending thread, giving reports already
+     * queued up to {@value #READ_TIMEOUT_MS} ms in total to be sent first.
+     * A program that
      * reports and then exits within milliseconds -- a CLI -- would otherwise
      * lose its one event to the race between queueing it and the thread
      * picking it up. The bound still holds: an unreachable server delays exit
@@ -733,6 +931,16 @@ public void close() {
         if (executor == null) {
             return;
         }
+        synchronized (lock) {
+            if (!closed) {
+                closed = true;
+                flush(true);
+                if (flushTimer != null) {
+                    flushTimer.cancel(false); // or shutdown() would wait for it
+                    flushTimer = null;
+                }
+            }
+        }
         executor.shutdown(); // no new work; queued reports still run
         try {
             if (!executor.awaitTermination(READ_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
@@ -860,6 +1068,7 @@ public static final class Builder {
         private String installId;
         private File installIdFile;
         private Logger logger;
+        private long coalesceWindowMs = COALESCE_WINDOW_MS;
 
         private Builder(String baseUrl, String application, String version) {
             if (baseUrl == null || baseUrl.trim().isEmpty()) {
@@ -975,6 +1184,12 @@ public Builder logger(Logger logger) {
             return this;
         }
 
+        /** A shorter coalescing window, so tests need not wait a minute. Not public: the window is the client's. */
+        Builder coalesceWindowMillis(long millis) {
+            this.coalesceWindowMs = millis;
+            return this;
+        }
+
         /**
          * Builds the client. The environment variables
          * {@value TraceClient#ENV_USAGE_REPORTING} and