From 70069629c1ba81ac2e17dd29f6a0cdf24b60f8ee Mon Sep 17 00:00:00 2001 From: Brice Dutheil Date: Wed, 9 Sep 2026 18:17:42 +0200 Subject: [PATCH 1/6] chore: improve some ConcurrentHashtable methods javadoc and arg names --- .../trace/util/ConcurrentHashtable.java | 170 +++++++++++++----- 1 file changed, 124 insertions(+), 46 deletions(-) diff --git a/internal-api/src/main/java/datadog/trace/util/ConcurrentHashtable.java b/internal-api/src/main/java/datadog/trace/util/ConcurrentHashtable.java index 693ca2a1b16..cc0b53ee10e 100644 --- a/internal-api/src/main/java/datadog/trace/util/ConcurrentHashtable.java +++ b/internal-api/src/main/java/datadog/trace/util/ConcurrentHashtable.java @@ -310,23 +310,38 @@ public boolean removeIf(@Nonnull Predicate predicate) { } /** - * Removes all entries and passes each one to {@code sink} while holding the table write lock. - * The sink should be quick and must not throw. If it throws, the partial drain is not rolled - * back and the size is not adjusted. + * Removes all entries and invokes {@code drainedEntryConsumer} for each one. + * + *

The drain holds the table write lock while detaching buckets and invoking the consumer. + * For each removed entry, the consumer is invoked synchronously after its bucket is detached. + * Capacity is released only after all invocations return. The consumer should be quick and must + * not throw; failures are not rolled back. * *

Use {@link #drain(Object, BiConsumer)} to avoid a capturing lambda. + * + * @param drainedEntryConsumer action invoked for each removed entry */ - public void drain(@Nonnull Consumer sink) { - ConcurrentHashtable.drain(state, sink); + public void drain(@Nonnull Consumer drainedEntryConsumer) { + ConcurrentHashtable.drain(state, drainedEntryConsumer); } /** - * Context-passing {@link #drain(Consumer)}. Pass a non-capturing {@link BiConsumer} (typically - * a {@code static final}) plus the accumulator as {@code context} (e.g. the target list or - * event builder) to avoid a capturing-lambda allocation. + * Context-passing {@link #drain(Consumer)}. The drain holds the table write lock while + * detaching buckets and invoking {@code drainedEntryConsumer}. For each removed entry, the + * consumer is invoked synchronously after its bucket is detached. Capacity is released only + * after all invocations return. + * + *

Pass a non-capturing {@link BiConsumer} (typically a {@code static final}) plus the + * accumulator as {@code context} (e.g. the target list or event builder) to avoid a + * capturing-lambda allocation. + * + * @param context type + * @param context context passed to each invocation of {@code drainedEntryConsumer} + * @param drainedEntryConsumer action invoked with the context and each removed entry */ - public void drain(C context, @Nonnull BiConsumer sink) { - ConcurrentHashtable.drain(state, context, sink); + public void drain( + C context, @Nonnull BiConsumer drainedEntryConsumer) { + ConcurrentHashtable.drain(state, context, drainedEntryConsumer); } /** Removes all entries. Lock-free readers mid-walk complete against the entries they hold. */ @@ -582,23 +597,38 @@ public boolean removeIf(@Nonnull Predicate predicate) { } /** - * Removes all entries and passes each one to {@code sink} while holding the table write lock. - * The sink should be quick and must not throw. If it throws, the partial drain is not rolled - * back and the size is not adjusted. + * Removes all entries and invokes {@code drainedEntryConsumer} for each one. + * + *

The drain holds the table write lock while detaching buckets and invoking the consumer. + * For each removed entry, the consumer is invoked synchronously after its bucket is detached. + * Capacity is released only after all invocations return. The consumer should be quick and must + * not throw; failures are not rolled back. * *

Use {@link #drain(Object, BiConsumer)} to avoid a capturing lambda. + * + * @param drainedEntryConsumer action invoked for each removed entry */ - public void drain(@Nonnull Consumer sink) { - ConcurrentHashtable.drain(state, sink); + public void drain(@Nonnull Consumer drainedEntryConsumer) { + ConcurrentHashtable.drain(state, drainedEntryConsumer); } /** - * Context-passing {@link #drain(Consumer)}. Pass a non-capturing {@link BiConsumer} (typically - * a {@code static final}) plus the accumulator as {@code context} (e.g. the target list or - * event builder) to avoid a capturing-lambda allocation. + * Context-passing {@link #drain(Consumer)}. The drain holds the table write lock while + * detaching buckets and invoking {@code drainedEntryConsumer}. For each removed entry, the + * consumer is invoked synchronously after its bucket is detached. Capacity is released only + * after all invocations return. + * + *

Pass a non-capturing {@link BiConsumer} (typically a {@code static final}) plus the + * accumulator as {@code context} (e.g. the target list or event builder) to avoid a + * capturing-lambda allocation. + * + * @param context type + * @param context context passed to each invocation of {@code drainedEntryConsumer} + * @param drainedEntryConsumer action invoked with the context and each removed entry */ - public void drain(C context, @Nonnull BiConsumer sink) { - ConcurrentHashtable.drain(state, context, sink); + public void drain( + C context, @Nonnull BiConsumer drainedEntryConsumer) { + ConcurrentHashtable.drain(state, context, drainedEntryConsumer); } /** Removes all entries. Lock-free readers mid-walk complete against the entries they hold. */ @@ -1168,25 +1198,32 @@ public static boolean removeIf( } /** - * Removes all entries while holding the table write lock. Each bucket head is cleared with a - * volatile write before its detached chain is passed to {@code sink}, so subsequent lock-free - * readers observe an empty bucket while readers already on that chain can continue through its - * retained {@code next} links. This overload does not update size accounting. + * Removes all entries from {@code buckets} and invokes {@code drainedEntryConsumer} for each one. + * + *

The drain holds the table write lock while detaching buckets and invoking the consumer. For + * each removed entry, the consumer is invoked synchronously after its bucket is detached. A + * lock-free reader already traversing the detached chain can continue through its retained links. * - *

The sink must not throw. If it does, the partial drain is not rolled back. + *

This overload does not update size accounting. The consumer should be quick and must not + * throw; failures are not rolled back. + * + * @param entry type + * @param buckets bucket array to drain + * @param drainedEntryConsumer action invoked for each entry after its bucket is detached */ public static void drain( - @Nonnull AtomicReferenceArray buckets, @Nonnull Consumer sink) { - drainCounting(buckets, sink); + @Nonnull AtomicReferenceArray buckets, + @Nonnull Consumer drainedEntryConsumer) { + drainCounting(buckets, drainedEntryConsumer); } /** - * {@link #drain(AtomicReferenceArray, Consumer)} returning how many entries it handed to {@code - * sink}, so a {@link State} form can subtract exactly that from its {@link SizeManager} instead - * of zeroing. The count is free here: the sweep already visits every entry. + * {@link #drain(AtomicReferenceArray, Consumer)} returning the number of entries passed to {@code + * drainedEntryConsumer} for size accounting. */ private static int drainCounting( - @Nonnull AtomicReferenceArray buckets, @Nonnull Consumer sink) { + @Nonnull AtomicReferenceArray buckets, + @Nonnull Consumer drainedEntryConsumer) { int removed = 0; synchronized (getTableWriteLock(buckets)) { for (int i = 0; i < buckets.length(); i++) { @@ -1197,26 +1234,40 @@ private static int drainCounting( buckets.set(i, null); for (TEntry e = head; e != null; e = e.next()) { removed++; - sink.accept(e); + drainedEntryConsumer.accept(e); } } } return removed; } - /** Context-passing variant of {@link #drain(AtomicReferenceArray, Consumer)}. Self-locking. */ + /** + * Removes all entries from {@code buckets} and invokes {@code drainedEntryConsumer} with {@code + * context} and each removed entry. + * + *

The drain holds the table write lock while detaching buckets and invoking the consumer. For + * each removed entry, the consumer is invoked synchronously after its bucket is detached. This + * overload does not update size accounting. The consumer should be quick and must not throw; + * failures are not rolled back. + * + * @param context type + * @param entry type + * @param buckets bucket array to drain + * @param context context passed to each invocation of {@code drainedEntryConsumer} + * @param drainedEntryConsumer action invoked with the context and each removed entry + */ public static void drain( @Nonnull AtomicReferenceArray buckets, C context, - @Nonnull BiConsumer sink) { - drainCounting(buckets, context, sink); + @Nonnull BiConsumer drainedEntryConsumer) { + drainCounting(buckets, context, drainedEntryConsumer); } /** {@link #drainCounting(AtomicReferenceArray, Consumer)}, context-passing form. */ private static int drainCounting( @Nonnull AtomicReferenceArray buckets, C context, - @Nonnull BiConsumer sink) { + @Nonnull BiConsumer drainedEntryConsumer) { int removed = 0; synchronized (getTableWriteLock(buckets)) { for (int i = 0; i < buckets.length(); i++) { @@ -1227,7 +1278,7 @@ private static int drainCounting( buckets.set(i, null); for (TEntry e = head; e != null; e = e.next()) { removed++; - sink.accept(context, e); + drainedEntryConsumer.accept(context, e); } } } @@ -1235,25 +1286,52 @@ private static int drainCounting( } /** - * {@link #drain(AtomicReferenceArray, Consumer)} plus the matching bookkeeping: empties {@code - * state} into {@code sink} and gives its {@link SizeManager} back exactly the slots the sweep - * freed. Draining without that leaves the cap permanently consumed, so the two belong in one call - * rather than as a pair the caller has to remember. + * Removes all entries from {@code state}, invokes {@code drainedEntryConsumer} for each one, and + * releases one capacity slot for each removed entry. + * + *

The drain holds the table write lock while detaching buckets and invoking the consumer. For + * each removed entry, the consumer is invoked synchronously after its bucket is detached. + * Capacity for removed entries is released only after all invocations return. Outstanding + * reservations remain counted. + * + *

The consumer should be quick and must not throw; if it throws, removed entries are not + * restored and their capacity is not released. + * + * @param entry type + * @param state table state to drain + * @param drainedEntryConsumer action invoked for each entry after its bucket is detached */ public static void drain( - @Nonnull State state, @Nonnull Consumer sink) { + @Nonnull State state, @Nonnull Consumer drainedEntryConsumer) { synchronized (getTableWriteLock(state)) { - state.sizeManager.release(drainCounting(state.buckets, sink)); + state.sizeManager.release(drainCounting(state.buckets, drainedEntryConsumer)); } } - /** Context-passing form of {@link #drain(State, Consumer)}. */ + /** + * Removes all entries from {@code state}, invokes {@code drainedEntryConsumer} with {@code + * context} and each removed entry, and releases one capacity slot for each removed entry. + * + *

The drain holds the table write lock while detaching buckets and invoking the consumer. For + * each removed entry, the consumer is invoked synchronously after its bucket is detached. + * Capacity for removed entries is released only after all invocations return. Outstanding + * reservations remain counted. + * + *

The consumer should be quick and must not throw; if it throws, removed entries are not + * restored and their capacity is not released. + * + * @param context type + * @param entry type + * @param state table state to drain + * @param context context passed to each invocation of {@code drainedEntryConsumer} + * @param drainedEntryConsumer action invoked with the context and each removed entry + */ public static void drain( @Nonnull State state, C context, - @Nonnull BiConsumer sink) { + @Nonnull BiConsumer drainedEntryConsumer) { synchronized (getTableWriteLock(state)) { - state.sizeManager.release(drainCounting(state.buckets, context, sink)); + state.sizeManager.release(drainCounting(state.buckets, context, drainedEntryConsumer)); } } From f4c513cc0f26b6e7d9c39f79b0432138e7b05c81 Mon Sep 17 00:00:00 2001 From: Brice Dutheil Date: Tue, 1 Sep 2026 22:07:36 +0200 Subject: [PATCH 2/6] perf: reduce LogCollector allocations --- internal-api/build.gradle.kts | 1 + .../api/telemetry/LogCollectorBenchmark.java | 26 +- .../trace/api/telemetry/LogCollector.java | 198 ++++++++-- .../api/telemetry/LogCollectorTest.groovy | 66 ---- .../trace/api/telemetry/LogCollectorTest.java | 369 ++++++++++++++++++ 5 files changed, 549 insertions(+), 111 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/build.gradle.kts b/internal-api/build.gradle.kts index 42670eb6bcb..04214ae1ed5 100644 --- a/internal-api/build.gradle.kts +++ b/internal-api/build.gradle.kts @@ -281,6 +281,7 @@ dependencies { testImplementation("org.snakeyaml:snakeyaml-engine:2.9") testImplementation(project(":utils:test-utils")) testImplementation(libs.bundles.junit5) + testImplementation(libs.assertj.core) testImplementation("org.junit.vintage:junit-vintage-engine:${libs.versions.junit5.get()}") testImplementation(libs.commons.math) testImplementation(libs.bundles.mockito) 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..da82a3bd9ad 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,22 @@ 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.api.internal.VisibleForTesting; +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 +26,8 @@ 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; + @VisibleForTesting + final ConcurrentHashtable.State rawLogMessages; public static LogCollector get() { return INSTANCE; @@ -35,8 +41,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,52 +59,154 @@ 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) { + long keyHash = RawLogMessage.computeHash(logLevel, message, throwable); + int bucketIndex = bucketIndex(rawLogMessages.buckets, keyHash); + // Fast path for a full table: reject without locking when the target bucket is populated. + // If the bucket is empty, drain() may have detached it before releasing capacity, so continue + // to the locked capacity check. + if (isFull(rawLogMessages) && bucketAt(rawLogMessages, bucketIndex) != null) { // 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(); + + // Fast path for duplicates: search the target bucket without locking. + RawLogMessage rawLogMessage = find(bucketIndex, keyHash, logLevel, message, throwable); + if (rawLogMessage != null) { + rawLogMessage.increment(); + return; + } + + // Slow path after a miss: repeat the lookup and capacity checks under the table write lock + // because another writer or drain() may have changed the table. + synchronized (getTableWriteLock(rawLogMessages)) { + rawLogMessage = find(bucketIndex, keyHash, logLevel, message, throwable); + if (rawLogMessage != null) { + rawLogMessage.increment(); + return; + } + // Capacity may have been released by drain() or consumed by another writer while waiting. + if (isFull(rawLogMessages)) { + return; + } + + // Allocate before reserving because a reservation cannot + // be rolled back if construction fails. + rawLogMessage = + new RawLogMessage(logLevel, message, throwable, tags, System.currentTimeMillis() / 1000); + // Reserve before linking so every published entry is included in the capacity count. + if (rawLogMessages.sizeManager.tryReserve()) { + insertReserved(rawLogMessages, keyHash, rawLogMessage); + } + } } + /** + * Removes all available log group from this collector and returns them. + * + *

The count of each returned group is captured during removal. Increments that complete + * after the count is captured are not included. + * + * @return a collection containing the removed log groups + */ 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); - } - + // Note drain takes the table lock + List list = new ArrayList<>(size); + ConcurrentHashtable.drain( + rawLogMessages, + list, + (drained, logMessage) -> { + // Snapshot each log group's count before adding it to the drain result. + logMessage.snapshotCount(); + drained.add(logMessage); + }); return list; } - public static final class RawLogMessage { + /** + * Finds a log group with the same level, message, and throwable in the selected bucket. + * + *

Note, throwables are matched by identity or by class and stack trace. + * + *

The bucket chain supports lock-free reads. A caller that inserts after a miss must repeat + * the search under the table write lock. + * + * @param bucketIndex bucket selected for {@code keyHash} + * @param keyHash precomputed hash of the level, message, and throwable class + * @param logLevel log level to match + * @param message message to match + * @param throwable optional throwable to match + * @return the matching log group, or {@code null} if none is present + */ + @Nullable + private RawLogMessage find( + int bucketIndex, + long keyHash, + String logLevel, + String message, + @Nullable Throwable throwable) { + // Start searching from given bucket, and follow the entry next links + StackTraceElement[] stackTrace = null; + for (RawLogMessage entry = bucketAt(rawLogMessages, bucketIndex); + entry != null; + entry = entry.next()) { + if (entry.keyHash != keyHash + || !Objects.equals(logLevel, entry.logLevel) + || !Objects.equals(message, entry.message)) { + continue; + } + // throwables are more costly to compare, check first the identity + 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; + } + + /** + * Groups equivalent log messages for a telemetry flush. + * + *

Messages are equivalent when their log level, message, and throwable type and stack trace + * match. The first message supplies the tags and timestamp; later messages only increment the + * occurrence count. + */ + public static final class RawLogMessage extends ConcurrentHashtable.Entry { + private static final AtomicIntegerFieldUpdater LIVE_OCCURRENCE_COUNT_UPDATER = + AtomicIntegerFieldUpdater.newUpdater(RawLogMessage.class, "liveOccurrenceCount"); + public final String message; public final String logLevel; public final Throwable throwable; public final String tags; public final long timestamp; + /** + * Number of equivalent log messages captured when this group was drained. + */ public int count; - private StackTraceElement[] cachedStackTrace = null; + /** + * Live counter equivalent log messages accumulated in this group. + */ + private volatile int liveOccurrenceCount = 1; + private volatile 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; @@ -110,11 +217,8 @@ public RawLogMessage( public StackTraceElement[] stackTrace() { if (throwable == null) return null; - // DQH - getStackTrace makes a defensive copy, so getStackTrace can become a significant - // source of allocation - // In the worst case of a hot exception, we'll constantly call hashCode & equals to - // check against the key stored in the map, so avoiding repeated allocation on each - // comparison does provide a measurable gain + // getStackTrace() makes a defensive copy. Cache one safely published copy for concurrent + // comparisons against equivalent throwables. StackTraceElement[] stackTrace = cachedStackTrace; if (stackTrace != null) return stackTrace; @@ -122,6 +226,17 @@ public StackTraceElement[] stackTrace() { return stackTrace; } + private void increment() { + LIVE_OCCURRENCE_COUNT_UPDATER.incrementAndGet(this); + } + + /** + * Snapshot this log's live occurrence count + */ + private void snapshotCount() { + count = LIVE_OCCURRENCE_COUNT_UPDATER.get(this); + } + @Override public boolean equals(Object o) { if (this == o) return true; @@ -149,7 +264,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..b3683f4b900 --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/api/telemetry/LogCollectorTest.java @@ -0,0 +1,369 @@ +package datadog.trace.api.telemetry; + +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.assertj.core.api.Assertions.assertThat; + +import datadog.trace.test.util.PollingConditions; +import datadog.trace.util.ConcurrentHashtable; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.FutureTask; +import org.junit.jupiter.api.Test; + +class LogCollectorTest { + private static final long TIMEOUT_SECONDS = 10; + + @Test + void setsTracerTime() { + // Given + LogCollector logCollector = new LogCollector(1); + long before = System.currentTimeMillis() / 1000; + + // When + logCollector.addLogMessage("ERROR", "Message 1", null); + + long after = System.currentTimeMillis() / 1000; + LogCollector.RawLogMessage log = singleLog(logCollector.drain()); + + // Then + assertThat(log.timestamp).isBetween(before, after); + } + + @Test + void limitsLogMessages() { + // Given + LogCollector 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); + + Collection logs = logCollector.drain(); + + // Then + assertThat(logs).hasSize(3); + assertContainsLog(logs, "Message 1", 1); + assertContainsLog(logs, "Message 2", 1); + assertContainsLog(logs, "Message 3", 1); + assertThat(logs).extracting(log -> log.message).doesNotContain("Message 4"); + } + + @Test + void groupsMessages() { + // Given + LogCollector logCollector = new LogCollector(10); + + // When + logCollector.addLogMessage("ERROR", "Foo Message", null); + logCollector.addLogMessage("ERROR", "Bar Message", null); + logCollector.addLogMessage("ERROR", "Baz Message", null); + logCollector.addLogMessage("ERROR", "Qux Message", null); + logCollector.addLogMessage("ERROR", "Bar Message", null); + logCollector.addLogMessage("ERROR", "Baz Message", null); + logCollector.addLogMessage("ERROR", "Qux Message", null); + logCollector.addLogMessage("ERROR", "Baz Message", null); + logCollector.addLogMessage("ERROR", "Qux Message", null); + logCollector.addLogMessage("ERROR", "Qux Message", null); + + Collection logs = logCollector.drain(); + + // Then + assertThat(logs).hasSize(4); + assertContainsLog(logs, "Foo Message", 1); + assertContainsLog(logs, "Bar Message", 2); + assertContainsLog(logs, "Baz Message", 3); + assertContainsLog(logs, "Qux Message", 4); + } + + @Test + void dropsDuplicatesWhenFull() { + // Given + LogCollector logCollector = new LogCollector(1); + logCollector.addLogMessage("ERROR", "Message", null); + + // When + logCollector.addLogMessage("ERROR", "Message", null); + + // Then + assertThat(singleLog(logCollector.drain()).count).isEqualTo(1); + } + + @Test + void reusesCapacityAfterDrain() { + // Given + LogCollector logCollector = new LogCollector(1); + + // When + logCollector.addLogMessage("ERROR", "First", null); + + // Then + assertThat(singleLog(logCollector.drain()).message).isEqualTo("First"); + + // When + logCollector.addLogMessage("ERROR", "Second", null); + + // Then + assertThat(singleLog(logCollector.drain()).message).isEqualTo("Second"); + assertThat(logCollector.drain()).isEmpty(); + } + + @Test + void acceptsMessageAfterDrainDetachesFullBucket() throws Exception { + // Given + LogCollector logCollector = new LogCollector(1); + logCollector.addLogMessage("ERROR", "First", null); + ConcurrentHashtable.State state = logCollector.rawLogMessages; + CountDownLatch bucketDetached = new CountDownLatch(1); + CountDownLatch releaseDrain = new CountDownLatch(1); + FutureTask> drainTask = + new FutureTask<>( + () -> { + List drained = new ArrayList<>(); + ConcurrentHashtable.drain( + state, + drained, + (logs, log) -> { + bucketDetached.countDown(); + await(releaseDrain); + logs.add(log); + }); + return drained; + }); + Thread drainThread = new Thread(drainTask, "log-collector-drain"); + FutureTask writerTask = + new FutureTask<>( + () -> { + logCollector.addLogMessage("ERROR", "Second", null); + return null; + }); + Thread writerThread = new Thread(writerTask, "log-collector-writer"); + + // When + try { + drainThread.start(); + await(bucketDetached); + writerThread.start(); + new PollingConditions(TIMEOUT_SECONDS) + .eventually(() -> assertThat(writerThread.getState()).isEqualTo(Thread.State.BLOCKED)); + } finally { + releaseDrain.countDown(); + } + + // Then + Collection firstDrain = await(drainTask); + await(writerTask); + assertThat(singleLog(firstDrain).message).isEqualTo("First"); + assertThat(singleLog(logCollector.drain()).message).isEqualTo("Second"); + } + + @Test + void groupsEquivalentThrowablesAndKeepsFirstMetadata() { + // Given + LogCollector logCollector = new LogCollector(2); + Throwable first = throwableWithMethod("run", 10); + Throwable second = throwableWithMethod("run", 10); + + // When + logCollector.addLogMessage("ERROR", "Message", first, "source:first"); + logCollector.addLogMessage("ERROR", "Message", second, "source:second"); + + LogCollector.RawLogMessage log = singleLog(logCollector.drain()); + + // Then + assertThat(log.count).isEqualTo(2); + assertThat(log.throwable).isSameAs(first); + assertThat(log.tags).isEqualTo("source:first"); + } + + @Test + void keepsDifferentStackTracesSeparate() { + // Given + LogCollector logCollector = new LogCollector(2); + + // When + logCollector.addLogMessage("ERROR", "Message", throwableWithMethod("run", 10)); + logCollector.addLogMessage("ERROR", "Message", throwableWithMethod("run", 20)); + + // Then + assertThat(logCollector.drain()).hasSize(2); + } + + @Test + void rawLogMessageEqualityMatchesDeduplication() { + // Given + LogCollector.RawLogMessage first = + new LogCollector.RawLogMessage("ERROR", "Message", throwableWithMethod("run", 10), "first", 1); + LogCollector.RawLogMessage equivalent = + new LogCollector.RawLogMessage("ERROR", "Message", throwableWithMethod("run", 10), "second", 2); + LogCollector.RawLogMessage different = + new LogCollector.RawLogMessage("ERROR", "Message", throwableWithMethod("runIt", 20), "first", 1); + + // Then + assertThat(first).isEqualTo(equivalent); + assertThat(first).hasSameHashCodeAs(equivalent); + assertThat(first).isNotEqualTo(different); + } + + @Test + void countsConcurrentDuplicates() throws Exception { + // Given + 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( + () -> { + await(start); + for (int message = 0; message < messagesPerThread; message++) { + logCollector.addLogMessage("ERROR", "Message", null); + } + return null; + }); + } + + // When + release(start); + await(futures); + } finally { + shutdown(executor); + } + + // Then + assertThat(singleLog(logCollector.drain()).count).isEqualTo(threadCount * messagesPerThread); + } + + @Test + void groupsConcurrentEquivalentThrowables() throws Exception { + // Given + int threadCount = 16; + LogCollector logCollector = new LogCollector(2); + Throwable first = throwableWithMethod("run", 10); + logCollector.addLogMessage("ERROR", "Message", first); + 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( + () -> { + await(start); + logCollector.addLogMessage("ERROR", "Message", throwableWithMethod("run", 10)); + return null; + }); + } + + // When + release(start); + await(futures); + } finally { + shutdown(executor); + } + + LogCollector.RawLogMessage log = singleLog(logCollector.drain()); + + // Then + assertThat(log.count).isEqualTo(threadCount + 1); + assertThat(log.throwable).isSameAs(first); + } + + @Test + void capsConcurrentDistinctMessages() throws Exception { + // Given + 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( + () -> { + await(start); + logCollector.addLogMessage("ERROR", message, null); + return null; + }); + } + + // When + release(start); + await(futures); + } finally { + shutdown(executor); + } + + // Then + assertThat(logCollector.drain()).hasSize(capacity); + } + + private static Throwable throwableWithMethod(String methodName, int lineNumber) { + Throwable throwable = new IllegalStateException("ignored by deduplication"); + throwable.setStackTrace( + new StackTraceElement[] { + new StackTraceElement("Example", methodName, "Example.java", lineNumber) + }); + return throwable; + } + + private static void await(CountDownLatch latch) { + try { + assertThat(latch.await(TIMEOUT_SECONDS, SECONDS)).isTrue(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + + private static void release(CountDownLatch latch) { + latch.countDown(); + } + + private static T await(Future future) throws Exception { + return future.get(TIMEOUT_SECONDS, SECONDS); + } + + private static void await(Future[] futures) throws Exception { + for (Future future : futures) { + await(future); + } + } + + private static void shutdown(ExecutorService executor) throws InterruptedException { + executor.shutdownNow(); + assertThat(executor.awaitTermination(TIMEOUT_SECONDS, SECONDS)).isTrue(); + } + + private static LogCollector.RawLogMessage singleLog(Collection logs) { + assertThat(logs).hasSize(1); + return logs.iterator().next(); + } + + private static void assertContainsLog( + Collection logs, String message, int count) { + assertThat(logs) + .as("logs containing message %s", message) + .filteredOn(candidate -> message.equals(candidate.message)) + .singleElement() + .satisfies( + log -> { + assertThat(log.logLevel).isEqualTo("ERROR"); + assertThat(log.count).isEqualTo(count); + assertThat(log.throwable).isNull(); + }); + } +} From 5fecbb7f4721f5f9d16447fb934d07d264eb4fcb Mon Sep 17 00:00:00 2001 From: Brice Dutheil Date: Wed, 9 Sep 2026 17:32:16 +0200 Subject: [PATCH 3/6] fix: count duplicate telemetry logs at capacity --- .../trace/api/telemetry/LogCollector.java | 16 ++++++++-------- .../trace/api/telemetry/LogCollectorTest.java | 17 +++++++++++++++-- 2 files changed, 23 insertions(+), 10 deletions(-) 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 da82a3bd9ad..c693c019b8d 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 @@ -61,14 +61,6 @@ public void addLogMessage( String logLevel, String message, @Nullable Throwable throwable, @Nullable String tags) { long keyHash = RawLogMessage.computeHash(logLevel, message, throwable); int bucketIndex = bucketIndex(rawLogMessages.buckets, keyHash); - // Fast path for a full table: reject without locking when the target bucket is populated. - // If the bucket is empty, drain() may have detached it before releasing capacity, so continue - // to the locked capacity check. - if (isFull(rawLogMessages) && bucketAt(rawLogMessages, bucketIndex) != null) { - // TODO: We could emit a metric for dropped logs. - return; - } - // Fast path for duplicates: search the target bucket without locking. RawLogMessage rawLogMessage = find(bucketIndex, keyHash, logLevel, message, throwable); if (rawLogMessage != null) { @@ -76,6 +68,14 @@ public void addLogMessage( return; } + // Fast path after a miss for a full table: reject without locking when the target bucket is + // populated. If the bucket is empty, drain() may have detached it before releasing capacity, + // so continue to the locked capacity check. + if (isFull(rawLogMessages) && bucketAt(rawLogMessages, bucketIndex) != null) { + // TODO: We could emit a metric for dropped logs. + return; + } + // Slow path after a miss: repeat the lookup and capacity checks under the table write lock // because another writer or drain() may have changed the table. synchronized (getTableWriteLock(rawLogMessages)) { 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 index b3683f4b900..20886f7faa7 100644 --- a/internal-api/src/test/java/datadog/trace/api/telemetry/LogCollectorTest.java +++ b/internal-api/src/test/java/datadog/trace/api/telemetry/LogCollectorTest.java @@ -83,7 +83,7 @@ void groupsMessages() { } @Test - void dropsDuplicatesWhenFull() { + void countsEquivalentMessagesWhenFull() { // Given LogCollector logCollector = new LogCollector(1); logCollector.addLogMessage("ERROR", "Message", null); @@ -92,7 +92,20 @@ void dropsDuplicatesWhenFull() { logCollector.addLogMessage("ERROR", "Message", null); // Then - assertThat(singleLog(logCollector.drain()).count).isEqualTo(1); + assertThat(singleLog(logCollector.drain()).count).isEqualTo(2); + } + + @Test + void dropsNewLogGroupWhenFull() { + // Given + LogCollector logCollector = new LogCollector(1); + logCollector.addLogMessage("ERROR", "Existing message", null); + + // When + logCollector.addLogMessage("ERROR", "New message", null); + + // Then + assertThat(singleLog(logCollector.drain()).message).isEqualTo("Existing message"); } @Test From c870704ff39f94c86bc297000f5a42b8e4a793bb Mon Sep 17 00:00:00 2001 From: Brice Dutheil Date: Wed, 9 Sep 2026 17:50:29 +0200 Subject: [PATCH 4/6] chore: align vocabulary in tests --- .../trace/api/telemetry/LogCollectorTest.java | 134 +++++++++--------- 1 file changed, 68 insertions(+), 66 deletions(-) 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 index 20886f7faa7..7cd84cd6509 100644 --- a/internal-api/src/test/java/datadog/trace/api/telemetry/LogCollectorTest.java +++ b/internal-api/src/test/java/datadog/trace/api/telemetry/LogCollectorTest.java @@ -19,7 +19,7 @@ class LogCollectorTest { private static final long TIMEOUT_SECONDS = 10; @Test - void setsTracerTime() { + void recordsLogGroupTimestamp() { // Given LogCollector logCollector = new LogCollector(1); long before = System.currentTimeMillis() / 1000; @@ -28,14 +28,14 @@ void setsTracerTime() { logCollector.addLogMessage("ERROR", "Message 1", null); long after = System.currentTimeMillis() / 1000; - LogCollector.RawLogMessage log = singleLog(logCollector.drain()); + LogCollector.RawLogMessage logGroup = singleLogGroup(logCollector.drain()); // Then - assertThat(log.timestamp).isBetween(before, after); + assertThat(logGroup.timestamp).isBetween(before, after); } @Test - void limitsLogMessages() { + void limitsLogGroupsToCapacity() { // Given LogCollector logCollector = new LogCollector(3); @@ -45,18 +45,18 @@ void limitsLogMessages() { logCollector.addLogMessage("ERROR", "Message 3", null); logCollector.addLogMessage("ERROR", "Message 4", null); - Collection logs = logCollector.drain(); + Collection logGroups = logCollector.drain(); // Then - assertThat(logs).hasSize(3); - assertContainsLog(logs, "Message 1", 1); - assertContainsLog(logs, "Message 2", 1); - assertContainsLog(logs, "Message 3", 1); - assertThat(logs).extracting(log -> log.message).doesNotContain("Message 4"); + assertThat(logGroups).hasSize(3); + assertContainsLogGroup(logGroups, "Message 1", 1); + assertContainsLogGroup(logGroups, "Message 2", 1); + assertContainsLogGroup(logGroups, "Message 3", 1); + assertThat(logGroups).extracting(logGroup -> logGroup.message).doesNotContain("Message 4"); } @Test - void groupsMessages() { + void groupsEquivalentMessages() { // Given LogCollector logCollector = new LogCollector(10); @@ -72,18 +72,18 @@ void groupsMessages() { logCollector.addLogMessage("ERROR", "Qux Message", null); logCollector.addLogMessage("ERROR", "Qux Message", null); - Collection logs = logCollector.drain(); + Collection logGroups = logCollector.drain(); // Then - assertThat(logs).hasSize(4); - assertContainsLog(logs, "Foo Message", 1); - assertContainsLog(logs, "Bar Message", 2); - assertContainsLog(logs, "Baz Message", 3); - assertContainsLog(logs, "Qux Message", 4); + assertThat(logGroups).hasSize(4); + assertContainsLogGroup(logGroups, "Foo Message", 1); + assertContainsLogGroup(logGroups, "Bar Message", 2); + assertContainsLogGroup(logGroups, "Baz Message", 3); + assertContainsLogGroup(logGroups, "Qux Message", 4); } @Test - void countsEquivalentMessagesWhenFull() { + void countsEquivalentMessagesWhenTableIsFull() { // Given LogCollector logCollector = new LogCollector(1); logCollector.addLogMessage("ERROR", "Message", null); @@ -92,11 +92,11 @@ void countsEquivalentMessagesWhenFull() { logCollector.addLogMessage("ERROR", "Message", null); // Then - assertThat(singleLog(logCollector.drain()).count).isEqualTo(2); + assertThat(singleLogGroup(logCollector.drain()).count).isEqualTo(2); } @Test - void dropsNewLogGroupWhenFull() { + void dropsNewLogGroupWhenTableIsFull() { // Given LogCollector logCollector = new LogCollector(1); logCollector.addLogMessage("ERROR", "Existing message", null); @@ -105,11 +105,11 @@ void dropsNewLogGroupWhenFull() { logCollector.addLogMessage("ERROR", "New message", null); // Then - assertThat(singleLog(logCollector.drain()).message).isEqualTo("Existing message"); + assertThat(singleLogGroup(logCollector.drain()).message).isEqualTo("Existing message"); } @Test - void reusesCapacityAfterDrain() { + void reusesLogGroupCapacityAfterDrain() { // Given LogCollector logCollector = new LogCollector(1); @@ -117,18 +117,18 @@ void reusesCapacityAfterDrain() { logCollector.addLogMessage("ERROR", "First", null); // Then - assertThat(singleLog(logCollector.drain()).message).isEqualTo("First"); + assertThat(singleLogGroup(logCollector.drain()).message).isEqualTo("First"); // When logCollector.addLogMessage("ERROR", "Second", null); // Then - assertThat(singleLog(logCollector.drain()).message).isEqualTo("Second"); + assertThat(singleLogGroup(logCollector.drain()).message).isEqualTo("Second"); assertThat(logCollector.drain()).isEmpty(); } @Test - void acceptsMessageAfterDrainDetachesFullBucket() throws Exception { + void acceptsNewLogGroupAfterDrainDetachesFullBucket() throws Exception { // Given LogCollector logCollector = new LogCollector(1); logCollector.addLogMessage("ERROR", "First", null); @@ -138,16 +138,16 @@ void acceptsMessageAfterDrainDetachesFullBucket() throws Exception { FutureTask> drainTask = new FutureTask<>( () -> { - List drained = new ArrayList<>(); + List drainedLogGroups = new ArrayList<>(); ConcurrentHashtable.drain( state, - drained, - (logs, log) -> { + drainedLogGroups, + (logGroups, logGroup) -> { bucketDetached.countDown(); await(releaseDrain); - logs.add(log); + logGroups.add(logGroup); }); - return drained; + return drainedLogGroups; }); Thread drainThread = new Thread(drainTask, "log-collector-drain"); FutureTask writerTask = @@ -170,33 +170,33 @@ void acceptsMessageAfterDrainDetachesFullBucket() throws Exception { } // Then - Collection firstDrain = await(drainTask); + Collection firstDrainedLogGroups = await(drainTask); await(writerTask); - assertThat(singleLog(firstDrain).message).isEqualTo("First"); - assertThat(singleLog(logCollector.drain()).message).isEqualTo("Second"); + assertThat(singleLogGroup(firstDrainedLogGroups).message).isEqualTo("First"); + assertThat(singleLogGroup(logCollector.drain()).message).isEqualTo("Second"); } @Test - void groupsEquivalentThrowablesAndKeepsFirstMetadata() { + void groupsMessagesWithEquivalentThrowablesAndKeepsFirstMetadata() { // Given LogCollector logCollector = new LogCollector(2); - Throwable first = throwableWithMethod("run", 10); - Throwable second = throwableWithMethod("run", 10); + Throwable firstThrowable = throwableWithMethod("run", 10); + Throwable equivalentThrowable = throwableWithMethod("run", 10); // When - logCollector.addLogMessage("ERROR", "Message", first, "source:first"); - logCollector.addLogMessage("ERROR", "Message", second, "source:second"); + logCollector.addLogMessage("ERROR", "Message", firstThrowable, "source:first"); + logCollector.addLogMessage("ERROR", "Message", equivalentThrowable, "source:second"); - LogCollector.RawLogMessage log = singleLog(logCollector.drain()); + LogCollector.RawLogMessage logGroup = singleLogGroup(logCollector.drain()); // Then - assertThat(log.count).isEqualTo(2); - assertThat(log.throwable).isSameAs(first); - assertThat(log.tags).isEqualTo("source:first"); + assertThat(logGroup.count).isEqualTo(2); + assertThat(logGroup.throwable).isSameAs(firstThrowable); + assertThat(logGroup.tags).isEqualTo("source:first"); } @Test - void keepsDifferentStackTracesSeparate() { + void keepsMessagesWithDifferentStackTracesInSeparateGroups() { // Given LogCollector logCollector = new LogCollector(2); @@ -209,7 +209,7 @@ void keepsDifferentStackTracesSeparate() { } @Test - void rawLogMessageEqualityMatchesDeduplication() { + void rawLogMessageEqualityMatchesGrouping() { // Given LogCollector.RawLogMessage first = new LogCollector.RawLogMessage("ERROR", "Message", throwableWithMethod("run", 10), "first", 1); @@ -225,7 +225,7 @@ void rawLogMessageEqualityMatchesDeduplication() { } @Test - void countsConcurrentDuplicates() throws Exception { + void countsConcurrentEquivalentMessages() throws Exception { // Given int threadCount = 16; int messagesPerThread = 1_000; @@ -239,7 +239,7 @@ void countsConcurrentDuplicates() throws Exception { executor.submit( () -> { await(start); - for (int message = 0; message < messagesPerThread; message++) { + for (int occurrence = 0; occurrence < messagesPerThread; occurrence++) { logCollector.addLogMessage("ERROR", "Message", null); } return null; @@ -254,16 +254,17 @@ void countsConcurrentDuplicates() throws Exception { } // Then - assertThat(singleLog(logCollector.drain()).count).isEqualTo(threadCount * messagesPerThread); + assertThat(singleLogGroup(logCollector.drain()).count) + .isEqualTo(threadCount * messagesPerThread); } @Test - void groupsConcurrentEquivalentThrowables() throws Exception { + void groupsConcurrentMessagesWithEquivalentThrowables() throws Exception { // Given int threadCount = 16; LogCollector logCollector = new LogCollector(2); - Throwable first = throwableWithMethod("run", 10); - logCollector.addLogMessage("ERROR", "Message", first); + Throwable firstThrowable = throwableWithMethod("run", 10); + logCollector.addLogMessage("ERROR", "Message", firstThrowable); ExecutorService executor = Executors.newFixedThreadPool(threadCount); CountDownLatch start = new CountDownLatch(1); Future[] futures = new Future[threadCount]; @@ -285,15 +286,15 @@ void groupsConcurrentEquivalentThrowables() throws Exception { shutdown(executor); } - LogCollector.RawLogMessage log = singleLog(logCollector.drain()); + LogCollector.RawLogMessage logGroup = singleLogGroup(logCollector.drain()); // Then - assertThat(log.count).isEqualTo(threadCount + 1); - assertThat(log.throwable).isSameAs(first); + assertThat(logGroup.count).isEqualTo(threadCount + 1); + assertThat(logGroup.throwable).isSameAs(firstThrowable); } @Test - void capsConcurrentDistinctMessages() throws Exception { + void limitsConcurrentLogGroupsToCapacity() throws Exception { // Given int capacity = 3; int threadCount = 16; @@ -325,7 +326,7 @@ void capsConcurrentDistinctMessages() throws Exception { } private static Throwable throwableWithMethod(String methodName, int lineNumber) { - Throwable throwable = new IllegalStateException("ignored by deduplication"); + Throwable throwable = new IllegalStateException("ignored when grouping"); throwable.setStackTrace( new StackTraceElement[] { new StackTraceElement("Example", methodName, "Example.java", lineNumber) @@ -361,22 +362,23 @@ private static void shutdown(ExecutorService executor) throws InterruptedExcepti assertThat(executor.awaitTermination(TIMEOUT_SECONDS, SECONDS)).isTrue(); } - private static LogCollector.RawLogMessage singleLog(Collection logs) { - assertThat(logs).hasSize(1); - return logs.iterator().next(); + private static LogCollector.RawLogMessage singleLogGroup( + Collection logGroups) { + assertThat(logGroups).hasSize(1); + return logGroups.iterator().next(); } - private static void assertContainsLog( - Collection logs, String message, int count) { - assertThat(logs) - .as("logs containing message %s", message) + private static void assertContainsLogGroup( + Collection logGroups, String message, int count) { + assertThat(logGroups) + .as("log group for message %s", message) .filteredOn(candidate -> message.equals(candidate.message)) .singleElement() .satisfies( - log -> { - assertThat(log.logLevel).isEqualTo("ERROR"); - assertThat(log.count).isEqualTo(count); - assertThat(log.throwable).isNull(); + logGroup -> { + assertThat(logGroup.logLevel).isEqualTo("ERROR"); + assertThat(logGroup.count).isEqualTo(count); + assertThat(logGroup.throwable).isNull(); }); } } From a15663ee2134c7bb440ac7ac7fbd827c95c3f64c Mon Sep 17 00:00:00 2001 From: Brice Dutheil Date: Wed, 9 Sep 2026 18:14:49 +0200 Subject: [PATCH 5/6] chore: improve benchmark robustness --- .../api/telemetry/LogCollectorBenchmark.java | 5 +++-- .../trace/api/telemetry/LogCollector.java | 20 ++++++++----------- .../trace/api/telemetry/LogCollectorTest.java | 9 ++++++--- 3 files changed, 17 insertions(+), 17 deletions(-) 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 0899c8fecf6..81d2a9180b6 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 @@ -17,10 +17,11 @@ public class LogCollectorBenchmark { @State(Scope.Benchmark) public static class CollectorState { - final LogCollector collector = new LogCollector(4); + LogCollector collector; - @Setup(Level.Trial) + @Setup(Level.Iteration) public void setup() { + collector = new LogCollector(4); collector.addLogMessage("error", "ugh!", null); } } 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 c693c019b8d..fcf6d144cda 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 @@ -26,8 +26,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(); - @VisibleForTesting - final ConcurrentHashtable.State rawLogMessages; + @VisibleForTesting final ConcurrentHashtable.State rawLogMessages; public static LogCollector get() { return INSTANCE; @@ -128,7 +127,8 @@ public Collection drain() { } /** - * Finds a log group with the same level, message, and throwable in the selected bucket. + * Finds a log group with the same level, message, and + * throwable in the selected bucket. * *

Note, throwables are matched by identity or by class and stack trace. * @@ -193,15 +193,13 @@ public static final class RawLogMessage extends ConcurrentHashtable.Entry { public final Throwable throwable; public final String tags; public final long timestamp; - /** - * Number of equivalent log messages captured when this group was drained. - */ + + /** Number of equivalent log messages captured when this group was drained. */ public int count; - /** - * Live counter equivalent log messages accumulated in this group. - */ + /** Live counter equivalent log messages accumulated in this group. */ private volatile int liveOccurrenceCount = 1; + private volatile StackTraceElement[] cachedStackTrace = null; public RawLogMessage( @@ -230,9 +228,7 @@ private void increment() { LIVE_OCCURRENCE_COUNT_UPDATER.incrementAndGet(this); } - /** - * Snapshot this log's live occurrence count - */ + /** Snapshot this log's live occurrence count */ private void snapshotCount() { count = LIVE_OCCURRENCE_COUNT_UPDATER.get(this); } 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 index 7cd84cd6509..23b65cc523b 100644 --- a/internal-api/src/test/java/datadog/trace/api/telemetry/LogCollectorTest.java +++ b/internal-api/src/test/java/datadog/trace/api/telemetry/LogCollectorTest.java @@ -212,11 +212,14 @@ void keepsMessagesWithDifferentStackTracesInSeparateGroups() { void rawLogMessageEqualityMatchesGrouping() { // Given LogCollector.RawLogMessage first = - new LogCollector.RawLogMessage("ERROR", "Message", throwableWithMethod("run", 10), "first", 1); + new LogCollector.RawLogMessage( + "ERROR", "Message", throwableWithMethod("run", 10), "first", 1); LogCollector.RawLogMessage equivalent = - new LogCollector.RawLogMessage("ERROR", "Message", throwableWithMethod("run", 10), "second", 2); + new LogCollector.RawLogMessage( + "ERROR", "Message", throwableWithMethod("run", 10), "second", 2); LogCollector.RawLogMessage different = - new LogCollector.RawLogMessage("ERROR", "Message", throwableWithMethod("runIt", 20), "first", 1); + new LogCollector.RawLogMessage( + "ERROR", "Message", throwableWithMethod("runIt", 20), "first", 1); // Then assertThat(first).isEqualTo(equivalent); From 4eb4107a759df9ccb9193d812cffb5b642d96c34 Mon Sep 17 00:00:00 2001 From: Brice Dutheil Date: Thu, 10 Sep 2026 16:47:52 +0200 Subject: [PATCH 6/6] fix: retry telemetry log lookup at capacity --- .../main/java/datadog/trace/api/telemetry/LogCollector.java | 6 ++++++ 1 file changed, 6 insertions(+) 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 fcf6d144cda..a6d23b6e73e 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 @@ -71,6 +71,12 @@ public void addLogMessage( // populated. If the bucket is empty, drain() may have detached it before releasing capacity, // so continue to the locked capacity check. if (isFull(rawLogMessages) && bucketAt(rawLogMessages, bucketIndex) != null) { + // Mitigate a race where another writer could claim the bucket before the previous find + rawLogMessage = find(bucketIndex, keyHash, logLevel, message, throwable); + if (rawLogMessage != null) { + rawLogMessage.increment(); + return; + } // TODO: We could emit a metric for dropped logs. return; }