From c1af8194a359c1e843a6f023e1e0679da7b739ff Mon Sep 17 00:00:00 2001 From: Brice Dutheil Date: Tue, 1 Sep 2026 22:07:36 +0200 Subject: [PATCH] perf: reduce LogCollector dedup allocations --- .../api/telemetry/LogCollectorBenchmark.java | 26 ++- .../trace/api/telemetry/LogCollector.java | 131 ++++++++--- .../api/telemetry/LogCollectorTest.groovy | 66 ------ .../trace/api/telemetry/LogCollectorTest.java | 211 ++++++++++++++++++ 4 files changed, 329 insertions(+), 105 deletions(-) delete mode 100644 internal-api/src/test/groovy/datadog/trace/api/telemetry/LogCollectorTest.groovy create mode 100644 internal-api/src/test/java/datadog/trace/api/telemetry/LogCollectorTest.java diff --git a/internal-api/src/jmh/java/datadog/trace/api/telemetry/LogCollectorBenchmark.java b/internal-api/src/jmh/java/datadog/trace/api/telemetry/LogCollectorBenchmark.java index d7c836b8621..0899c8fecf6 100644 --- a/internal-api/src/jmh/java/datadog/trace/api/telemetry/LogCollectorBenchmark.java +++ b/internal-api/src/jmh/java/datadog/trace/api/telemetry/LogCollectorBenchmark.java @@ -2,7 +2,11 @@ import org.openjdk.jmh.annotations.Benchmark; import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; import org.openjdk.jmh.annotations.Threads; import org.openjdk.jmh.annotations.Warmup; @@ -11,15 +15,25 @@ @Measurement(iterations = 5) @Threads(8) public class LogCollectorBenchmark { + @State(Scope.Benchmark) + public static class CollectorState { + final LogCollector collector = new LogCollector(4); + + @Setup(Level.Trial) + public void setup() { + collector.addLogMessage("error", "ugh!", null); + } + } + @Benchmark - public void noException_before() { - LogCollector.get().addLogMessage("error", "ugh!", null); + public void duplicateWithoutException(CollectorState state) { + state.collector.addLogMessage("error", "ugh!", null); } static final Object NULL = null; @Benchmark - public void nullPointerException() { + public void nullPointerException(CollectorState state) { // Represents the fast throw case where the JVM switches to using // a single Exception instance to handle a hot throw location // of NullPointerException, ArrayIndexOutOfBoundsException, etc. @@ -27,18 +41,18 @@ public void nullPointerException() { try { NULL.hashCode(); } catch (Throwable t) { - LogCollector.get().addLogMessage("error", "npe", t); + state.collector.addLogMessage("error", "npe", t); } } @Benchmark - public void unsupportedOperationException() { + public void unsupportedOperationException(CollectorState state) { // Represents the common case where stack trace is preserved // despite hot throw try { unsupportedOperation(); } catch (Throwable t) { - LogCollector.get().addLogMessage("error", "unsupported", t); + state.collector.addLogMessage("error", "unsupported", t); } } diff --git a/internal-api/src/main/java/datadog/trace/api/telemetry/LogCollector.java b/internal-api/src/main/java/datadog/trace/api/telemetry/LogCollector.java index b7ad3cb0eb0..39d600a421b 100644 --- a/internal-api/src/main/java/datadog/trace/api/telemetry/LogCollector.java +++ b/internal-api/src/main/java/datadog/trace/api/telemetry/LogCollector.java @@ -1,16 +1,21 @@ package datadog.trace.api.telemetry; -import datadog.trace.util.HashingUtils; +import static datadog.trace.util.ConcurrentHashtable.bucketAt; +import static datadog.trace.util.ConcurrentHashtable.bucketIndex; +import static datadog.trace.util.ConcurrentHashtable.estimateSize; +import static datadog.trace.util.ConcurrentHashtable.getTableWriteLock; +import static datadog.trace.util.ConcurrentHashtable.insertReserved; +import static datadog.trace.util.ConcurrentHashtable.isFull; +import static datadog.trace.util.LongHashingUtils.hash; + +import datadog.trace.util.ConcurrentHashtable; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; -import java.util.Iterator; import java.util.List; -import java.util.Map; import java.util.Objects; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import javax.annotation.Nullable; import org.slf4j.Marker; import org.slf4j.MarkerFactory; @@ -20,8 +25,7 @@ public class LogCollector { public static final Marker EXCLUDE_TELEMETRY = MarkerFactory.getMarker("EXCLUDE_TELEMETRY"); private static final int DEFAULT_MAX_CAPACITY = 10; private static final LogCollector INSTANCE = new LogCollector(); - private final Map rawLogMessages; - private final int maxCapacity; + private final ConcurrentHashtable.State rawLogMessages; public static LogCollector get() { return INSTANCE; @@ -35,8 +39,7 @@ private LogCollector() { value = "SING_SINGLETON_HAS_NONPRIVATE_CONSTRUCTOR", justification = "Usage in tests") LogCollector(int maxCapacity) { - this.maxCapacity = maxCapacity; - this.rawLogMessages = new ConcurrentHashMap<>(maxCapacity); + this.rawLogMessages = ConcurrentHashtable.State.createBounded(RawLogMessage.class, maxCapacity); } public void addLogMessage(String logLevel, String message, @Nullable Throwable throwable) { @@ -54,41 +57,88 @@ public void addLogMessage(String logLevel, String message, @Nullable Throwable t */ public void addLogMessage( String logLevel, String message, @Nullable Throwable throwable, @Nullable String tags) { - if (rawLogMessages.size() >= maxCapacity) { + if (isFull(rawLogMessages)) { // TODO: We could emit a metric for dropped logs. return; } - RawLogMessage rawLogMessage = - new RawLogMessage(logLevel, message, throwable, tags, System.currentTimeMillis() / 1000); - AtomicInteger count = rawLogMessages.computeIfAbsent(rawLogMessage, k -> new AtomicInteger()); - count.incrementAndGet(); + + long keyHash = RawLogMessage.computeHash(logLevel, message, throwable); + int index = bucketIndex(rawLogMessages.buckets, keyHash); + RawLogMessage rawLogMessage = find(index, keyHash, logLevel, message, throwable); + if (rawLogMessage != null) { + rawLogMessage.increment(); + return; + } + + synchronized (getTableWriteLock(rawLogMessages)) { + rawLogMessage = find(index, keyHash, logLevel, message, throwable); + if (rawLogMessage != null) { + rawLogMessage.increment(); + return; + } + if (isFull(rawLogMessages)) { + return; + } + + rawLogMessage = + new RawLogMessage(logLevel, message, throwable, tags, System.currentTimeMillis() / 1000); + if (rawLogMessages.sizeManager.tryReserve()) { + insertReserved(rawLogMessages, keyHash, rawLogMessage); + } + } } public Collection drain() { - if (rawLogMessages.isEmpty()) { + int size = estimateSize(rawLogMessages); + if (size == 0) { return Collections.emptyList(); } - List list = new ArrayList<>(rawLogMessages.size()); - Iterator> iterator = - rawLogMessages.entrySet().iterator(); - - while (iterator.hasNext()) { - Map.Entry entry = iterator.next(); - RawLogMessage logMessage = entry.getKey(); - // XXX: There might be lost writers to the counters under concurrency if another thread - // increments it - // while we are reading it here. At the moment, we are not overdoing this to prevent some - // counter losses. - logMessage.count = entry.getValue().get(); - iterator.remove(); - list.add(logMessage); - } - + List list = new ArrayList<>(size); + ConcurrentHashtable.drain( + rawLogMessages, + list, + (drained, logMessage) -> { + // A writer that found this entry before drain detached it can still increment too late. + logMessage.snapshotCount(); + drained.add(logMessage); + }); return list; } - public static final class RawLogMessage { + @Nullable + private RawLogMessage find( + int index, long keyHash, String logLevel, String message, @Nullable Throwable throwable) { + StackTraceElement[] stackTrace = null; + for (RawLogMessage entry = bucketAt(rawLogMessages, index); + entry != null; + entry = entry.next()) { + if (entry.keyHash != keyHash + || !Objects.equals(logLevel, entry.logLevel) + || !Objects.equals(message, entry.message)) { + continue; + } + if (throwable == entry.throwable) { + return entry; + } + if (throwable != null + && entry.throwable != null + && throwable.getClass().equals(entry.throwable.getClass())) { + if (stackTrace == null) { + stackTrace = throwable.getStackTrace(); + } + if (Objects.deepEquals(stackTrace, entry.stackTrace())) { + return entry; + } + } + } + return null; + } + + public static final class RawLogMessage extends ConcurrentHashtable.Entry { + private static final AtomicIntegerFieldUpdater DEDUP_COUNT = + AtomicIntegerFieldUpdater.newUpdater(RawLogMessage.class, "dedupCount"); + public final String message; public final String logLevel; public final Throwable throwable; @@ -96,10 +146,12 @@ public static final class RawLogMessage { public final long timestamp; public int count; + private volatile int dedupCount = 1; private StackTraceElement[] cachedStackTrace = null; public RawLogMessage( String logLevel, String message, Throwable throwable, String tags, long timestamp) { + super(computeHash(logLevel, message, throwable)); this.logLevel = logLevel; this.message = message; this.throwable = throwable; @@ -122,6 +174,14 @@ public StackTraceElement[] stackTrace() { return stackTrace; } + private void increment() { + DEDUP_COUNT.incrementAndGet(this); + } + + private void snapshotCount() { + count = DEDUP_COUNT.get(this); + } + @Override public boolean equals(Object o) { if (this == o) return true; @@ -149,7 +209,12 @@ public boolean equals(Object o) { @Override public int hashCode() { - return HashingUtils.hash(logLevel, message, throwable == null ? null : throwable.getClass()); + return (int) keyHash; + } + + private static long computeHash( + String logLevel, String message, @Nullable Throwable throwable) { + return hash(logLevel, message, throwable == null ? null : throwable.getClass()); } } } diff --git a/internal-api/src/test/groovy/datadog/trace/api/telemetry/LogCollectorTest.groovy b/internal-api/src/test/groovy/datadog/trace/api/telemetry/LogCollectorTest.groovy deleted file mode 100644 index 4f798f5bf9f..00000000000 --- a/internal-api/src/test/groovy/datadog/trace/api/telemetry/LogCollectorTest.groovy +++ /dev/null @@ -1,66 +0,0 @@ -package datadog.trace.api.telemetry - -import datadog.trace.test.util.DDSpecification - -class LogCollectorTest extends DDSpecification { - - void "tracer time is set"() { - setup: - def logCollector = new LogCollector(1) - - when: - logCollector.addLogMessage("ERROR", "Message 1", null) - - then: - def log = logCollector.drain().toList().get(0) - def ts = log.timestamp - ts > 0L - // Check tracer time is not in millis - ts < 1706529524286L - } - - void "limit log messages in LogCollector"() { - setup: - def logCollector = new LogCollector(3) - - when: - logCollector.addLogMessage("ERROR", "Message 1", null) - logCollector.addLogMessage("ERROR", "Message 2", null) - logCollector.addLogMessage("ERROR", "Message 3", null) - logCollector.addLogMessage("ERROR", "Message 4", null) - - then: - logCollector.rawLogMessages.size() == 3 - } - - void "grouping messages in LogCollector"() { - when: - LogCollector.get().addLogMessage("ERROR", "First Message", null) - LogCollector.get().addLogMessage("ERROR", "Second Message", null) - LogCollector.get().addLogMessage("ERROR", "Third Message", null) - LogCollector.get().addLogMessage("ERROR", "Forth Message", null) - LogCollector.get().addLogMessage("ERROR", "Second Message", null) - LogCollector.get().addLogMessage("ERROR", "Third Message", null) - LogCollector.get().addLogMessage("ERROR", "Forth Message", null) - LogCollector.get().addLogMessage("ERROR", "Third Message", null) - LogCollector.get().addLogMessage("ERROR", "Forth Message", null) - LogCollector.get().addLogMessage("ERROR", "Forth Message", null) - - then: - def list = LogCollector.get().drain() - list.size() == 4 - listContains(list, 'ERROR', "First Message", null, 1) - listContains(list, 'ERROR', "Second Message", null, 2) - listContains(list, 'ERROR', "Third Message", null,3) - listContains(list, 'ERROR', "Forth Message", null, 4) - } - - boolean listContains(Collection list, String logLevel, String message, Throwable t, int count) { - for (final def logMsg in list) { - if (logMsg.logLevel == logLevel && logMsg.message == message && logMsg.throwable == t && logMsg.count == count) { - return true - } - } - return false - } -} diff --git a/internal-api/src/test/java/datadog/trace/api/telemetry/LogCollectorTest.java b/internal-api/src/test/java/datadog/trace/api/telemetry/LogCollectorTest.java new file mode 100644 index 00000000000..34e221b254a --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/api/telemetry/LogCollectorTest.java @@ -0,0 +1,211 @@ +package datadog.trace.api.telemetry; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Collection; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import org.junit.jupiter.api.Test; + +class LogCollectorTest { + + @Test + void setsTracerTime() { + LogCollector logCollector = new LogCollector(1); + long before = System.currentTimeMillis() / 1000; + + logCollector.addLogMessage("ERROR", "Message 1", null); + + long after = System.currentTimeMillis() / 1000; + LogCollector.RawLogMessage log = onlyLog(logCollector.drain()); + assertTrue(log.timestamp >= before); + assertTrue(log.timestamp <= after); + } + + @Test + void limitsLogMessages() { + LogCollector logCollector = new LogCollector(3); + + logCollector.addLogMessage("ERROR", "Message 1", null); + logCollector.addLogMessage("ERROR", "Message 2", null); + logCollector.addLogMessage("ERROR", "Message 3", null); + logCollector.addLogMessage("ERROR", "Message 4", null); + + assertEquals(3, logCollector.drain().size()); + } + + @Test + void groupsMessages() { + LogCollector logCollector = new LogCollector(10); + + logCollector.addLogMessage("ERROR", "First Message", null); + logCollector.addLogMessage("ERROR", "Second Message", null); + logCollector.addLogMessage("ERROR", "Third Message", null); + logCollector.addLogMessage("ERROR", "Fourth Message", null); + logCollector.addLogMessage("ERROR", "Second Message", null); + logCollector.addLogMessage("ERROR", "Third Message", null); + logCollector.addLogMessage("ERROR", "Fourth Message", null); + logCollector.addLogMessage("ERROR", "Third Message", null); + logCollector.addLogMessage("ERROR", "Fourth Message", null); + logCollector.addLogMessage("ERROR", "Fourth Message", null); + + Collection logs = logCollector.drain(); + assertEquals(4, logs.size()); + assertLog(logs, "First Message", 1); + assertLog(logs, "Second Message", 2); + assertLog(logs, "Third Message", 3); + assertLog(logs, "Fourth Message", 4); + } + + @Test + void dropsDuplicatesWhenFull() { + LogCollector logCollector = new LogCollector(1); + + logCollector.addLogMessage("ERROR", "Message", null); + logCollector.addLogMessage("ERROR", "Message", null); + + assertEquals(1, onlyLog(logCollector.drain()).count); + } + + @Test + void reusesCapacityAfterDrain() { + LogCollector logCollector = new LogCollector(1); + + logCollector.addLogMessage("ERROR", "First", null); + assertEquals("First", onlyLog(logCollector.drain()).message); + logCollector.addLogMessage("ERROR", "Second", null); + + assertEquals("Second", onlyLog(logCollector.drain()).message); + assertTrue(logCollector.drain().isEmpty()); + } + + @Test + void groupsEquivalentThrowablesAndKeepsFirstMetadata() { + LogCollector logCollector = new LogCollector(2); + Throwable first = throwableAtLine(10); + Throwable second = throwableAtLine(10); + + logCollector.addLogMessage("ERROR", "Message", first, "source:first"); + logCollector.addLogMessage("ERROR", "Message", second, "source:second"); + + LogCollector.RawLogMessage log = onlyLog(logCollector.drain()); + assertEquals(2, log.count); + assertSame(first, log.throwable); + assertEquals("source:first", log.tags); + } + + @Test + void keepsDifferentStackTracesSeparate() { + LogCollector logCollector = new LogCollector(2); + + logCollector.addLogMessage("ERROR", "Message", throwableAtLine(10)); + logCollector.addLogMessage("ERROR", "Message", throwableAtLine(20)); + + assertEquals(2, logCollector.drain().size()); + } + + @Test + void rawLogMessageEqualityMatchesDeduplication() { + LogCollector.RawLogMessage first = + new LogCollector.RawLogMessage("ERROR", "Message", throwableAtLine(10), "first", 1); + LogCollector.RawLogMessage equivalent = + new LogCollector.RawLogMessage("ERROR", "Message", throwableAtLine(10), "second", 2); + LogCollector.RawLogMessage different = + new LogCollector.RawLogMessage("ERROR", "Message", throwableAtLine(20), "first", 1); + + assertEquals(first, equivalent); + assertEquals(first.hashCode(), equivalent.hashCode()); + assertNotEquals(first, different); + } + + @Test + void countsConcurrentDuplicates() throws Exception { + int threadCount = 16; + int messagesPerThread = 1_000; + LogCollector logCollector = new LogCollector(2); + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + CountDownLatch start = new CountDownLatch(1); + Future[] futures = new Future[threadCount]; + try { + for (int i = 0; i < threadCount; i++) { + futures[i] = + executor.submit( + () -> { + start.await(); + for (int message = 0; message < messagesPerThread; message++) { + logCollector.addLogMessage("ERROR", "Message", null); + } + return null; + }); + } + start.countDown(); + for (Future future : futures) { + future.get(); + } + } finally { + executor.shutdownNow(); + } + + assertEquals(threadCount * messagesPerThread, onlyLog(logCollector.drain()).count); + } + + @Test + void capsConcurrentDistinctMessages() throws Exception { + int capacity = 3; + int threadCount = 16; + LogCollector logCollector = new LogCollector(capacity); + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + CountDownLatch start = new CountDownLatch(1); + Future[] futures = new Future[threadCount]; + try { + for (int i = 0; i < threadCount; i++) { + String message = "Message " + i; + futures[i] = + executor.submit( + () -> { + start.await(); + logCollector.addLogMessage("ERROR", message, null); + return null; + }); + } + start.countDown(); + for (Future future : futures) { + future.get(); + } + } finally { + executor.shutdownNow(); + } + + assertEquals(capacity, logCollector.drain().size()); + } + + private static Throwable throwableAtLine(int lineNumber) { + Throwable throwable = new IllegalStateException("ignored by deduplication"); + throwable.setStackTrace( + new StackTraceElement[] { + new StackTraceElement("Example", "run", "Example.java", lineNumber) + }); + return throwable; + } + + private static LogCollector.RawLogMessage onlyLog(Collection logs) { + assertEquals(1, logs.size()); + return logs.iterator().next(); + } + + private static void assertLog( + Collection logs, String message, int count) { + LogCollector.RawLogMessage log = + logs.stream() + .filter(candidate -> message.equals(candidate.message)) + .findFirst() + .orElseThrow(() -> new AssertionError("Missing log message: " + message)); + assertEquals("ERROR", log.logLevel); + assertEquals(count, log.count); + } +}