dedup

Author: niesfisch

README.md

@@ -16,6 +16,7 @@ If you work in a legacy app and ask things like "Can we remove this?" or "Is thi
 - [Configure](#configure)
 - [Run an Application with JCT](#run-an-application-with-jct)
 - [Available Processors](#available-processors)
+- [Stack Volume Control (All vs New Stacks)](#stack-volume-control-all-vs-new-stacks)
 - [Message Format](#message-format)
 - [Logging](#logging)
 - [Hello World Walkthrough](#hello-world-walkthrough)
@@ -168,6 +169,40 @@ Quick chooser:
 - Pick `udp` if you optimize for throughput and can tolerate some loss
 - Pick `tcp` if you want better transport reliability for ELK pipelines
 
+## Stack Volume Control (All vs New Stacks)
+
+On busy systems, writing every single captured stack can create huge event volume.
+If your main question is only "Was this path hit at least once?", you can report only new stacks.
+
+Processor flags:
+
+- `processor.reportAllStacks`
+  - `true` (default): report every captured event
+  - `false`: report only first-seen stack hashes within the current dedup window
+- `processor.stackHashResetIntervalMillis`
+  - Periodically clears remembered stack hashes to cap memory usage in long-running JVMs
+  - Default: `300000` (5 minutes)
+
+Example: report only new stacks (good for high-traffic legacy systems)
+
+```yaml
+processor:
+  fullQualifiedClass: de.marcelsauer.profiler.processor.udp.AsyncUdpStackProcessor
+  udpHost: localhost
+  udpPort: 9999
+  reportAllStacks: false
+  stackHashResetIntervalMillis: 300000
+```
+
+Example: report all stacks (full event stream)
+
+```yaml
+processor:
+  fullQualifiedClass: de.marcelsauer.profiler.processor.file.AsyncFileWritingStackProcessor
+  stackFolderName: /tmp/stacks/
+  reportAllStacks: true
+```
+
 ## Message Format
 
 ```json

doc/config-sample-file.yaml

@@ -10,3 +10,7 @@ classes:
 processor:
     fullQualifiedClass: de.marcelsauer.profiler.processor.file.AsyncFileWritingStackProcessor
     stackFolderName: /tmp/stacks/
+    # true = report every captured stack event
+    reportAllStacks: true
+    # used in dedup mode (reportAllStacks: false) to reset remembered hashes
+    stackHashResetIntervalMillis: 300000

doc/config-sample-helloworld-tcp.yaml

@@ -10,4 +10,8 @@ processor:
     tcpPort: 9999
     tcpConnectTimeoutMillis: 1000
     tcpReconnectDelayMillis: 1000
+    # true = report every captured stack event
+    reportAllStacks: true
+    # used in dedup mode (reportAllStacks: false)
+    stackHashResetIntervalMillis: 300000
 

doc/config-sample-helloworld-udp.yaml

@@ -8,3 +8,7 @@ processor:
     fullQualifiedClass: de.marcelsauer.profiler.processor.udp.AsyncUdpStackProcessor
     udpHost: localhost
     udpPort: 9999
+    # report only first-seen stacks in a time window
+    reportAllStacks: false
+    # reset dedup hashes every 5 minutes to cap memory usage
+    stackHashResetIntervalMillis: 300000

doc/config-sample-helloworld.yaml

@@ -7,3 +7,5 @@ classes:
 processor:
     fullQualifiedClass: de.marcelsauer.profiler.processor.file.AsyncFileWritingStackProcessor
     stackFolderName: /tmp/stacks/
+    reportAllStacks: true
+    stackHashResetIntervalMillis: 300000

src/main/java/de/marcelsauer/profiler/config/Processor.java

@@ -8,6 +8,17 @@
 public class Processor {
     public String fullQualifiedClass;
 
+    /**
+     * true: report every captured stack event
+     * false: report only first-seen stack hashes until periodic reset
+     */
+    public boolean reportAllStacks = true;
+
+    /**
+     * reset interval for remembered stack hashes in dedup mode to cap memory usage
+     */
+    public long stackHashResetIntervalMillis = 300_000L;
+
     public String udpHost = "localhost";
     public int udpPort = 9999;
 

src/main/java/de/marcelsauer/profiler/processor/AbstractAsyncStackProcessor.java

@@ -1,10 +1,13 @@
 package de.marcelsauer.profiler.processor;
 
+import de.marcelsauer.profiler.config.Config;
 import org.apache.log4j.Logger;
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -28,6 +31,9 @@ public abstract class AbstractAsyncStackProcessor implements StackProcessor {
     private final BlockingQueue<RecordingEvent> workerQueue = new ArrayBlockingQueue<>(CAPACITY);
     private final AtomicBoolean started = new AtomicBoolean(false);
     private final List<RecordingEvent> drainBuffer = new ArrayList<>(DRAIN_BATCH_SIZE);
+    private final List<RecordingEvent> dedupBuffer = new ArrayList<>(DRAIN_BATCH_SIZE);
+    private final Set<Integer> seenStackHashes = new HashSet<>();
+    private long nextStackHashResetAtMillis = 0L;
 
     /**
      * acting like cron
@@ -56,6 +62,7 @@ public void process(RecordingEvent event) {
     @Override
     public void start() {
         logger.info("starting " + this.getClass().getName());
+        resetDedupWindow();
         doStart();
         startScheduler();
     }
@@ -112,11 +119,17 @@ private void flushQueue() {
             if (snapshot.isEmpty()) {
                 return;
             }
-            doProcess(snapshot);
-            successfullyProcessedStacksCounter.addAndGet(snapshot.size());
+
+            Collection<RecordingEvent> reportableSnapshot = filterReportableStacks(snapshot);
+            if (reportableSnapshot.isEmpty()) {
+                continue;
+            }
+
+            doProcess(reportableSnapshot);
+            successfullyProcessedStacksCounter.addAndGet(reportableSnapshot.size());
             logger.info(
                 String.format("delegated %d stacks to %s. successfully processed so far %d",
-                    snapshot.size(),
+                    reportableSnapshot.size(),
                     this.getClass().getSimpleName(),
                     successfullyProcessedStacksCounter.intValue()));
         }
@@ -128,6 +141,49 @@ private Collection<RecordingEvent> drainNextBatch() {
         return drainBuffer;
     }
 
+    private Collection<RecordingEvent> filterReportableStacks(Collection<RecordingEvent> snapshot) {
+        if (isReportAllStacks()) {
+            return snapshot;
+        }
+
+        maybeResetSeenStackHashes();
+
+        dedupBuffer.clear();
+        for (RecordingEvent event : snapshot) {
+            if (seenStackHashes.add(event.getStackHash())) {
+                dedupBuffer.add(event);
+            }
+        }
+        return dedupBuffer;
+    }
+
+    private void maybeResetSeenStackHashes() {
+        long now = currentTimeMillis();
+        if (now < nextStackHashResetAtMillis) {
+            return;
+        }
+
+        seenStackHashes.clear();
+        nextStackHashResetAtMillis = now + getStackHashResetIntervalMillis();
+    }
+
+    private void resetDedupWindow() {
+        seenStackHashes.clear();
+        nextStackHashResetAtMillis = currentTimeMillis() + getStackHashResetIntervalMillis();
+    }
+
+    protected boolean isReportAllStacks() {
+        return Config.get().processor.reportAllStacks;
+    }
+
+    protected long getStackHashResetIntervalMillis() {
+        return Math.max(1_000L, Config.get().processor.stackHashResetIntervalMillis);
+    }
+
+    protected long currentTimeMillis() {
+        return System.currentTimeMillis();
+    }
+
     class WorkQueueProcessorTask implements Runnable {
 
         @Override

src/main/java/de/marcelsauer/profiler/processor/RecordingEvent.java

@@ -6,10 +6,12 @@
 public class RecordingEvent {
     public final Stack stack;
     public final long timestampMillis;
+    private final int stackHash;
 
     public RecordingEvent(Stack stack) {
         this.stack = stack;
         this.timestampMillis = System.currentTimeMillis();
+        this.stackHash = calculateStackHash();
     }
 
     List<Stack.StackEntry> getStackEntries() {
@@ -33,4 +35,16 @@ public String asJson() {
         sb.append("]}");
         return sb.toString();
     }
+
+    public int getStackHash() {
+        return stackHash;
+    }
+
+    private int calculateStackHash() {
+        int result = 1;
+        for (Stack.StackEntry entry : getStackEntries()) {
+            result = 31 * result + (entry.methodName != null ? entry.methodName.hashCode() : 0);
+        }
+        return result;
+    }
 }

src/test/java/de/marcelsauer/profiler/processor/AbstractAsyncStackProcessorTest.java

@@ -43,15 +43,59 @@ public void thatQueueIsDrainedInBatchesWithoutLosingEvents() {
         }
     }
 
+    @Test
+    public void thatOnlyNewStacksAreReportedWhenDedupModeIsEnabled() {
+        TestProcessor processor = new TestProcessor(false, 60_000L);
+        processor.start();
+
+        processor.process(newEvent("m1"));
+        processor.process(newEvent("m1"));
+        processor.process(newEvent("m2"));
+        processor.process(newEvent("m1"));
+
+        processor.stop();
+
+        assertEquals(2, processor.getProcessedCount());
+    }
+
+    @Test
+    public void thatSeenHashesAreResetAfterConfiguredInterval() throws InterruptedException {
+        TestProcessor processor = new TestProcessor(false, 1000L);
+        processor.start();
+
+        processor.process(newEvent("m1"));
+        Thread.sleep(1200L);
+
+        processor.advanceClockMillis(2000L);
+        processor.process(newEvent("m1"));
+        Thread.sleep(1200L);
+
+        processor.stop();
+
+        assertEquals(2, processor.getProcessedCount());
+    }
+
     private RecordingEvent newEvent(String methodName) {
         Stack.StackEntry entry = new Stack.StackEntry(methodName, 1L, 0);
         return new RecordingEvent(new Stack(Collections.singletonList(entry)));
     }
 
     private static class TestProcessor extends AbstractAsyncStackProcessor {
+        private final boolean reportAllStacks;
+        private final long stackHashResetIntervalMillis;
+        private long nowMillis;
         private final AtomicInteger processedCount = new AtomicInteger();
         private final List<Integer> batchSizes = Collections.synchronizedList(new ArrayList<Integer>());
 
+        TestProcessor() {
+            this(true, 300_000L);
+        }
+
+        TestProcessor(boolean reportAllStacks, long stackHashResetIntervalMillis) {
+            this.reportAllStacks = reportAllStacks;
+            this.stackHashResetIntervalMillis = stackHashResetIntervalMillis;
+        }
+
         @Override
         protected void doStart() {
             // no-op
@@ -68,6 +112,25 @@ protected void doProcess(Collection<RecordingEvent> snapshots) {
             processedCount.addAndGet(snapshots.size());
         }
 
+        @Override
+        protected boolean isReportAllStacks() {
+            return reportAllStacks;
+        }
+
+        @Override
+        protected long getStackHashResetIntervalMillis() {
+            return stackHashResetIntervalMillis;
+        }
+
+        @Override
+        protected long currentTimeMillis() {
+            return nowMillis;
+        }
+
+        void advanceClockMillis(long millis) {
+            nowMillis += millis;
+        }
+
         int getProcessedCount() {
             return processedCount.get();
         }