Skip to content

Use ConcurrentHashtable for telemetry log deduplication #12367

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
bric3 wants to merge 1 commit into from
Draft

Use ConcurrentHashtable for telemetry log deduplication #12367

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 20 additions & 6 deletions internal-api/src/jmh/java/datadog/trace/api/telemetry/LogCollectorBenchmark.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -11,34 +15,44 @@
@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.
// In this case, the stacktrace of the exception will not be available.
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);
}
}

Expand Down
131 changes: 98 additions & 33 deletions internal-api/src/main/java/datadog/trace/api/telemetry/LogCollector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand All @@ -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<RawLogMessage, AtomicInteger> rawLogMessages;
private final int maxCapacity;
private final ConcurrentHashtable.State<RawLogMessage> rawLogMessages;

public static LogCollector get() {
return INSTANCE;
Expand All @@ -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) {
Expand All @@ -54,52 +57,101 @@ 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<RawLogMessage> drain() {
if (rawLogMessages.isEmpty()) {
int size = estimateSize(rawLogMessages);
if (size == 0) {
return Collections.emptyList();
}

List<RawLogMessage> list = new ArrayList<>(rawLogMessages.size());
Iterator<Map.Entry<RawLogMessage, AtomicInteger>> iterator =
rawLogMessages.entrySet().iterator();

while (iterator.hasNext()) {
Map.Entry<RawLogMessage, AtomicInteger> 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<RawLogMessage> 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<RawLogMessage> DEDUP_COUNT =
AtomicIntegerFieldUpdater.newUpdater(RawLogMessage.class, "dedupCount");

public final String message;
public final String logLevel;
public final Throwable throwable;
public final String tags;
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;
Expand All @@ -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;
Expand Down Expand Up @@ -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());
}
}
}
66 changes: 0 additions & 66 deletions internal-api/src/test/groovy/datadog/trace/api/telemetry/LogCollectorTest.groovy
View file
Open in desktop

This file was deleted.

Loading